Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix directed advertising test #61

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Fix directed advertising test #61

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 49 additions & 1 deletion avatar/cases/le_host_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
from avatar import BumblePandoraDevice
from avatar import PandoraDevice
from avatar import PandoraDevices
from avatar import aio
from avatar import pandora_snippet
from mobly import base_test
from mobly import test_runner
from mobly.asserts import assert_equal # type: ignore
Expand All @@ -35,6 +37,7 @@
from pandora.host_pb2 import Connection
from pandora.host_pb2 import DataTypes
from pandora.host_pb2 import OwnAddressType
from pandora.security_pb2 import LE_LEVEL3
from typing import Any, Dict, Literal, Optional, Union


Expand Down Expand Up @@ -67,6 +70,7 @@ def setup_class(self) -> None:
for device in self.devices:
if isinstance(device, BumblePandoraDevice):
device.config.setdefault('classic_enabled', True)
device.config.setdefault('address_resolution_offload', True)

def teardown_class(self) -> None:
if self.devices:
Expand Down Expand Up @@ -105,6 +109,45 @@ def test_scan(
scan_response_data = DataTypes() if scannable == 'scannable' else None
target = self.dut.address if directed == 'directed' else None

async def setup_pairing(
initiator: PandoraDevice,
acceptor: PandoraDevice,
initiator_addr_type: OwnAddressType,
acceptor_addr_type: OwnAddressType,
) -> None:
# Acceptor - Advertise
advertisement = acceptor.aio.host.Advertise(
legacy=True,
connectable=True,
own_address_type=acceptor_addr_type,
data=DataTypes(manufacturer_specific_data=b'pause cafe'),
)

# Initiator - Scan and fetch the address
scan = initiator.aio.host.Scan(own_address_type=initiator_addr_type)
acceptor_scan = await anext(
(x async for x in scan if b'pause cafe' in x.data.manufacturer_specific_data)
) # pytype: disable=name-error
scan.cancel()

ini_cer, cer_ini = await pandora_snippet.connect_le(
initiator, advertisement, acceptor_scan, initiator_addr_type
)

await asyncio.gather(
initiator.aio.security.Secure(connection=ini_cer, le=LE_LEVEL3),
acceptor.aio.security.WaitSecurity(connection=cer_ini, le=LE_LEVEL3),
)

await asyncio.gather(
initiator.aio.host.Disconnect(connection=ini_cer),
acceptor.aio.host.WaitDisconnection(connection=cer_ini),
)

if directed == 'directed' and self.dut.name == 'android':
# Android cannot advertise with Public address, so devices must be paired and enable resolution to perform directed advertisement
aio.run_until_complete(setup_pairing(self.dut, self.ref, RANDOM, RANDOM))

advertise = self.ref.host.Advertise(
legacy=True,
connectable=is_connectable,
Expand All @@ -114,7 +157,12 @@ def test_scan(
own_address_type=PUBLIC,
)

scan = self.dut.host.Scan(legacy=False, passive=False, timeout=self.scan_timeout)
scan = self.dut.host.Scan(
legacy=False,
passive=False,
own_address_type=RANDOM if self.dut.name == 'android' else PUBLIC,
timeout=self.scan_timeout,
)
report = next((x for x in scan if x.public == self.ref.address))
try:
report = next((x for x in scan if x.public == self.ref.address))
Expand Down