# Lint as: python2, python3
# Copyright 2023 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import logging
import os
import re
import sys

from autotest_lib.client.bin import utils as client_utils
from autotest_lib.client.common_lib import error
from autotest_lib.server import utils
from autotest_lib.server.cros.bluetooth import bluetooth_adapter_tests
from autotest_lib.server.cros.tradefed import bundle_utils
from autotest_lib.server.cros.tradefed.tradefed_test import BundleSpecification
from autotest_lib.server.cros.tradefed.tradefed_test import TradefedTest

CTS_TARGET_MODULE = 'CtsBluetooth'
CTS_BUNDLE = 'arm'
CTS_URI = 'DEV'
CTS_TIMEOUT_SECONDS = 5400
CTS_CONFIG_RELATIVE_PATH = '../../cheets_CTS_{}/bundle_url_config.json'
ARC_FIELD_ID = 'CHROMEOS_ARC_ANDROID_SDK_VERSION'
ARC_VERSION_MAPPING = {'28': 'P', '30': 'R', '33': 'T'}
BT_ADDR_PATTERN = '([0-9A-F]{2}:){5}[0-9A-F]{2}'


class bluetooth_AdapterCTSHealth(TradefedTest,
                                 bluetooth_adapter_tests.BluetoothAdapterTests
                                 ):
    """Class to run Bluetooth CTS health tests"""

    def _tradefed_retry_command(self, template, session_id):
        """Needed by tradefed.

        Content is copy-pasted, except the forget device part.
        The reason is, we need to forget the device AFTER chrome is restarted,
        so that cannot be done earlier.
        """
        cmd = []
        for arg in template:
            cmd.append(arg.format(session_id=session_id))

        self.forget_bonded_devices()
        return cmd

    def _tradefed_run_command(self, template):
        """Needed by tradefed.

        Content is copy-pasted, except the forget device part.
        The reason is, we need to forget the device AFTER chrome is restarted,
        so that cannot be done earlier.
        """
        cmd = template[:]

        if self.arc_version == 'P':
            # Apply this PATH change only for chroot environment
            if not utils.is_in_container() and not client_utils.is_moblab():
                try:
                    os.environ['JAVA_HOME'] = '/opt/icedtea-bin-3.4.0'
                    os.environ['PATH'] = os.environ['JAVA_HOME']\
                                       + '/bin:' + os.environ['PATH']
                except OSError:
                    logging.error('Can\'t change current PATH directory')

            # Suppress redundant output from tradefed.
            cmd.append('--quiet-output=true')

        self.forget_bonded_devices()
        return cmd

    def _get_tradefed_base_dir(self):
        """Needed by tradefed. Content is copy-pasted."""
        return 'android-cts'

    def _tradefed_cmd_path(self):
        """Needed by tradefed. Content is copy-pasted."""
        return os.path.join(self._repository, 'tools', 'cts-tradefed')

    def _get_bundle_specification(self, uri, bundle):
        """Tradefed function to get the test bundle.

        This function is overrided to obtain the correct path for each arc
        versions.

        @param uri: determined by tradefed, could be 'DEV' or 'LATEST'
        @param bundle: always 'arm' in our case, but can also be 'x86'
        """
        test_path = sys.modules[self.__class__.__module__].__file__
        config_path = os.path.abspath(
                os.path.join(test_path,
                             CTS_CONFIG_RELATIVE_PATH.format(
                                     self.arc_version)))
        url_config = bundle_utils.load_config(config_path)
        bundle_password = bundle_utils.get_bundle_password(url_config)
        suite_name = bundle_utils.get_suite_name(url_config)

        return BundleSpecification(
                bundle_utils.make_bundle_url(url_config, uri, bundle),
                bundle_password, suite_name)

    def probe_arc_version(self):
        """Detects the installed ARC version."""
        arc_version = ''
        cmd_result = self.host.run('cat', args=('/etc/lsb-release', ))
        output_text = cmd_result.stdout
        for line in output_text.splitlines():
            if ARC_FIELD_ID not in line:
                continue
            arc_version_num = line.split('=')[1].strip()
            arc_version = ARC_VERSION_MAPPING.get(arc_version_num)
            if not arc_version:
                raise error.TestFail('Unknown ARC version: %s; expected: %s',
                                     arc_version_num,
                                     ARC_VERSION_MAPPING.keys())
        if arc_version == '':
            raise error.TestFail('ARC version not found')

        self.arc_version = arc_version
        logging.info('ARC version %s detected', arc_version)

    def initialize_arc_templates(self):
        """Assigns the ARC templates.

        The templates are required by the tradefed test. The content is copy
        pasted from CTS test files.
        """
        if self.arc_version == 'P':
            self.run_template = [
                    'run', 'commandAndExit', 'cts', '--module',
                    'CtsBluetoothTestCases', '--logcat-on-failure',
                    '--dynamic-config-url='
            ]
            self.retry_template = [
                    'run', 'commandAndExit', 'retry', '--retry',
                    '{session_id}', '--dynamic-config-url='
            ]
        elif self.arc_version == 'R' or self.arc_version == 'T':
            self.run_template = [
                    'run', 'commandAndExit', 'cts', '--include-filter',
                    'CtsBluetoothTestCases', '--include-filter',
                    'CtsBluetoothTestCases[secondary_user]',
                    '--logcat-on-failure'
            ]
            self.retry_template = [
                    'run', 'commandAndExit', 'retry', '--retry', '{session_id}'
            ]

    def initialize(self, host):
        """Tradefed function to initialize the test.

        This function is overrided to store the arc_version, in order to
        avoid creating one test file for each of the arc versions.

        @param host: The DUT, usually a chromebook
        """
        if not host:
            raise error.TestFail('The host is unspecified')

        self.host = host
        self.probe_arc_version()
        self.initialize_arc_templates()

        TradefedTest.initialize(self,
                                bundle=CTS_BUNDLE,
                                uri=CTS_URI,
                                host=host)
        bluetooth_adapter_tests.BluetoothAdapterTests.initialize(self)

    def prepare_btpeers(self, args_dict):
        """Prepares the btpeers for the CTS test.

        We need two btpeers: a classic peer to be discoverable and an LE peer
        to advertise. We need to be able to receive both inquiry and
        advertisement to pass all CTS tests, therefore we require the two peers.

        It's ideal to use the same framework as the other BT adapter tests, but
        that is difficult to achieve since Tradefed tests has it's own quirks.
        Directly using both of them makes things complicated (such as both are
        trying to restart UI). Therefore, here we only reuse some methods and
        copy some parts of Bluetooth adapter tests.

        @param args_dict: The arguments passed from the command line
        """

        btpeer_args = self.host.get_btpeer_arguments(args_dict)
        self.host.initialize_btpeer(btpeer_args=btpeer_args)
        for btpeer in self.host.btpeer_list:
            btpeer.register_raspPi_log(self.outputdir)

        self.btpeer_group = dict()
        self.btpeer_group_copy = dict()
        self.group_btpeers_type()

        devices = {'KEYBOARD': 1, 'BLE_MOUSE': 1}
        is_enough_peer = self.get_device_rasp(devices)
        if not is_enough_peer:
            raise error.TestNAError('Not enough peers!')

        classic_peer = self.devices['KEYBOARD'][0]
        classic_peer.SetDiscoverable(True)
        ble_peer = self.devices['BLE_MOUSE'][0]
        ble_peer.SetDiscoverable(True)

    def stop_btpeers(self):
        """Stops the btpeers from being discoverable.

        This is just a courtesy to stop the scanning that is started on the
        prepare function. The outcome of the test is not impacted by anything
        inside this function, and it's safe not to call this at all.
        """

        classic_peer = self.devices['KEYBOARD'][0]
        classic_peer.SetDiscoverable(False)
        ble_peer = self.devices['BLE_MOUSE'][0]
        ble_peer.SetDiscoverable(False)

    def forget_bonded_devices(self):
        """Removes the bonded devices.

        We need to remove the bonded devices in order to prevent them
        interfering with the test result (e.g. DUT auto connects to them).
        This is done in the beginning of the test, similar to other Bluetooth
        tests. However, instead of calling test_reset_on_adapter(), we directly
        call some commands. This is done because test_reset_on_adapter needs
        bluetooth_facade with conflicts with the tradefed test.
        """
        if self.floss:
            # in btclient, peer address is enclosed with [], host address isn't.
            regex = re.compile('\[' + BT_ADDR_PATTERN + '\]')
            self.host.run('btclient',
                          args=('--command', 'adapter enable', '--timeout',
                                '5'))
            cmd_list = self.host.run('btclient',
                                     args=('--command', 'list bonded',
                                           '--timeout', '5'))
        else:
            regex = re.compile(BT_ADDR_PATTERN)
            self.host.run('bluetoothctl', args=('power', 'on'))
            cmd_list = self.host.run('bluetoothctl',
                                     args=('devices', 'Bonded'))

        output_list = cmd_list.stdout
        for line in output_list.splitlines():
            result = regex.search(line)
            if result is None:
                continue

            addr = result.group().strip('[]')
            if self.floss:
                self.host.run('btclient',
                              args=('--command', 'bond remove ' + addr,
                                    '--timeout', '5'))
            else:
                self.host.run('bluetoothctl', args=('remove', addr))

    def run_once(self, test_name=None, floss=False, args_dict=None):
        """Runs the batch of Bluetooth CTS health tests

        @param test_name: The name of the test
        @param floss: Is Bluetooth running Floss?
        @param args_dict: The arguments passed from the command line
        """
        self.floss = floss
        self.prepare_btpeers(args_dict)

        chrome_feature = None
        if floss:
            # A dirty way to disable floss prevention only for this test
            chrome_feature = 'Floss --disable-feature=FlossIsAvailabilityCheckNeeded'
        try:
            self._run_tradefed_with_retries(test_name=test_name,
                                            run_template=self.run_template,
                                            retry_template=self.retry_template,
                                            target_module=CTS_TARGET_MODULE,
                                            bundle=CTS_BUNDLE,
                                            timeout=CTS_TIMEOUT_SECONDS,
                                            chrome_feature=chrome_feature)
        finally:
            self.stop_btpeers()