#!/usr/bin/env python3
#
# Copyright (C) 2019 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""This is the PTS base class that is inherited from all PTS
Tests.
"""

import ctypes
import random
import re
import time
import traceback

from ctypes import *
from datetime import datetime

from acts import signals
from acts.base_test import BaseTestClass
from acts.controllers.bluetooth_pts_device import VERDICT_STRINGS
from acts.controllers.fuchsia_device import FuchsiaDevice
from acts.signals import TestSignal
from acts.test_utils.abstract_devices.bluetooth_device import create_bluetooth_device
from acts.test_utils.bt.bt_constants import gatt_transport
from acts.test_utils.fuchsia.bt_test_utils import le_scan_for_device_by_name


class PtsBaseClass(BaseTestClass):
    """ Class for representing common functionality across all PTS tests.

    This includes the ability to rerun tests due to PTS instability,
    common PTS action mappings, and setup/teardown related devices.

    """
    scan_timeout_seconds = 10
    peer_identifier = None

    def __init__(self, controllers):
        BaseTestClass.__init__(self, controllers)
        if 'dut' in self.user_params:
            if self.user_params['dut'] == 'fuchsia_devices':
                self.dut = create_bluetooth_device(self.fuchsia_devices[0])
            elif self.user_params['dut'] == 'android_devices':
                self.dut = create_bluetooth_device(self.android_devices[0])
            else:
                raise ValueError('Invalid DUT specified in config. (%s)' %
                                 self.user_params['dut'])
        else:
            # Default is an fuchsia device
            self.dut = create_bluetooth_device(self.fuchsia_devices[0])

        self.characteristic_read_not_permitted_uuid = self.user_params.get(
            "characteristic_read_not_permitted_uuid")
        self.characteristic_read_not_permitted_handle = self.user_params.get(
            "characteristic_read_not_permitted_handle")
        self.characteristic_read_invalid_handle = self.user_params.get(
            "characteristic_read_invalid_handle")
        self.characteristic_attribute_not_found_uuid = self.user_params.get(
            "characteristic_attribute_not_found_uuid")
        self.characteristic_write_not_permitted_handle = self.user_params.get(
            "characteristic_write_not_permitted_handle")

        self.pts = self.bluetooth_pts_device[0]
        # MMI functions commented out until implemented. Added for tracking
        # purposes.
        self.pts_action_mapping = {
            "A2DP": {
                1: self.a2dp_mmi_iut_connectable,
                1002: self.a2dp_mmi_iut_accept_connect,
                1020: self.a2dp_mmi_initiate_open_stream,
            },
            "GATT": {
                1: self.mmi_make_iut_connectable,
                2: self.mmi_iut_initiate_connection,
                3: self.mmi_iut_initiate_disconnection,
                # 4: self.mmi_iut_no_security,
                # 5: self.mmi_iut_initiate_br_connection,
                10: self.mmi_discover_primary_service,
                # 11: self.mmi_confirm_no_primary_service_small,
                # 12: self.mmi_iut_mtu_exchange,
                # 13: self.mmi_discover_all_service_record,
                # 14: self.mmi_iut_discover_gatt_service_record,
                15: self.mmi_iut_find_included_services,
                # 16: self.mmi_confirm_no_characteristic_uuid_small,
                17: self.mmi_confirm_primary_service,
                # 18: self.mmi_send_primary_service_uuid,
                # 19: self.mmi_confirm_primary_service_uuid,
                # 22: self.confirm_primary_service_1801,
                24: self.mmi_confirm_include_service,
                26: self.mmi_confirm_characteristic_service,
                # 27: self.perform_read_all_characteristics,
                29: self.
                mmi_discover_service_uuid_range,  # AKA: discover service by uuid
                # 31: self.perform_read_all_descriptors,
                48: self.mmi_iut_send_read_characteristic_handle,
                58: self.mmi_iut_send_read_descriptor_handle,
                70: self.mmi_send_write_command,
                74: self.mmi_send_write_request,
                76: self.mmi_send_prepare_write,
                77: self.mmi_iut_send_prepare_write_greater_offset,
                80: self.mmi_iut_send_prepare_write_greater,
                110: self.mmi_iut_enter_handle_read_not_permitted,
                111: self.mmi_iut_enter_uuid_read_not_permitted,
                118: self.mmi_iut_enter_handle_invalid,
                119: self.mmi_iut_enter_uuid_attribute_not_found,
                120: self.mmi_iut_enter_handle_write_not_permitted,
                2000: self.mmi_verify_secure_id,  # Enter pairing pin from DUT.
            },
            "SDP": {
                # TODO: Implement MMIs as necessary
            }
        }

    def setup_class(self):
        self.pts.setup_pts()
        self.pts.bind_to(self.process_next_action)

    def teardown_class(self):
        self.pts.clean_up()

    def setup_test(self):
        # Always start the test with RESULT_INCOMP
        self.pts.pts_test_result = VERDICT_STRINGS['RESULT_INCOMP']

    def teardown_test(self):
        return True

    @staticmethod
    def pts_test_wrap(fn):
        def _safe_wrap_test_case(self, *args, **kwargs):
            test_id = "{}:{}:{}".format(self.__class__.__name__, fn.__name__,
                                        time.time())
            log_string = "[Test ID] {}".format(test_id)
            self.log.info(log_string)
            try:
                self.dut.log_info("Started " + log_string)
                result = fn(self, *args, **kwargs)
                self.dut.log_info("Finished " + log_string)
                rerun_count = self.user_params.get("pts_auto_rerun_count", 0)
                for i in range(int(rerun_count)):
                    if result is not True:
                        self.teardown_test()
                        log_string = "[Rerun Test ID] {}. Run #{} run failed... Retrying".format(
                            test_id, i + 1)
                        self.log.info(log_string)
                        self.setup_test()
                        self.dut.log_info("Rerun Started " + log_string)
                        result = fn(self, *args, **kwargs)
                    else:
                        return result
                return result
            except TestSignal:
                raise
            except Exception as e:
                self.log.error(traceback.format_exc())
                self.log.error(str(e))
                raise
            return fn(self, *args, **kwargs)

        return _safe_wrap_test_case

    def process_next_action(self, profile, action):
        func = self.pts_action_mapping.get(
            self.pts.pts_profile_mmi_request).get(action, "Nothing")
        if func is not 'Nothing':
            func()

    ### BEGIN A2DP MMI Actions ###

    def a2dp_mmi_iut_connectable(self):
        self.dut.start_profile_a2dp_sink()
        self.dut.set_discoverable(True)

    def a2dp_mmi_iut_accept_connect(self):
        self.dut.start_profile_a2dp_sink()
        self.dut.set_discoverable(True)

    def a2dp_mmi_initiate_open_stream(self):
        self.dut.a2dp_initiate_open_stream()

    ### END A2DP MMI Actions ###

    ### BEGIN GATT MMI Actions ###

    def create_write_value_by_size(self, size):
        write_value = []
        for i in range(size):
            write_value.append(i % 256)
        return write_value

    def mmi_send_write_command(self):
        description_to_parse = self.pts.current_implicit_send_description
        raw_handle = re.search('handle = \'(.*)\'O with', description_to_parse)
        handle = int(raw_handle.group(1), 16)
        raw_size = re.search('with <= \'(.*)\' byte', description_to_parse)
        size = int(raw_size.group(1))
        self.dut.gatt_client_write_char_with_no_response_to_input_handle(
            self.peer_identifier, handle,
            self.create_write_value_by_size(size))

    def mmi_send_write_request(self):
        description_to_parse = self.pts.current_implicit_send_description
        raw_handle = re.search('handle = \'(.*)\'O with', description_to_parse)
        handle = int(raw_handle.group(1), 16)
        raw_size = re.search('with <= \'(.*)\' byte', description_to_parse)
        size = int(raw_size.group(1))
        offset = 0
        self.dut.gatt_client_write_to_input_handle(
            self.peer_identifier, handle, offset,
            self.create_write_value_by_size(size))

    def mmi_send_prepare_write(self):
        description_to_parse = self.pts.current_implicit_send_description
        raw_handle = re.search('handle = \'(.*)\'O <=', description_to_parse)
        handle = int(raw_handle.group(1), 16)
        raw_size = re.search('<= \'(.*)\' byte', description_to_parse)
        size = int(math.floor(int(raw_size.group(1)) / 2))
        offset = int(size / 2)
        self.dut.gatt_client_write_to_input_handle(
            self.peer_identifier, handle, offset,
            self.create_write_value_by_size(size))

    def mmi_iut_send_prepare_write_greater_offset(self):
        description_to_parse = self.pts.current_implicit_send_description
        raw_handle = re.search('handle = \'(.*)\'O and', description_to_parse)
        handle = int(raw_handle.group(1), 16)
        raw_offset = re.search('greater than \'(.*)\' byte',
                               description_to_parse)
        offset = int(raw_offset.group(1))
        size = 1
        self.dut.gatt_client_write_to_input_handle(
            self.peer_identifier, handle, offset,
            self.create_write_value_by_size(size))

    def mmi_iut_send_prepare_write_greater(self):
        description_to_parse = self.pts.current_implicit_send_description
        raw_handle = re.search('handle = \'(.*)\'O with', description_to_parse)
        handle = int(raw_handle.group(1), 16)
        raw_size = re.search('greater than \'(.*)\' byte',
                             description_to_parse)
        size = int(raw_size.group(1))
        offset = 0
        self.dut.gatt_client_write_to_input_handle(
            self.peer_identifier, handle, offset,
            self.create_write_value_by_size(size))

    def mmi_make_iut_connectable(self):
        adv_data = {"name": self.dut_bluetooth_local_name}
        self.dut.start_le_advertisement(adv_data, self.ble_advertise_interval)

    def mmi_iut_enter_uuid_read_not_permitted(self):
        self.pts.extra_answers.append(
            self.characteristic_read_not_permitted_uuid)

    def mmi_iut_enter_handle_read_not_permitted(self):
        self.pts.extra_answers.append(
            self.characteristic_read_not_permitted_handle)

    def mmi_iut_enter_handle_invalid(self):
        self.pts.extra_answers.append(self.characteristic_read_invalid_handle)

    def mmi_iut_enter_uuid_attribute_not_found(self):
        self.pts.extra_answers.append(
            self.characteristic_attribute_not_found_uuid)

    def mmi_iut_enter_handle_write_not_permitted(self):
        self.pts.extra_answers.append(
            self.characteristic_write_not_permitted_handle)

    def mmi_verify_secure_id(self):
        self.pts.extra_answers.append(self.dut.get_pairing_pin())

    def mmi_discover_service_uuid_range(self, uuid):
        self.dut.gatt_client_mmi_discover_service_uuid_range(
            self.peer_identifier, uuid)

    def mmi_iut_initiate_connection(self):
        autoconnect = False
        transport = gatt_transport['le']
        adv_name = "PTS"
        self.peer_identifier = self.dut.le_scan_with_name_filter(
            "PTS", self.scan_timeout_seconds)
        if self.peer_identifier is None:
            raise signals.TestFailure("Scanner unable to find advertisement.")
        tries = 3
        for _ in range(tries):
            if self.dut.gatt_connect(self.peer_identifier, transport,
                                     autoconnect):
                return

        raise signals.TestFailure("Unable to connect to peripheral.")

    def mmi_iut_initiate_disconnection(self):
        if not self.dut.gatt_disconnect(self.peer_identifier):
            raise signals.TestFailure("Failed to disconnect from peer.")

    def mmi_discover_primary_service(self):
        self.dut.gatt_refresh()

    def mmi_iut_find_included_services(self):
        self.dut.gatt_refresh()

        test_result = self.pts.execute_test(test_name)
        return test_result

    def mmi_confirm_primary_service(self):
        # TODO: Write verifier that 1800 and 1801 exists. For now just pass.
        return True

    def mmi_confirm_characteristic_service(self):
        # TODO: Write verifier that no services exist. For now just pass.
        return True

    def mmi_confirm_include_service(self, uuid_description):
        # TODO: Write verifier that input services exist. For now just pass.
        # Note: List comes in the form of a long string to parse:
        # Attribute Handle = '0002'O Included Service Attribute handle = '0080'O,End Group Handle = '0085'O,Service UUID = 'A00B'O
        # \n
        # Attribute Handle = '0021'O Included Service Attribute handle = '0001'O,End Group Handle = '0006'O,Service UUID = 'A00D'O
        # \n ...
        return True

    def _read_generic_handle(self):
        description_to_parse = self.pts.current_implicit_send_description
        raw_handle = re.search('handle = \'(.*)\'O to', description_to_parse)
        handle = int(raw_handle.group(1), 16)
        self.dut.gatt_client_read_input_handle(self.peer_identifier, handle)

    def mmi_iut_send_read_characteristic_handle(self):
        self._read_generic_handle()

    def mmi_iut_send_read_descriptor_handle(self):
        self._read_generic_handle()

    ### END GATT MMI Actions ###