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

Commit 54043c67 authored by Henri Chataing's avatar Henri Chataing Committed by Gerrit Code Review
Browse files

Merge "Remove gd cert test suite NeighborTest" into main

parents 0c027b37 d6879be6
Loading
Loading
Loading
Loading
+0 −71
Original line number Diff line number Diff line
#!/usr/bin/env python3
#
#   Copyright 2020 - 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 blueberry.tests.gd.cert.event_stream import EventStream
from blueberry.tests.gd.cert.event_stream import IEventStream
from blueberry.tests.gd.cert.closable import Closable
from blueberry.tests.gd.cert.closable import safeClose
from blueberry.tests.gd.cert.truth import assertThat
from google.protobuf import empty_pb2 as empty_proto
from blueberry.facade.hci import hci_facade_pb2 as hci_facade
from blueberry.facade.neighbor import facade_pb2 as neighbor_facade
import hci_packets as hci


class InquirySession(Closable, IEventStream):

    def __init__(self, device, inquiry_msg):
        self.inquiry_event_stream = EventStream(device.neighbor.SetInquiryMode(inquiry_msg))

    def get_event_queue(self):
        return self.inquiry_event_stream.get_event_queue()

    def close(self):
        safeClose(self.inquiry_event_stream)


class GetRemoteNameSession(Closable):

    def __init__(self, device):
        self.remote_name_stream = EventStream(device.neighbor.GetRemoteNameEvents(empty_proto.Empty()))

    def verify_name(self, name):
        assertThat(self.remote_name_stream).emits(lambda msg: bytes(name) in msg.name)

    def close(self):
        safeClose(self.remote_name_stream)


class PyNeighbor(object):

    def __init__(self, device):
        self.device = device

    def set_inquiry_mode(self, inquiry_msg):
        """
        Set the inquiry mode and return a session which can be used for event queue assertion
        """
        return InquirySession(self.device, inquiry_msg)

    def get_remote_name(self, remote_address: str):
        """
        Get the remote name and return a session which can be used for event queue assertion
        """
        self.device.neighbor.ReadRemoteName(
            neighbor_facade.RemoteNameRequestMsg(address=remote_address.encode('utf8'),
                                                 page_scan_repetition_mode=1,
                                                 clock_offset=0x6855))
        return GetRemoteNameSession(self.device)
+1 −2
Original line number Diff line number Diff line
@@ -24,7 +24,6 @@ from blueberry.tests.gd.hci.le_advertising_manager_test import LeAdvertisingMana
from blueberry.tests.gd.hci.le_scanning_manager_test import LeScanningManagerTest
from blueberry.tests.gd.hci.le_scanning_with_security_test import LeScanningWithSecurityTest
from blueberry.tests.gd.iso.le_iso_test import LeIsoTest
from blueberry.tests.gd.neighbor.neighbor_test import NeighborTest
from blueberry.tests.gd.security.le_security_test import LeSecurityTest
from blueberry.tests.gd.security.security_test import SecurityTest
from blueberry.tests.gd.shim.shim_test import ShimTest
@@ -35,7 +34,7 @@ from mobly import suite_runner
ALL_TESTS = {
    CertSelfTest, SimpleHalTest, AclManagerTest, ControllerTest, DirectHciTest, LeAclManagerTest,
    LeAdvertisingManagerTest, LeScanningManagerTest, LeScanningWithSecurityTest, LeIsoTest,
    NeighborTest, LeSecurityTest, SecurityTest, ShimTest, StackTest
    LeSecurityTest, SecurityTest, ShimTest, StackTest
}

DISABLED_TESTS = set()
+0 −94
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 blueberry.tests.gd.cert.matchers import HciMatchers, NeighborMatchers
from blueberry.tests.gd.cert.py_hci import PyHci
from blueberry.tests.gd.cert.truth import assertThat
from blueberry.tests.gd.cert.py_neighbor import PyNeighbor
from blueberry.facade.neighbor import facade_pb2 as neighbor_facade
from blueberry.tests.gd.cert import gd_base_test
from mobly import test_runner
import hci_packets as hci


