1# Copyright (C) 2022 The Android Open Source Project 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# http://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 15"""Fast Pair seeker role.""" 16 17import json 18from typing import Callable, Optional 19 20from mobly import asserts 21from mobly.controllers import android_device 22from mobly.controllers.android_device_lib import snippet_event 23 24from test_helper import event_helper 25from test_helper import utils 26 27# The package name of the Nearby Mainline Fast Pair seeker Mobly snippet. 28FP_SEEKER_SNIPPETS_PACKAGE = 'android.nearby.multidevices' 29 30# Events reported from the seeker snippet. 31ON_PROVIDER_FOUND_EVENT = 'onDiscovered' 32ON_MANAGE_ACCOUNT_DEVICE_EVENT = 'onManageAccountDevice' 33 34# Abbreviations for common use type. 35AndroidDevice = android_device.AndroidDevice 36JsonObject = utils.JsonObject 37ProviderAccountKeyCallable = Callable[[], Optional[str]] 38SnippetEvent = snippet_event.SnippetEvent 39wait_for_event = event_helper.wait_callback_event 40 41 42class FastPairSeeker: 43 """A proxy for seeker snippet on the device.""" 44 45 def __init__(self, ad: AndroidDevice) -> None: 46 self._ad = ad 47 self._ad.debug_tag = 'MainlineFastPairSeeker' 48 self._scan_result_callback = None 49 self._pairing_result_callback = None 50 51 def load_snippet(self) -> None: 52 """Starts the seeker snippet and connects. 53 54 Raises: 55 SnippetError: Illegal load operations are attempted. 56 """ 57 self._ad.load_snippet(name='fp', package=FP_SEEKER_SNIPPETS_PACKAGE) 58 59 def start_scan(self) -> None: 60 """Starts scanning to find Fast Pair provider devices.""" 61 self._scan_result_callback = self._ad.fp.startScan() 62 63 def stop_scan(self) -> None: 64 """Stops the Fast Pair seeker scanning.""" 65 self._ad.fp.stopScan() 66 67 def wait_and_assert_provider_found(self, timeout_seconds: int, 68 expected_model_id: str, 69 expected_ble_mac_address: str) -> None: 70 """Waits and asserts any onDiscovered event from the seeker. 71 72 Args: 73 timeout_seconds: The number of seconds to wait before giving up. 74 expected_model_id: The expected model ID of the remote Fast Pair provider 75 device. 76 expected_ble_mac_address: The expected BLE MAC address of the remote Fast 77 Pair provider device. 78 """ 79 80 def _on_provider_found_event_received(provider_found_event: SnippetEvent, 81 elapsed_time: int) -> bool: 82 nearby_device_str = provider_found_event.data['device'] 83 self._ad.log.info('Seeker discovered first provider(%s) in %d seconds.', 84 nearby_device_str, elapsed_time) 85 return expected_ble_mac_address in nearby_device_str 86 87 def _on_provider_found_event_waiting(elapsed_time: int) -> None: 88 self._ad.log.info( 89 'Still waiting "%s" event callback from seeker side ' 90 'after %d seconds...', ON_PROVIDER_FOUND_EVENT, elapsed_time) 91 92 def _on_provider_found_event_missed() -> None: 93 asserts.fail(f'Timed out after {timeout_seconds} seconds waiting for ' 94 f'the specific "{ON_PROVIDER_FOUND_EVENT}" event.') 95 96 wait_for_event( 97 callback_event_handler=self._scan_result_callback, 98 event_name=ON_PROVIDER_FOUND_EVENT, 99 timeout_seconds=timeout_seconds, 100 on_received=_on_provider_found_event_received, 101 on_waiting=_on_provider_found_event_waiting, 102 on_missed=_on_provider_found_event_missed) 103 104 def put_anti_spoof_key_device_metadata(self, model_id: str, kdm_json_file_name: str) -> None: 105 """Puts a model id to FastPairAntispoofKeyDeviceMetadata pair into test data cache. 106 107 Args: 108 model_id: A string of model id to be associated with. 109 kdm_json_file_name: The FastPairAntispoofKeyDeviceMetadata JSON object. 110 """ 111 self._ad.log.info('Puts FastPairAntispoofKeyDeviceMetadata into test data cache for ' 112 'model id "%s".', model_id) 113 kdm_json_object = utils.load_json_fast_pair_test_data(kdm_json_file_name) 114 self._ad.fp.putAntispoofKeyDeviceMetadata( 115 model_id, 116 utils.serialize_as_simplified_json_str(kdm_json_object)) 117 118 def set_fast_pair_scan_enabled(self, enable: bool) -> None: 119 """Writes into Settings whether Fast Pair scan is enabled. 120 121 Args: 122 enable: whether the Fast Pair scan should be enabled. 123 """ 124 self._ad.log.info('%s Fast Pair scan in Android settings.', 125 'Enables' if enable else 'Disables') 126 self._ad.fp.setFastPairScanEnabled(enable) 127 128 def wait_and_assert_halfsheet_showed(self, timeout_seconds: int, 129 expected_model_id: str) -> None: 130 """Waits and asserts the onHalfSheetShowed event from the seeker. 131 132 Args: 133 timeout_seconds: The number of seconds to wait before giving up. 134 expected_model_id: A 3-byte hex string for seeker side to recognize 135 the remote provider device (ex: 0x00000c). 136 """ 137 self._ad.log.info('Waits and asserts the half sheet showed for model id "%s".', 138 expected_model_id) 139 self._ad.fp.waitAndAssertHalfSheetShowed(expected_model_id, timeout_seconds) 140 141 def dismiss_halfsheet(self) -> None: 142 """Dismisses the half sheet UI if showed.""" 143 self._ad.fp.dismissHalfSheet() 144 145 def start_pairing(self) -> None: 146 """Starts pairing the provider via "Connect" button on half sheet UI.""" 147 self._pairing_result_callback = self._ad.fp.startPairing() 148 149 def wait_and_assert_account_device( 150 self, timeout_seconds: int, 151 get_account_key_from_provider: ProviderAccountKeyCallable) -> None: 152 """Waits and asserts the onHalfSheetShowed event from the seeker. 153 154 Args: 155 timeout_seconds: The number of seconds to wait before giving up. 156 get_account_key_from_provider: The callable to get expected account key from the provider 157 side. 158 """ 159 160 def _on_manage_account_device_event_received(manage_account_device_event: SnippetEvent, 161 elapsed_time: int) -> bool: 162 account_key_json_str = manage_account_device_event.data['accountDeviceJsonString'] 163 account_key_from_seeker = json.loads(account_key_json_str)['account_key'] 164 account_key_from_provider = get_account_key_from_provider() 165 self._ad.log.info('Seeker add an account device with account key "%s" in %d seconds.', 166 account_key_from_seeker, elapsed_time) 167 self._ad.log.info('The latest provider side account key is "%s".', 168 account_key_from_provider) 169 return account_key_from_seeker == account_key_from_provider 170 171 def _on_manage_account_device_event_waiting(elapsed_time: int) -> None: 172 self._ad.log.info( 173 'Still waiting "%s" event callback from seeker side ' 174 'after %d seconds...', ON_MANAGE_ACCOUNT_DEVICE_EVENT, elapsed_time) 175 176 def _on_manage_account_device_event_missed() -> None: 177 asserts.fail(f'Timed out after {timeout_seconds} seconds waiting for ' 178 f'the specific "{ON_MANAGE_ACCOUNT_DEVICE_EVENT}" event.') 179 180 wait_for_event( 181 callback_event_handler=self._pairing_result_callback, 182 event_name=ON_MANAGE_ACCOUNT_DEVICE_EVENT, 183 timeout_seconds=timeout_seconds, 184 on_received=_on_manage_account_device_event_received, 185 on_waiting=_on_manage_account_device_event_waiting, 186 on_missed=_on_manage_account_device_event_missed) 187