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
16
17from abc import ABC, abstractmethod
18
19from mobly import base_test
20
21from avatar import BumblePandoraDevice, PandoraDevice, PandoraDevices
22from avatar.aio import asynchronous
23from typing import Optional
24
25class PairTestBase(ABC, base_test.BaseTestClass):  # type: ignore[misc]
26
27    devices: Optional[PandoraDevices] = None
28
29    dut: PandoraDevice
30    ref: PandoraDevice
31
32    @abstractmethod
33    def _setup_devices(self):
34        raise NotImplementedError
35
36    def setup_class(self) -> None:
37        self.devices = PandoraDevices(self)
38        self.dut, self.ref, *_ = self.devices
39
40        self._setup_devices()
41
42    def teardown_class(self) -> None:
43        if self.devices:
44            self.devices.stop_all()
45
46    @asynchronous
47    async def setup_test(self) -> None:
48        await asyncio.gather(self.dut.reset(), self.ref.reset())
49
50    def teardown_test(self):
51        pass
52
53    @property
54    def acl_initiator(self):
55        return self._acl_initiator
56
57    @acl_initiator.setter
58    def acl_initiator(self, acl_initiator: PandoraDevice):
59        self._acl_initiator = acl_initiator
60
61    @property
62    def acl_responder(self):
63        return self._acl_responder
64
65    @acl_responder.setter
66    def acl_responder(self, acl_responder: PandoraDevice):
67        self._acl_responder = acl_responder
68
69    @property
70    def service_initiator(self):
71        return self._service_initiator
72
73    @service_initiator.setter
74    def service_initiator(self, service_initiator: PandoraDevice):
75        self._service_initiator = service_initiator
76
77    @property
78    def service_responder(self):
79        return self._service_responder
80
81    @service_responder.setter
82    def service_responder(self, service_responder: PandoraDevice):
83        self._service_responder = service_responder
84
85    @property
86    def pairing_initiator(self):
87        return self._pairing_initiator
88
89    @pairing_initiator.setter
90    def pairing_initiator(self, pairing_initiator: PandoraDevice):
91        self._pairing_initiator = pairing_initiator
92
93    @property
94    def initiator_pairing_event_stream(self):
95        if self.pairing_initiator == self.dut:
96            return self.android_pairing_stream
97
98        return self.bumble_pairing_stream
99
100    @property
101    def pairing_responder(self):
102        return self._pairing_responder
103
104    @pairing_responder.setter
105    def pairing_responder(self, pairing_responder: PandoraDevice):
106        self._pairing_responder = pairing_responder
107
108    @property
109    def responder_pairing_event_stream(self):
110        if self.pairing_initiator == self.dut:
111            return self.bumble_pairing_stream
112
113        return self.android_pairing_stream
114
115    @abstractmethod
116    async def start_acl_connection(self):
117        raise NotImplementedError
118
119    @abstractmethod
120    async def start_pairing(
121        self,
122        initiator_acl_connection,
123        responder_acl_connection,
124    ):
125        raise NotImplementedError
126
127    @abstractmethod
128    async def start_service_access(
129        self,
130        initiator_acl_connection,
131        responder_acl_connection,
132    ):
133        raise NotImplementedError
134
135    @abstractmethod
136    async def accept_pairing(self):
137        raise NotImplementedError
138
139    def prepare_pairing(self):
140        self.android_pairing_stream = self.dut.aio.security.OnPairing()
141        setattr(self.android_pairing_stream, 'device', self.dut)
142
143        self.bumble_pairing_stream = self.ref.aio.security.OnPairing()
144        setattr(self.bumble_pairing_stream, 'device', self.ref)
145