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

Commit d6879be6 authored by Henri Chataing's avatar Henri Chataing
Browse files

Remove gd cert test suite NeighborTest

Tests unused module InquiryModule

Bug: 333555245
Test: m .
Flag: EXEMPT, test change
Change-Id: Ib05f281f8388cc8e6a1f63be1f5aa91056e10b3e
parent 59d8660b
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()