1# Copyright (C) 2024 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# Lint as: python3 16"""Wi-Fi Aware Matchfilter test reimplemented in Mobly.""" 17import base64 18import enum 19import logging 20import random 21import sys 22 23from android.platform.test.annotations import ApiTest 24from aware import aware_lib_utils as autils 25from aware import constants 26from mobly import asserts 27from mobly import base_test 28from mobly import records 29from mobly import test_runner 30from mobly import utils 31from mobly.controllers import android_device 32from mobly.snippet import errors 33 34RUNTIME_PERMISSIONS = ( 35 'android.permission.ACCESS_FINE_LOCATION', 36 'android.permission.ACCESS_COARSE_LOCATION', 37 'android.permission.NEARBY_WIFI_DEVICES', 38) 39PACKAGE_NAME = constants.WIFI_AWARE_SNIPPET_PACKAGE_NAME 40snippets_to_load = [ 41 ('wifi_aware_snippet', PACKAGE_NAME), 42 ('wifi', constants.WIFI_SNIPPET_PACKAGE_NAME), 43] 44_DEFAULT_TIMEOUT = constants.WAIT_WIFI_STATE_TIME_OUT.total_seconds() 45_MSG_ID_SUB_TO_PUB = random.randint(1000, 5000) 46_MSG_ID_PUB_TO_SUB = random.randint(5001, 9999) 47_MSG_SUB_TO_PUB = "Let's talk [Random Identifier: %s]" % utils.rand_ascii_str(5) 48_MSG_PUB_TO_SUB = 'Ready [Random Identifier: %s]' % utils.rand_ascii_str(5) 49_CALLBACK_NAME = constants.DiscoverySessionCallbackParamsType.CALLBACK_NAME 50_IS_SESSION_INIT = constants.DiscoverySessionCallbackParamsType.IS_SESSION_INIT 51 52# Publish & Subscribe Config keys. 53_PAYLOAD_SIZE_MIN = 0 54_PAYLOAD_SIZE_TYPICAL = 1 55_PAYLOAD_SIZE_MAX = 2 56_PUBLISH_TYPE_UNSOLICITED = 0 57_PUBLISH_TYPE_SOLICITED = 1 58_SUBSCRIBE_TYPE_PASSIVE = 0 59_SUBSCRIBE_TYPE_ACTIVE = 1 60 61 62@enum.unique 63class AttachCallBackMethodType(enum.StrEnum): 64 """Represents Attach Callback Method Type in Wi-Fi Aware. 65 66 https://developer.android.com/reference/android/net/wifi/aware/AttachCallback 67 """ 68 ATTACHED = 'onAttached' 69 ATTACH_FAILED = 'onAttachFailed' 70 AWARE_SESSION_TERMINATED = 'onAwareSessionTerminated' 71 72 73class WifiAwareMatchFilterTest(base_test.BaseTestClass): 74 """Set of tests for Wi-Fi Aware Match Filter behavior. These all 75 use examples from Appendix H of the Wi-Fi Aware standard.""" 76 77 ads: list[android_device.AndroidDevice] 78 publisher: android_device.AndroidDevice 79 subscriber: android_device.AndroidDevice 80 81 SERVICE_NAME = "GoogleTestServiceMFMFMF" 82 83 MF_NNNNN = bytes([0x0, 0x0, 0x0, 0x0, 0x0]) 84 MF_12345 = bytes([0x1, 0x1, 0x1, 0x2, 0x1, 0x3, 0x1, 0x4, 0x1, 0x5]) 85 MF_12145 = bytes([0x1, 0x1, 0x1, 0x2, 0x1, 0x1, 0x1, 0x4, 0x1, 0x5]) 86 MF_1N3N5 = bytes([0x1, 0x1, 0x0, 0x1, 0x3, 0x0, 0x1, 0x5]) 87 MF_N23N5 = bytes([0x0, 0x1, 0x2, 0x1, 0x3, 0x0, 0x1, 0x5]) 88 MF_N2N4 = bytes([0x0, 0x1, 0x2, 0x0, 0x1, 0x4]) 89 MF_1N3N = bytes([0x1, 0x1, 0x0, 0x1, 0x3, 0x0]) 90 91 match_filters = [ 92 [None, None, True, True], 93 [None, MF_NNNNN, True, True], 94 [MF_NNNNN, None, True, True], 95 [None, MF_12345, True, False], 96 [MF_12345, None, False, True], 97 [MF_NNNNN, MF_12345, True, True], 98 [MF_12345, MF_NNNNN, True, True], 99 [MF_12345, MF_12345, True, True], 100 [MF_12345, MF_12145, False, False], 101 [MF_1N3N5, MF_12345, True, True], 102 [MF_12345, MF_N23N5, True, True], 103 [MF_N2N4, MF_12345, True, False], 104 [MF_12345, MF_1N3N, False, True] 105 ] 106 107 def setup_class(self): 108 # Register two Android devices. 109 self.ads = self.register_controller(android_device, min_number=2) 110 self.publisher = self.ads[0] 111 self.subscriber = self.ads[1] 112 113 def setup_device(device: android_device.AndroidDevice): 114 for snippet_name, package_name in snippets_to_load: 115 device.load_snippet(snippet_name, package_name) 116 for permission in RUNTIME_PERMISSIONS: 117 device.adb.shell(['pm', 'grant', package_name, permission]) 118 asserts.abort_all_if( 119 not device.wifi_aware_snippet.wifiAwareIsAvailable(), 120 f'{device} Wi-Fi Aware is not available.', 121 ) 122 123 # Set up devices in parallel. 124 utils.concurrent_exec( 125 setup_device, 126 ((self.publisher,), (self.subscriber,)), 127 max_workers=2, 128 raise_on_exception=True, 129 ) 130 131 def setup_test(self): 132 for ad in self.ads: 133 ad.wifi.wifiEnable() 134 aware_avail = ad.wifi_aware_snippet.wifiAwareIsAvailable() 135 if not aware_avail: 136 ad.log.info('Aware not available. Waiting ...') 137 state_handler = ( 138 ad.wifi_aware_snippet.wifiAwareMonitorStateChange()) 139 state_handler.waitAndGet( 140 constants.WifiAwareBroadcast.WIFI_AWARE_AVAILABLE) 141 142 def teardown_test(self): 143 utils.concurrent_exec( 144 self._teardown_test_on_device, 145 ((self.publisher,), (self.subscriber,)), 146 max_workers=2, 147 raise_on_exception=True, 148 ) 149 utils.concurrent_exec( 150 lambda d: d.services.create_output_excerpts_all( 151 self.current_test_info), 152 param_list=[[ad] for ad in self.ads], 153 raise_on_exception=True, 154 ) 155 156 def _teardown_test_on_device(self, 157 ad: android_device.AndroidDevice) -> None: 158 ad.wifi_aware_snippet.wifiAwareCloseAllWifiAwareSession() 159 ad.wifi.wifiClearConfiguredNetworks() 160 ad.wifi.wifiEnable() 161 if ad.is_adb_root: 162 autils.reset_device_parameters(ad) 163 autils.reset_device_statistics(ad) 164 autils.validate_forbidden_callbacks(ad) 165 166 def on_fail(self, record: records.TestResult) -> None: 167 android_device.take_bug_reports(self.ads, 168 destination = 169 self.current_test_info.output_path) 170 171 def _start_attach(self, ad: android_device.AndroidDevice) -> str: 172 """Starts the attach process on the provided device.""" 173 handler = ad.wifi_aware_snippet.wifiAwareAttach() 174 attach_event = handler.waitAndGet( 175 event_name=AttachCallBackMethodType.ATTACHED, 176 timeout=_DEFAULT_TIMEOUT, 177 ) 178 asserts.assert_true( 179 ad.wifi_aware_snippet.wifiAwareIsSessionAttached( 180 handler.callback_id), 181 f'{ad} attach succeeded, but Wi-Fi Aware session is still null.' 182 ) 183 ad.log.info('Attach Wi-Fi Aware session succeeded.') 184 return attach_event.callback_id 185 186 def run_discovery(self, p_dut, s_dut, 187 p_mf, 188 s_mf, 189 do_unsolicited_passive, 190 expect_discovery): 191 """Creates a discovery session (publish and subscribe) with. 192 the specified configuration. 193 194 Args: 195 p_dut: Device to use as publisher. 196 s_dut: Device to use as subscriber. 197 p_mf: Publish's match filter. 198 s_mf: Subscriber's match filter. 199 do_unsolicited_passive: True to use an Unsolicited/ 200 Passive discovery, 201 False for a Solicited/ 202 Active discovery session. 203 expect_discovery: True if service should be discovered, 204 False otherwise. 205 Returns: True on success, False on failure (based on expect_discovery 206 arg) 207 """ 208 209 # Encode the match filter 210 p_mf = base64.b64encode( 211 p_mf).decode("utf-8") if p_mf is not None else None 212 s_mf = base64.b64encode( 213 s_mf).decode("utf-8") if s_mf is not None else None 214 215 # Publisher+Subscriber: attach and wait for confirmation 216 p_id = self._start_attach(p_dut) 217 s_id = self._start_attach(s_dut) 218 219 # Publisher: start publish and wait for confirmation 220 p_config = autils.create_discovery_config(self.SERVICE_NAME, 221 p_type = 222 _PUBLISH_TYPE_UNSOLICITED 223 if do_unsolicited_passive 224 else _PUBLISH_TYPE_SOLICITED, 225 s_type = None, 226 match_filter_list = p_mf) 227 dut_p_mf = p_dut.wifi_aware_snippet.wifiAwarePublish( 228 p_id, p_config 229 ) 230 p_dut.log.info('Created the DUT publish session %s', dut_p_mf) 231 p_discovery = dut_p_mf.waitAndGet( 232 constants.DiscoverySessionCallbackMethodType.DISCOVER_RESULT) 233 callback_name = p_discovery.data[_CALLBACK_NAME] 234 asserts.assert_equal( 235 constants.DiscoverySessionCallbackMethodType.PUBLISH_STARTED, 236 callback_name, 237 f'{p_dut} DUT publish failed, got callback: {callback_name}.', 238 ) 239 240 # Subscriber: start subscribe and wait for confirmation 241 s_config = autils.create_discovery_config(self.SERVICE_NAME, 242 p_type = None, 243 s_type = 244 _SUBSCRIBE_TYPE_PASSIVE 245 if do_unsolicited_passive 246 else _SUBSCRIBE_TYPE_ACTIVE, 247 match_filter_list=s_mf) 248 dut_s_mf = s_dut.wifi_aware_snippet.wifiAwareSubscribe( 249 s_id, s_config 250 ) 251 s_dut.log.info('Created the DUT subscribe session.: %s', dut_s_mf) 252 s_discovery = dut_s_mf.waitAndGet( 253 constants.DiscoverySessionCallbackMethodType.DISCOVER_RESULT, 254 timeout=_DEFAULT_TIMEOUT) 255 callback_name = s_discovery.data[_CALLBACK_NAME] 256 asserts.assert_equal( 257 constants.DiscoverySessionCallbackMethodType.SUBSCRIBE_STARTED, 258 callback_name, 259 f'{s_dut} DUT subscribe failed, got callback: {callback_name}.', 260 ) 261 event = None 262 try: 263 event = dut_s_mf.waitAndGet( 264 constants.DiscoverySessionCallbackMethodType.SERVICE_DISCOVERED, 265 timeout=_DEFAULT_TIMEOUT) 266 s_dut.log.info( 267 "[Subscriber] SESSION_CB_ON_SERVICE_DISCOVERED: %s",event) 268 except errors.CallbackHandlerTimeoutError: 269 s_dut.log.info( 270 "[Subscriber] No SESSION_CB_ON_SERVICE_DISCOVERED: %s",event) 271 pass 272 p_dut.wifi_aware_snippet.wifiAwareCloseDiscoverSession( 273 dut_p_mf.callback_id) 274 s_dut.wifi_aware_snippet.wifiAwareCloseDiscoverSession( 275 dut_s_mf.callback_id) 276 277 p_dut.wifi_aware_snippet.wifiAwareCloseAllWifiAwareSession() 278 279 s_dut.wifi_aware_snippet.wifiAwareCloseAllWifiAwareSession() 280 281 if expect_discovery: 282 return event is not None 283 else: 284 return event is None 285 286 def run_match_filters_per_spec(self, do_unsolicited_passive): 287 """Validate all the match filter combinations in the Wi-Fi Aware spec, 288 Appendix H. 289 290 Args: 291 do_unsolicited_passive: True to run the Unsolicited/Passive tests, 292 False to run the Solicited/Active tests. 293 """ 294 p_dut = self.ads[0] 295 s_dut = self.ads[1] 296 p_dut.pretty_name = "Publisher" 297 s_dut.pretty_name = "Subscriber" 298 fails = [] 299 for i in range(len(self.match_filters)): 300 test_info = self.match_filters[i] 301 if do_unsolicited_passive: 302 pub_type = "Unsolicited" 303 sub_type = "Passive" 304 pub_mf = test_info[0] 305 sub_mf = test_info[1] 306 expect_discovery = test_info[3] 307 else: 308 pub_type = "Solicited" 309 sub_type = "Active" 310 pub_mf = test_info[1] 311 sub_mf = test_info[0] 312 expect_discovery = test_info[2] 313 314 logging.info("Test #%d: %s Pub MF=%s, %s Sub MF=%s: Discovery %s", 315 i, pub_type, pub_mf, sub_type, sub_mf, "EXPECTED" 316 if test_info[2] else "UNEXPECTED") 317 result = self.run_discovery( 318 p_dut, 319 s_dut, 320 p_mf=pub_mf, 321 s_mf=sub_mf, 322 do_unsolicited_passive = do_unsolicited_passive, 323 expect_discovery = expect_discovery) 324 logging.info("Test #%d %s Pub/%s Sub %s", i, pub_type, sub_type, 325 "PASS" if result else "FAIL") 326 if not result: 327 fails.append(i) 328 logging.info("fails: %s", fails) 329 330 asserts.assert_true( 331 len(fails) == 0, 332 "Some match filter tests are failing", 333 extras={"data": fails}) 334 335 @ApiTest( 336 apis=[ 337 'android.net.wifi.aware.WifiAwareManager#attach(android.net.wifi.aware.AttachCallback, android.net.wifi.aware.IdentityChangedListener, android.os.Handler)', 338 'android.net.wifi.aware.WifiAwareSession#publish(android.net.wifi.aware.PublishConfig, android.net.wifi.aware.DiscoverySessionCallback, android.os.Handler)', 339 'android.net.wifi.aware.WifiAwareSession#subscrible(android.net.wifi.aware.SubscribeConfig, android.net.wifi.aware.DiscoverySessionCallback, android.os.Handler)', 340 'android.net.wifi.aware.PublishConfig.Builder#setPublishType(PublishConfig.PUBLISH_TYPE_UNSOLICITED)', 341 'android.net.wifi.aware.SubscribeConfig.Builder#setSubscribeType(SubscribeConfig.SUBSCRIBE_TYPE_PASSIVE)', 342 'android.net.wifi.aware.DiscoverySession#sendMessage(int, byte[])', 343 ] 344 ) 345 346 def test_match_filters_per_spec_unsolicited_passive(self): 347 """Validate all the match filter combinations in the Wi-Fi Aware spec, 348 Appendix H for Unsolicited Publish (tx filter) Passive Subscribe (rx 349 filter)""" 350 self.run_match_filters_per_spec(do_unsolicited_passive=True) 351 352 @ApiTest( 353 apis=[ 354 'android.net.wifi.aware.WifiAwareManager#attach(android.net.wifi.aware.AttachCallback, android.net.wifi.aware.IdentityChangedListener, android.os.Handler)', 355 'android.net.wifi.aware.WifiAwareSession#publish(android.net.wifi.aware.PublishConfig, android.net.wifi.aware.DiscoverySessionCallback, android.os.Handler)', 356 'android.net.wifi.aware.WifiAwareSession#subscrible(android.net.wifi.aware.SubscribeConfig, android.net.wifi.aware.DiscoverySessionCallback, android.os.Handler)', 357 'android.net.wifi.aware.PublishConfig.Builder#setPublishType(PublishConfig.PUBLISH_TYPE_SOLICITED)', 358 'android.net.wifi.aware.SubscribeConfig.Builder#setSubscribeType(SubscribeConfig.SUBSCRIBE_TYPE_ACTIVE)', 359 'android.net.wifi.aware.DiscoverySession#sendMessage(int, byte[])', 360 ] 361 ) 362 363 def test_match_filters_per_spec_solicited_active(self): 364 """Validate all the match filter combinations in the Wi-Fi Aware spec, 365 Appendix H for Solicited Publish (rx filter) Active Subscribe (tx 366 filter)""" 367 self.run_match_filters_per_spec(do_unsolicited_passive=False) 368 369 370if __name__ == '__main__': 371 # Take test args 372 if '--' in sys.argv: 373 index = sys.argv.index('--') 374 sys.argv = sys.argv[:1] + sys.argv[index + 1:] 375 376 test_runner.main() 377