#
#   Copyright 2021 - 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.

import time
import logging

from bluetooth_packets_python3 import hci_packets
from cert.event_stream import EventStream
from cert.matchers import HciMatchers, IsoMatchers, L2capMatchers
from cert.metadata import metadata
from cert.py_hci import PyHci
from cert.py_l2cap import PyLeL2cap
from cert.py_le_iso import PyLeIso
from cert.py_le_iso import CisTestParameters
from cert.truth import assertThat
from datetime import timedelta
from facade import common_pb2 as common
from hci.facade import controller_facade_pb2 as controller_facade
from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade
from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade
from google.protobuf import empty_pb2 as empty_proto
from neighbor.facade import facade_pb2 as neighbor_facade
from l2cap.le.cert.cert_le_l2cap import CertLeL2cap
from iso.cert.cert_le_iso import CertLeIso
from mobly import asserts


class LeIsoTestBase():
    """
        Collection of tests that each sample results from
        different (unique) combinations of io capabilities, authentication requirements, and oob data.
    """

    def setup_test(self, dut, cert):
        self.dut = dut
        self.cert = cert

        self.dut_l2cap = PyLeL2cap(self.dut)
        self.cert_l2cap = CertLeL2cap(self.cert)
        self.dut_address = common.BluetoothAddressWithType(
            address=common.BluetoothAddress(address=bytes(b'D0:05:04:03:02:01')), type=common.RANDOM_DEVICE_ADDRESS)
        self.cert_address = common.BluetoothAddressWithType(
            address=common.BluetoothAddress(address=bytes(b'C0:11:FF:AA:33:22')), type=common.RANDOM_DEVICE_ADDRESS)
        dut_privacy_policy = le_initiator_address_facade.PrivacyPolicy(
            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
            address_with_type=self.dut_address,
            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
            minimum_rotation_time=0,
            maximum_rotation_time=0)
        self.dut_l2cap._device.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(dut_privacy_policy)
        privacy_policy = le_initiator_address_facade.PrivacyPolicy(
            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
            address_with_type=self.cert_address,
            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
            minimum_rotation_time=0,
            maximum_rotation_time=0)
        self.cert_l2cap._device.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(privacy_policy)

        self.dut_iso = PyLeIso(self.dut)
        self.cert_iso = CertLeIso(self.cert)

    def teardown_test(self):
        self.dut_iso.close()
        self.cert_iso.close()

        self.cert_l2cap.close()
        self.dut_l2cap.close()

    #cert becomes central of connection, dut peripheral
    def _setup_link_from_cert(self):
        # DUT Advertises
        gap_name = hci_packets.GapData()
        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
        gap_name.data = list(bytes(b'Im_The_DUT'))
        gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
        config = le_advertising_facade.AdvertisingConfig(
            advertisement=[gap_data],
            interval_min=512,
            interval_max=768,
            advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
            own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
            channel_map=7,
            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
        request = le_advertising_facade.CreateAdvertiserRequest(config=config)
        create_response = self.dut.hci_le_advertising_manager.CreateAdvertiser(request)
        self.cert_l2cap.connect_le_acl(self.dut_address)

    def _setup_cis_from_cert(self, cig_id, sdu_interval_m_to_s, sdu_interval_s_to_m, ft_m_to_s, ft_s_to_m, iso_interval,
                             peripherals_clock_accuracy, packing, framing, max_transport_latency_m_to_s,
                             max_transport_latency_s_to_m, cis_configs):
        self.cert_iso.le_set_cig_parameters_test(cig_id, sdu_interval_m_to_s, sdu_interval_s_to_m, ft_m_to_s, ft_s_to_m,
                                                 iso_interval, peripherals_clock_accuracy, packing, framing,
                                                 max_transport_latency_m_to_s, max_transport_latency_s_to_m,
                                                 cis_configs)

        cis_handles = self.cert_iso.wait_le_set_cig_parameters_complete()

        cis_handle = cis_handles[0]

        acl_connection_handle = self.cert_l2cap._le_acl.handle
        self.cert_iso.le_cretate_cis([(cis_handle, acl_connection_handle)])
        dut_cis_stream = self.dut_iso.wait_le_cis_established()
        cert_cis_stream = self.cert_iso.wait_le_cis_established()
        return (dut_cis_stream, cert_cis_stream)

    def skip_if_iso_not_supported(self):
        supported = self.dut.hci_controller.IsSupportedCommand(
            controller_facade.OpCodeMsg(op_code=int(hci_packets.OpCode.LE_SET_CIG_PARAMETERS)))
        if (not supported.supported):
            asserts.skip("Skipping this test.  The chip doesn't support LE ISO")

    @metadata(
        pts_test_id="IAL/CIS/UNF/SLA/BV-01-C",
        pts_test_name="connected isochronous stream, unframed data, peripheral role")
    def test_iso_cis_unf_sla_bv_01_c(self):
        self.skip_if_iso_not_supported()
        """
            Verify that the IUT can send an SDU with length ≤ the Isochronous PDU length.
        """
        cig_id = 0x01
        sdu_interval_m_to_s = 0
        sdu_interval_s_to_m = 0x186a
        ft_m_to_s = 0
        ft_s_to_m = 1
        iso_interval = 0x0A
        peripherals_clock_accuracy = 0
        packing = 0
        framing = 0
        max_transport_latency_m_to_s = 0
        max_transport_latency_s_to_m = 0
        cis_configs = [
            CisTestParameters(
                cis_id=0x01,
                nse=2,
                max_sdu_m_to_s=100,
                max_sdu_s_to_m=100,
                max_pdu_m_to_s=100,
                max_pdu_s_to_m=100,
                phy_m_to_s=0x02,
                phy_s_to_m=0x00,
                bn_m_to_s=0,
                bn_s_to_m=2,
            )
        ]

        self._setup_link_from_cert()
        (dut_cis_stream, cert_cis_stream) = self._setup_cis_from_cert(
            cig_id, sdu_interval_m_to_s, sdu_interval_s_to_m, ft_m_to_s, ft_s_to_m, iso_interval,
            peripherals_clock_accuracy, packing, framing, max_transport_latency_m_to_s, max_transport_latency_s_to_m,
            cis_configs)
        dut_cis_stream.send(b'abcdefgh' * 10)
        assertThat(cert_cis_stream).emits(IsoMatchers.Data(b'abcdefgh' * 10))

    @metadata(
        pts_test_id="IAL/CIS/UNF/SLA/BV-25-C",
        pts_test_name="connected isochronous stream, unframed data, peripheral role")
    def test_iso_cis_unf_sla_bv_25_c(self):
        self.skip_if_iso_not_supported()
        """
            Verify that the IUT can send an SDU with length ≤ the Isochronous PDU length.
        """
        cig_id = 0x01
        sdu_interval_m_to_s = 0x7530
        sdu_interval_s_to_m = 0x7530
        ft_m_to_s = 3
        ft_s_to_m = 2
        iso_interval = 0x18
        peripherals_clock_accuracy = 0
        packing = 0
        framing = 0
        max_transport_latency_m_to_s = 0
        max_transport_latency_s_to_m = 0
        cis_configs = [
            CisTestParameters(
                cis_id=0x01,
                nse=5,
                max_sdu_m_to_s=100,
                max_sdu_s_to_m=100,
                max_pdu_m_to_s=100,
                max_pdu_s_to_m=100,
                phy_m_to_s=0x02,
                phy_s_to_m=0x00,
                bn_m_to_s=3,
                bn_s_to_m=1,
            )
        ]

        self._setup_link_from_cert()
        (dut_cis_stream, cert_cis_stream) = self._setup_cis_from_cert(
            cig_id, sdu_interval_m_to_s, sdu_interval_s_to_m, ft_m_to_s, ft_s_to_m, iso_interval,
            peripherals_clock_accuracy, packing, framing, max_transport_latency_m_to_s, max_transport_latency_s_to_m,
            cis_configs)
        dut_cis_stream.send(b'abcdefgh' * 10)
        assertThat(cert_cis_stream).emits(IsoMatchers.Data(b'abcdefgh' * 10))

    @metadata(
        pts_test_id="IAL/CIS/FRA/SLA/BV-03-C",
        pts_test_name="connected isochronous stream, framed data, peripheral role")
    def test_iso_cis_fra_sla_bv_03_c(self):
        self.skip_if_iso_not_supported()
        """
            Verify that the IUT can send an SDU with length ≤ the Isochronous PDU length.
        """
        cig_id = 0x01
        sdu_interval_m_to_s = 0x0000
        sdu_interval_s_to_m = 0x4e30
        ft_m_to_s = 0
        ft_s_to_m = 2
        iso_interval = 0x14
        peripherals_clock_accuracy = 0
        packing = 0
        framing = 1
        max_transport_latency_m_to_s = 0
        max_transport_latency_s_to_m = 0
        cis_configs = [
            CisTestParameters(
                cis_id=0x01,
                nse=4,
                max_sdu_m_to_s=100,
                max_sdu_s_to_m=100,
                max_pdu_m_to_s=100,
                max_pdu_s_to_m=100,
                phy_m_to_s=0x02,
                phy_s_to_m=0x00,
                bn_m_to_s=0,
                bn_s_to_m=2,
            )
        ]

        self._setup_link_from_cert()
        (dut_cis_stream, cert_cis_stream) = self._setup_cis_from_cert(
            cig_id, sdu_interval_m_to_s, sdu_interval_s_to_m, ft_m_to_s, ft_s_to_m, iso_interval,
            peripherals_clock_accuracy, packing, framing, max_transport_latency_m_to_s, max_transport_latency_s_to_m,
            cis_configs)
        dut_cis_stream.send(b'abcdefgh' * 10)
        assertThat(cert_cis_stream).emits(IsoMatchers.Data(b'abcdefgh' * 10))

    @metadata(
        pts_test_id="IAL/CIS/FRA/SLA/BV-26-C",
        pts_test_name="connected isochronous stream, framed data, peripheral role")
    def test_iso_cis_fra_sla_bv_26_c(self):
        self.skip_if_iso_not_supported()
        """
            Verify that the IUT can send an SDU with length ≤ the Isochronous PDU length.
        """
        cig_id = 0x01
        sdu_interval_m_to_s = 0x14D5
        sdu_interval_s_to_m = 0x14D5
        ft_m_to_s = 1
        ft_s_to_m = 1
        iso_interval = 0x08
        peripherals_clock_accuracy = 0
        packing = 0
        framing = 1
        max_transport_latency_m_to_s = 0
        max_transport_latency_s_to_m = 0
        cis_configs = [
            CisTestParameters(
                cis_id=0x01,
                nse=2,
                max_sdu_m_to_s=100,
                max_sdu_s_to_m=100,
                max_pdu_m_to_s=100,
                max_pdu_s_to_m=100,
                phy_m_to_s=0x02,
                phy_s_to_m=0x00,
                bn_m_to_s=1,
                bn_s_to_m=1,
            )
        ]

        self._setup_link_from_cert()
        (dut_cis_stream, cert_cis_stream) = self._setup_cis_from_cert(
            cig_id, sdu_interval_m_to_s, sdu_interval_s_to_m, ft_m_to_s, ft_s_to_m, iso_interval,
            peripherals_clock_accuracy, packing, framing, max_transport_latency_m_to_s, max_transport_latency_s_to_m,
            cis_configs)
        dut_cis_stream.send(b'abcdefgh' * 10)
        assertThat(cert_cis_stream).emits(IsoMatchers.Data(b'abcdefgh' * 10))