class NeighborTest(gd_base_test.GdBaseTestClass):

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

    def setup_test(self):
        gd_base_test.GdBaseTestClass.setup_test(self)
        self.cert_hci = PyHci(self.cert, acl_streaming=True)
        self.cert_hci.send_command(hci.WriteScanEnable(scan_enable=hci.ScanEnable.INQUIRY_AND_PAGE_SCAN))
        self.cert_name = b'Im_A_Cert'
        self.cert_address = self.cert_hci.read_own_address()
        self.cert_name += b'@' + repr(self.cert_address).encode('utf-8')
        self.dut_neighbor = PyNeighbor(self.dut)

    def teardown_test(self):
        self.cert_hci.close()
        gd_base_test.GdBaseTestClass.teardown_test(self)

    def _set_name(self):
        padded_name = self.cert_name
        while len(padded_name) < 248:
            padded_name = padded_name + b'\0'
        self.cert_hci.send_command(hci.WriteLocalName(local_name=padded_name))

        assertThat(self.cert_hci.get_event_stream()).emits(HciMatchers.CommandComplete(hci.OpCode.WRITE_LOCAL_NAME))

    def test_inquiry_from_dut(self):
        inquiry_msg = neighbor_facade.InquiryMsg(inquiry_mode=neighbor_facade.DiscoverabilityMode.GENERAL,
                                                 result_mode=neighbor_facade.ResultMode.STANDARD,
                                                 length_1_28s=30,
                                                 max_results=0)
        session = self.dut_neighbor.set_inquiry_mode(inquiry_msg)
        self.cert_hci.send_command(hci.WriteScanEnable(scan_enable=hci.ScanEnable.INQUIRY_AND_PAGE_SCAN))
        assertThat(session).emits(NeighborMatchers.InquiryResult(self.cert_address))

    def test_inquiry_rssi_from_dut(self):
        inquiry_msg = neighbor_facade.InquiryMsg(inquiry_mode=neighbor_facade.DiscoverabilityMode.GENERAL,
                                                 result_mode=neighbor_facade.ResultMode.RSSI,
                                                 length_1_28s=31,
                                                 max_results=0)
        session = self.dut_neighbor.set_inquiry_mode(inquiry_msg)
        self.cert_hci.send_command(hci.WriteScanEnable(scan_enable=hci.ScanEnable.INQUIRY_AND_PAGE_SCAN))
        assertThat(session).emits(NeighborMatchers.InquiryResultwithRssi(self.cert_address))

    def test_inquiry_extended_from_dut(self):
        self._set_name()
        self.cert_hci.send_command(
            hci.WriteExtendedInquiryResponse(fec_required=hci.FecRequired.NOT_REQUIRED,
                                             extended_inquiry_response=[
                                                 hci.GapData(data_type=hci.GapDataType.COMPLETE_LOCAL_NAME,
                                                             data=list(bytes(self.cert_name)))
                                             ]))
        inquiry_msg = neighbor_facade.InquiryMsg(inquiry_mode=neighbor_facade.DiscoverabilityMode.GENERAL,
                                                 result_mode=neighbor_facade.ResultMode.EXTENDED,
                                                 length_1_28s=32,
                                                 max_results=0)
        session = self.dut_neighbor.set_inquiry_mode(inquiry_msg)
        self.cert_hci.send_command(hci.WriteScanEnable(scan_enable=hci.ScanEnable.INQUIRY_AND_PAGE_SCAN))
        assertThat(session).emits(NeighborMatchers.ExtendedInquiryResult(self.cert_address))

    def test_remote_name(self):
        self._set_name()
        session = self.dut_neighbor.get_remote_name(repr(self.cert_address))
        session.verify_name(self.cert_name)


if __name__ == '__main__':
    test_runner.main()