Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit a996b8e2 authored by Jizheng Chu's avatar Jizheng Chu Committed by Gerrit Code Review
Browse files

Merge "Unbind GD Cert test classes from GdBaseTestClass"

parents b3fa70d1 630c24bd
Loading
Loading
Loading
Loading
+1 −1
Original line number Diff line number Diff line
@@ -64,6 +64,7 @@ class GdBaseTestClass(BaseTestClass):
        self.rootcanal_running = self.info['rootcanal_running']
        self.rootcanal_logpath = self.info['rootcanal_logpath']
        self.rootcanal_process = self.info['rootcanal_process']
        self.rootcanal_logger = self.info['rootcanal_logger']

        if 'rootcanal' in self.controller_configs:
            asserts.assert_true(self.info['rootcanal_exist'],
@@ -77,7 +78,6 @@ class GdBaseTestClass(BaseTestClass):
                msg="Cannot start root-canal at " + str(self.info['rootcanal']))
            asserts.assert_true(self.info['is_subprocess_alive'], msg="root-canal stopped immediately after running")

            self.rootcanal_logger = self.info['rootcanal_logger']
            self.controller_configs = self.info['controller_configs']

        # Parse and construct GD device objects
+3 −0
Original line number Diff line number Diff line
@@ -42,6 +42,9 @@ def setup_class_core(dut_module, cert_module, verbose_mode, log_path_base, contr

    # Start root-canal if needed
    info['rootcanal_running'] = False
    info['rootcanal_logpath'] = None
    info['rootcanal_process'] = None
    info['rootcanal_logger'] = None
    if 'rootcanal' in info['controller_configs']:
        info['rootcanal_running'] = True
        # Get root canal binary
+7 −114
Original line number Diff line number Diff line
@@ -14,128 +14,21 @@
#   See the License for the specific language governing permissions and
#   limitations under the License.

from datetime import timedelta

from cert.gd_base_test import GdBaseTestClass
from cert.event_stream import EventStream
from cert.truth import assertThat
from cert.py_hal import PyHal
from cert.matchers import HciMatchers
from cert.captures import HciCaptures
from google.protobuf import empty_pb2
from facade import rootservice_pb2 as facade_rootservice_pb2
from hal import hal_facade_pb2 as hal_facade_pb2
from bluetooth_packets_python3 import hci_packets
import bluetooth_packets_python3 as bt_packets
from bluetooth_packets_python3.hci_packets import AclBuilder
from bluetooth_packets_python3 import RawBuilder
from hal.cert.simple_hal_test_lib import SimpleHalTestBase

_GRPC_TIMEOUT = 10


class SimpleHalTest(GdBaseTestClass):
class SimpleHalTest(GdBaseTestClass, SimpleHalTestBase):

    def setup_class(self):
        super().setup_class(dut_module='HAL', cert_module='HAL')
        GdBaseTestClass.setup_class(self, dut_module='HAL', cert_module='HAL')

    def setup_test(self):
        super().setup_test()

        self.dut_hal = PyHal(self.dut)
        self.cert_hal = PyHal(self.cert)

        self.dut_hal.reset()
        self.cert_hal.reset()
        GdBaseTestClass.setup_test(self)
        SimpleHalTestBase.setup_test(self, self.dut, self.cert)

    def teardown_test(self):
        self.dut_hal.close()
        self.cert_hal.close()
        super().teardown_test()

    def test_stream_events(self):
        self.dut_hal.send_hci_command(
            hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

        assertThat(self.dut_hal.get_hci_event_stream()).emits(
            HciMatchers.Exactly(hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)))

    def test_loopback_hci_command(self):
        self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))

        command = hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,
                                                              '0C:05:04:03:02:01')
        self.dut_hal.send_hci_command(command)

        assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LoopbackOf(command))

    def test_inquiry_from_dut(self):
        self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))

        lap = hci_packets.Lap()
        lap.lap = 0x33
        self.dut_hal.send_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'\x02\x0f' in packet.payload
                                                              # Expecting an HCI Event (code 0x02, length 0x0f)
                                                             )

    def test_le_ad_scan_cert_advertises(self):
        self.dut_hal.set_random_le_address('0D:05:04:03:02:01')

        self.dut_hal.set_scan_parameters()
        self.dut_hal.start_scanning()

        advertisement = self.cert_hal.create_advertisement(
            0,
            '0C:05:04:03:02:01',
            min_interval=512,
            max_interval=768,
            peer_address='A6:A5:A4:A3:A2:A1',
            tx_power=0x7f,
            sid=1)
        advertisement.set_data(b'Im_A_Cert')
        advertisement.start()

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload)

        advertisement.stop()

        self.dut_hal.stop_scanning()

    def test_le_connection_dut_advertises(self):
        self.cert_hal.set_random_le_address('0C:05:04:03:02:01')
        self.cert_hal.initiate_le_connection('0D:05:04:03:02:01')

        # DUT Advertises
        advertisement = self.dut_hal.create_advertisement(0, '0D:05:04:03:02:01')
        advertisement.set_data(b'Im_The_DUT')
        advertisement.set_scan_response(b'Im_The_D')
        advertisement.start()

        cert_acl = self.cert_hal.complete_le_connection()
        dut_acl = self.dut_hal.complete_le_connection()

        dut_acl.send_first(b'Just SomeAclData')
        cert_acl.send_first(b'Just SomeMoreAclData')

        assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload)
        assertThat(self.dut_hal.get_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.payload)

    def test_le_connect_list_connection_cert_advertises(self):
        self.dut_hal.set_random_le_address('0D:05:04:03:02:01')
        self.dut_hal.add_to_connect_list('0C:05:04:03:02:01')
        self.dut_hal.initiate_le_connection_by_connect_list('BA:D5:A4:A3:A2:A1')

        advertisement = self.cert_hal.create_advertisement(
            1,
            '0C:05:04:03:02:01',
            min_interval=512,
            max_interval=768,
            peer_address='A6:A5:A4:A3:A2:A1',
            tx_power=0x7F,
            sid=0)
        advertisement.set_data(b'Im_A_Cert')
        advertisement.start()

        assertThat(self.cert_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
        assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
        SimpleHalTestBase.teardown_test(self)
        GdBaseTestClass.teardown_test(self)
+134 −0
Original line number Diff line number Diff line
#!/usr/bin/env python3
#
#   Copyright 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.

from datetime import timedelta

from cert.event_stream import EventStream
from cert.truth import assertThat
from cert.py_hal import PyHal
from cert.matchers import HciMatchers
from cert.captures import HciCaptures
from google.protobuf import empty_pb2
from facade import rootservice_pb2 as facade_rootservice_pb2
from hal import hal_facade_pb2 as hal_facade_pb2
from bluetooth_packets_python3 import hci_packets
import bluetooth_packets_python3 as bt_packets
from bluetooth_packets_python3.hci_packets import AclBuilder
from bluetooth_packets_python3 import RawBuilder

_GRPC_TIMEOUT = 10


class SimpleHalTestBase():

    def setup_test(self, dut, cert):
        self.dut_hal = PyHal(dut)
        self.cert_hal = PyHal(cert)

        self.dut_hal.reset()
        self.cert_hal.reset()

    def teardown_test(self):
        self.dut_hal.close()
        self.cert_hal.close()

    def test_stream_events(self):
        self.dut_hal.send_hci_command(
            hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

        assertThat(self.dut_hal.get_hci_event_stream()).emits(
            HciMatchers.Exactly(hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)))

    def test_loopback_hci_command(self):
        self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))

        command = hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,
                                                              '0C:05:04:03:02:01')
        self.dut_hal.send_hci_command(command)

        assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LoopbackOf(command))

    def test_inquiry_from_dut(self):
        self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))

        lap = hci_packets.Lap()
        lap.lap = 0x33
        self.dut_hal.send_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'\x02\x0f' in packet.payload
                                                              # Expecting an HCI Event (code 0x02, length 0x0f)
                                                             )

    def test_le_ad_scan_cert_advertises(self):
        self.dut_hal.set_random_le_address('0D:05:04:03:02:01')

        self.dut_hal.set_scan_parameters()
        self.dut_hal.start_scanning()

        advertisement = self.cert_hal.create_advertisement(
            0,
            '0C:05:04:03:02:01',
            min_interval=512,
            max_interval=768,
            peer_address='A6:A5:A4:A3:A2:A1',
            tx_power=0x7f,
            sid=1)
        advertisement.set_data(b'Im_A_Cert')
        advertisement.start()

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload)

        advertisement.stop()

        self.dut_hal.stop_scanning()

    def test_le_connection_dut_advertises(self):
        self.cert_hal.set_random_le_address('0C:05:04:03:02:01')
        self.cert_hal.initiate_le_connection('0D:05:04:03:02:01')

        # DUT Advertises
        advertisement = self.dut_hal.create_advertisement(0, '0D:05:04:03:02:01')
        advertisement.set_data(b'Im_The_DUT')
        advertisement.set_scan_response(b'Im_The_D')
        advertisement.start()

        cert_acl = self.cert_hal.complete_le_connection()
        dut_acl = self.dut_hal.complete_le_connection()

        dut_acl.send_first(b'Just SomeAclData')
        cert_acl.send_first(b'Just SomeMoreAclData')

        assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload)
        assertThat(self.dut_hal.get_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.payload)

    def test_le_connect_list_connection_cert_advertises(self):
        self.dut_hal.set_random_le_address('0D:05:04:03:02:01')
        self.dut_hal.add_to_connect_list('0C:05:04:03:02:01')
        self.dut_hal.initiate_le_connection_by_connect_list('BA:D5:A4:A3:A2:A1')

        advertisement = self.cert_hal.create_advertisement(
            1,
            '0C:05:04:03:02:01',
            min_interval=512,
            max_interval=768,
            peer_address='A6:A5:A4:A3:A2:A1',
            tx_power=0x7F,
            sid=0)
        advertisement.set_data(b'Im_A_Cert')
        advertisement.start()

        assertThat(self.cert_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
        assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
+5 −20
Original line number Diff line number Diff line
@@ -14,32 +14,17 @@
#   See the License for the specific language governing permissions and
#   limitations under the License.

import time

from cert.gd_base_test import GdBaseTestClass
from cert.truth import assertThat
from google.protobuf import empty_pb2 as empty_proto
from facade import rootservice_pb2 as facade_rootservice
from hci.facade import controller_facade_pb2 as controller_facade
from hci.cert.controller_test_lib import ControllerTestBase


class ControllerTest(GdBaseTestClass):
class ControllerTest(GdBaseTestClass, ControllerTestBase):

    def setup_class(self):
        super().setup_class(dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')
        GdBaseTestClass.setup_class(self, dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')

    def test_get_addresses(self):
        cert_address = self.cert.hci_controller.GetMacAddressSimple()
        dut_address = self.dut.hci_controller.GetMacAddressSimple()

        assertThat(cert_address).isNotEqualTo(dut_address)
        time.sleep(1)  # This shouldn't be needed b/149120542
        ControllerTestBase.test_get_addresses(self, self.dut, self.cert)

    def test_write_local_name(self):
        self.dut.hci_controller.WriteLocalName(controller_facade.NameMsg(name=b'ImTheDUT'))
        self.cert.hci_controller.WriteLocalName(controller_facade.NameMsg(name=b'ImTheCert'))
        cert_name = self.cert.hci_controller.GetLocalNameSimple()
        dut_name = self.dut.hci_controller.GetLocalNameSimple()

        assertThat(dut_name).isEqualTo(b'ImTheDUT')
        assertThat(cert_name).isEqualTo(b'ImTheCert')
        ControllerTestBase.test_write_local_name(self, self.dut, self.cert)
Loading