• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2022 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import asyncio
16import avatar
17import grpc
18import logging
19
20from avatar import BumblePandoraDevice
21from avatar import PandoraDevice
22from avatar import PandoraDevices
23from mobly import base_test
24from mobly import signals
25from mobly import test_runner
26from mobly.asserts import assert_equal  # type: ignore
27from mobly.asserts import assert_false  # type: ignore
28from mobly.asserts import assert_is_none  # type: ignore
29from mobly.asserts import assert_is_not_none  # type: ignore
30from mobly.asserts import assert_true  # type: ignore
31from mobly.asserts import explicit_pass  # type: ignore
32from pandora.host_pb2 import DISCOVERABLE_GENERAL
33from pandora.host_pb2 import DISCOVERABLE_LIMITED
34from pandora.host_pb2 import NOT_DISCOVERABLE
35from pandora.host_pb2 import Connection
36from pandora.host_pb2 import DiscoverabilityMode
37from typing import Optional
38
39
40class HostTest(base_test.BaseTestClass):  # type: ignore[misc]
41    devices: Optional[PandoraDevices] = None
42
43    # pandora devices.
44    dut: PandoraDevice
45    ref: PandoraDevice
46
47    def setup_class(self) -> None:
48        self.devices = PandoraDevices(self)
49        self.dut, self.ref, *_ = self.devices
50
51        # Enable BR/EDR mode for Bumble devices.
52        for device in self.devices:
53            if isinstance(device, BumblePandoraDevice):
54                device.config.setdefault('classic_enabled', True)
55
56    def teardown_class(self) -> None:
57        if self.devices:
58            self.devices.stop_all()
59
60    @avatar.asynchronous
61    async def setup_test(self) -> None:  # pytype: disable=wrong-arg-types
62        await asyncio.gather(self.dut.reset(), self.ref.reset())
63
64    @avatar.parameterized(
65        (DISCOVERABLE_LIMITED,),
66        (DISCOVERABLE_GENERAL,),
67    )  # type: ignore[misc]
68    def test_discoverable(self, mode: DiscoverabilityMode) -> None:
69        self.dut.host.SetDiscoverabilityMode(mode=mode)
70        inquiry = self.ref.host.Inquiry(timeout=15.0)
71        try:
72            assert_is_not_none(next((x for x in inquiry if x.address == self.dut.address), None))
73        finally:
74            inquiry.cancel()
75
76    # This test should reach the `Inquiry` timeout.
77    @avatar.rpc_except(
78        {
79            grpc.StatusCode.DEADLINE_EXCEEDED: lambda e: explicit_pass(e.details()),
80        }
81    )
82    def test_not_discoverable(self) -> None:
83        self.dut.host.SetDiscoverabilityMode(mode=NOT_DISCOVERABLE)
84        inquiry = self.ref.host.Inquiry(timeout=3.0)
85        try:
86            assert_is_none(next((x for x in inquiry if x.address == self.dut.address), None))
87        finally:
88            inquiry.cancel()
89
90    @avatar.asynchronous
91    async def test_connect(self) -> None:
92        if self.dut.name == 'android':
93            raise signals.TestSkip('TODO: Android connection is too flaky (b/285634621)')
94        ref_dut_res, dut_ref_res = await asyncio.gather(
95            self.ref.aio.host.WaitConnection(address=self.dut.address),
96            self.dut.aio.host.Connect(address=self.ref.address),
97        )
98        assert_is_not_none(ref_dut_res.connection)
99        assert_is_not_none(dut_ref_res.connection)
100        ref_dut, dut_ref = ref_dut_res.connection, dut_ref_res.connection
101        assert ref_dut and dut_ref
102        assert_true(await self.is_connected(self.ref, ref_dut), "")
103
104    @avatar.asynchronous
105    async def test_accept(self) -> None:
106        dut_ref_res, ref_dut_res = await asyncio.gather(
107            self.dut.aio.host.WaitConnection(address=self.ref.address),
108            self.ref.aio.host.Connect(address=self.dut.address),
109        )
110        assert_is_not_none(ref_dut_res.connection)
111        assert_is_not_none(dut_ref_res.connection)
112        ref_dut, dut_ref = ref_dut_res.connection, dut_ref_res.connection
113        assert ref_dut and dut_ref
114        assert_true(await self.is_connected(self.ref, ref_dut), "")
115
116    @avatar.asynchronous
117    async def test_disconnect(self) -> None:
118        if self.dut.name == 'android':
119            raise signals.TestSkip('TODO: Android disconnection is too flaky (b/286081956)')
120        dut_ref_res, ref_dut_res = await asyncio.gather(
121            self.dut.aio.host.WaitConnection(address=self.ref.address),
122            self.ref.aio.host.Connect(address=self.dut.address),
123        )
124        assert_is_not_none(ref_dut_res.connection)
125        assert_is_not_none(dut_ref_res.connection)
126        ref_dut, dut_ref = ref_dut_res.connection, dut_ref_res.connection
127        assert ref_dut and dut_ref
128        await self.dut.aio.host.Disconnect(connection=dut_ref)
129        assert_false(await self.is_connected(self.ref, ref_dut), "")
130
131    async def is_connected(self, device: PandoraDevice, connection: Connection) -> bool:
132        try:
133            await device.aio.host.WaitDisconnection(connection=connection, timeout=5)
134            return False
135        except grpc.RpcError as e:
136            assert_equal(e.code(), grpc.StatusCode.DEADLINE_EXCEEDED)  # type: ignore
137            return True
138
139
140if __name__ == '__main__':
141    logging.basicConfig(level=logging.DEBUG)
142    test_runner.main()  # type: ignore
143