1# Copyright 2024 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 logging
17
18from pairing.br_edr.test_base import BREDRPairTestBase
19
20from mobly.asserts import assert_equal, fail
21
22from avatar import asynchronous
23
24from pandora.security_pb2 import PairingEventAnswer
25
26class BREDRLegacyTestClass(BREDRPairTestBase):
27
28    def _setup_devices(self) -> None:
29
30        self.ref.config.setdefault('classic_enabled', True)
31        self.ref.config.setdefault('le_enabled', False)
32        self.ref.config.setdefault('classic_ssp_enabled', False)
33
34        self.ref.config.setdefault(
35                    'server',
36                    {
37                        # Android io_capability: display_yesno
38                        'io_capability': 'keyboard_input_only',
39                    },
40                )
41
42    async def accept_pairing(self):
43        expected_pairing_method = 'pin_code_request'
44        pairing_pin_code = b'123456'
45
46        # initiator receives pin code request
47        init_ev = await anext(self.initiator_pairing_event_stream)
48        logging.debug(f'init_ev.method_variant():{init_ev.method_variant()}')
49        assert_equal(init_ev.method_variant(), expected_pairing_method)
50        init_ev_ans = PairingEventAnswer(event=init_ev, pin=pairing_pin_code)
51
52        # accept pairing on initiator with pairing pin code
53        self.initiator_pairing_event_stream.send_nowait(init_ev_ans)
54
55        # responder receives pin code request
56        responder_ev = await anext(self.responder_pairing_event_stream)
57        logging.debug(f'responder_ev.method_variant():{responder_ev.method_variant()}')
58        assert_equal(responder_ev.method_variant(), expected_pairing_method)
59        responder_ev_ans = PairingEventAnswer(event=responder_ev, pin=pairing_pin_code)
60
61        # accept pairing on responder with pairing pin code
62        self.responder_pairing_event_stream.send_nowait(responder_ev_ans)
63
64    @asynchronous
65    async def test_dedicated_pairing_acl_init_by_bumble_and_bumble_as_pair_initiator(self) -> None:
66        '''
67        acl:
68            ref: initiator
69            dut: responder
70
71        pairing:
72            ref: initiator
73            dut: responder
74        '''
75
76        # setting up roles
77        self.acl_initiator = self.ref
78        self.acl_responder = self.dut
79        self.pairing_initiator = self.ref
80        self.pairing_responder = self.dut
81
82        # do pairing test
83        self.prepare_pairing()
84
85        # first initiate an ACL connection from bumble to android
86        bumble_res, android_res = await self.start_acl_connection()
87
88        # bumble initiates the pairing
89        pairing_task = asyncio.create_task(self.start_pairing(bumble_res.connection, android_res.connection))
90
91        await self.accept_pairing()
92
93        await asyncio.wait_for(pairing_task, timeout=10.0)
94
95    @asynchronous
96    async def test_dedicated_pairing_acl_init_by_bumble_and_bumble_as_pair_responder(self) -> None:
97        '''
98        acl:
99            ref: initiator
100            dut: responder
101
102        pairing:
103            ref: responder
104            dut: initiator
105        '''
106
107        # role setup
108        self.acl_initiator = self.ref
109        self.acl_responder = self.dut
110        self.pairing_initiator = self.dut
111        self.pairing_responder = self.ref
112
113        self.prepare_pairing()
114
115        # first initiate an ACL connection from bumble to android
116        bumble_res, android_res = await self.start_acl_connection()
117
118        # Android initiates the pairing
119        pairing_task = asyncio.create_task(self.start_pairing(android_res.connection, bumble_res.connection))
120
121        await self.accept_pairing()
122
123        await asyncio.wait_for(pairing_task, timeout=10.0)
124
125    @asynchronous
126    async def test_dedicated_pairing_acl_init_by_phone_and_bumble_as_pair_responder(self) -> None:
127        '''
128        acl:
129            ref: responder
130            dut: initiator
131
132        pairing:
133            ref: responder
134            dut: initiator
135
136        Note: we can not change the role of pairing actions in the current avatar
137        implementation, as the implementation of Connect (initiating acl connection)
138        on Android will initiate pairing.
139
140        Pairing initiated from ref is not supported yet
141        '''
142        # role setup
143        self.acl_initiator = self.dut
144        self.acl_responder = self.ref
145        self.pairing_initiator = self.dut
146        self.pairing_responder = self.ref
147
148        self.prepare_pairing()
149
150        acl_connection_task = asyncio.create_task(self.start_acl_connection())
151
152        # with the ACL connection, pairing will be automatically started
153        # on Android
154        await self.accept_pairing()
155
156        await asyncio.wait_for(acl_connection_task, timeout=10.0)
157
158    @asynchronous
159    async def test_general_pairing(self) -> None:
160        # role setup
161        self.acl_initiator = self.ref
162        self.acl_responder = self.dut
163        self.pairing_initiator = self.dut
164        self.pairing_responder = self.ref
165        self.service_initiator = self.ref
166        self.service_responder = self.dut
167
168        self.prepare_pairing()
169
170        # first initiate an ACL connection from bumble to android
171        android_res, bumble_res = await self.start_acl_connection()
172
173        service_access_task = asyncio.create_task(self.start_service_access(bumble_res.connection, android_res.connection))
174
175        # pairing will be started automatically when a secure service is
176        # accessed
177        await self.accept_pairing()
178
179        try:
180            _ = await asyncio.wait_for(service_access_task, timeout=5.0)
181        except:
182            fail("access should have succeeded")
183