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"""Test cases for Wi-Fi p2p service discovery.""" 17from collections.abc import Sequence 18import logging 19 20from android.platform.test.annotations import ApiTest 21from direct import constants 22from direct import p2p_utils 23from mobly import asserts 24from mobly import base_test 25from mobly import records 26from mobly import test_runner 27from mobly import utils 28from mobly.controllers import android_device 29 30import wifi_test_utils 31 32 33@ApiTest( 34 [ 35 "android.net.wifi.p2p.WifiP2pManager#discoverServices(android.net.wifi.p2p.WifiP2pManager.Channel, android.net.wifi.p2p.WifiP2pManager.ActionListener)", 36 "android.net.wifi.p2p.WifiP2pManager#addServiceRequest(android.net.wifi.p2p.WifiP2pManager.Channel, android.net.wifi.p2p.WifiP2pManager.WifiP2pServiceRequest, android.net.wifi.p2p.WifiP2pManager.ActionListener)", 37 "android.net.wifi.p2p.WifiP2pManager#setUpnpServiceResponseListener(android.net.wifi.p2p.WifiP2pManager.Channel, android.net.wifi.p2p.WifiP2pManager.UpnpServiceResponseListener)", 38 "android.net.wifi.p2p.WifiP2pManager#setDnsSdResponseListeners(android.net.wifi.p2p.WifiP2pManager.Channel, android.net.wifi.p2p.WifiP2pManager.DnsSdServiceResponseListener, android.net.wifi.p2p.WifiP2pManager.DnsSdTxtRecordListener)", 39 "android.net.wifi.p2p.WifiP2pManager#removeServiceRequest(android.net.wifi.p2p.WifiP2pManager.Channel, android.net.wifi.p2p.WifiP2pManager.WifiP2pServiceRequest, android.net.wifi.p2p.WifiP2pManager.ActionListener)", 40 "android.net.wifi.p2p.WifiP2pManager#clearServiceRequests(android.net.wifi.p2p.WifiP2pManager.Channel, android.net.wifi.p2p.WifiP2pManager.ActionListener)" 41 ] 42) 43class ServiceDiscoveryTest(base_test.BaseTestClass): 44 """Test cases for Wi-Fi p2p service discovery. 45 46 Test Preconditions: 47 Two Android phones that support Wi-Fi Direct. 48 49 Test steps are described in the docstring of each test case. 50 """ 51 52 ads: Sequence[android_device.AndroidDevice] 53 responder_ad: android_device.AndroidDevice 54 requester_ad: android_device.AndroidDevice 55 56 def setup_class(self) -> None: 57 super().setup_class() 58 self.ads = self.register_controller(android_device, min_number=2) 59 utils.concurrent_exec( 60 self._setup_device, 61 param_list=[[ad] for ad in self.ads], 62 raise_on_exception=True, 63 ) 64 self.responder_ad, self.requester_ad, *_ = self.ads 65 self.responder_ad.debug_tag = ( 66 f'{self.responder_ad.serial}(Responder)' 67 ) 68 self.requester_ad.debug_tag = f'{self.requester_ad.serial}(Requester)' 69 70 def _setup_device(self, ad: android_device.AndroidDevice) -> None: 71 ad.load_snippet('wifi', constants.WIFI_SNIPPET_PACKAGE_NAME) 72 wifi_test_utils.set_screen_on_and_unlock(ad) 73 wifi_test_utils.enable_wifi_verbose_logging(ad) 74 # Clear all saved Wi-Fi networks. 75 ad.wifi.wifiDisable() 76 ad.wifi.wifiClearConfiguredNetworks() 77 ad.wifi.wifiEnable() 78 79 def test_search_all_services_01(self) -> None: 80 """Searches all p2p services with API 81 `WifiP2pServiceRequest.newInstance(WifiP2pServiceInfo.SERVICE_TYPE_ALL)`. 82 83 Test Steps: 84 1. Initialize Wi-Fi p2p on both devices. 85 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 86 (responder). 87 3. Add service request on another device (requester) with API 88 `WifiP2pServiceRequest.newInstance(WifiP2pServiceInfo.SERVICE_TYPE_ALL)`. 89 4. Initiate p2p service discovery on the requester. Verify that the requester 90 discovers all services. 91 """ 92 requester, responder = self._setup_wifi_p2p() 93 self._add_p2p_services(responder) 94 requester.ad.wifi.wifiP2pAddServiceRequest( 95 constants.ServiceType.ALL 96 ) 97 self._search_p2p_services( 98 responder, 99 requester, 100 expected_dns_sd_sequence=constants.ServiceData.ALL_DNS_SD, 101 expected_dns_txt_sequence=constants.ServiceData.ALL_DNS_TXT, 102 expected_upnp_sequence=constants.ServiceData.ALL_UPNP_SERVICES, 103 ) 104 105 def test_search_all_services_02(self) -> None: 106 """Searches all p2p services with API 107 `WifiP2pServiceRequest.newInstance(SERVICE_TYPE_BONJOUR/SERVICE_TYPE_UPNP)` 108 109 Test Steps: 110 1. Initialize Wi-Fi p2p on both devices. 111 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 112 (responder). 113 3. Add service request on another device (requester) with API 114 `WifiP2pServiceRequest.newInstance(SERVICE_TYPE_BONJOUR)` and 115 `WifiP2pServiceRequest.newInstance(SERVICE_TYPE_UPNP)`. 116 4. Initiate p2p service discovery on the requester. Verify that the requester discovers 117 all services. 118 """ 119 requester, responder = self._setup_wifi_p2p() 120 self._add_p2p_services(responder) 121 requester.ad.wifi.wifiP2pAddServiceRequest( 122 constants.ServiceType.BONJOUR, 123 ) 124 requester.ad.wifi.wifiP2pAddServiceRequest( 125 constants.ServiceType.UPNP, 126 ) 127 self._search_p2p_services( 128 responder, 129 requester, 130 expected_dns_sd_sequence=constants.ServiceData.ALL_DNS_SD, 131 expected_dns_txt_sequence=constants.ServiceData.ALL_DNS_TXT, 132 expected_upnp_sequence=constants.ServiceData.ALL_UPNP_SERVICES, 133 ) 134 135 def test_search_all_services_03(self) -> None: 136 """Searches all p2p services with API `WifiP2pDnsSdServiceRequest.newInstance()` 137 and `WifiP2pUpnpServiceRequest.newInstance()`. 138 139 Test Steps: 140 1. Initialize Wi-Fi p2p on both devices. 141 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 142 (responder). 143 3. Add service request on another device (requester) with API 144 `WifiP2pUpnpServiceRequest.newInstance()` and 145 `WifiP2pDnsSdServiceRequest.newInstance()`. 146 4. Initiate p2p service discovery on the requester. Verify that the requester discovers 147 all services. 148 """ 149 requester, responder = self._setup_wifi_p2p() 150 self._add_p2p_services(responder) 151 requester.ad.wifi.wifiP2pAddBonjourServiceRequest() 152 requester.ad.wifi.wifiP2pAddUpnpServiceRequest() 153 self._search_p2p_services( 154 responder, 155 requester, 156 expected_dns_sd_sequence=constants.ServiceData.ALL_DNS_SD, 157 expected_dns_txt_sequence=constants.ServiceData.ALL_DNS_TXT, 158 expected_upnp_sequence=constants.ServiceData.ALL_UPNP_SERVICES, 159 ) 160 161 def test_serv_req_dns_ptr(self) -> None: 162 """Searches Bonjour services with Bonjour domain. 163 164 Test Steps: 165 1. Initialize Wi-Fi p2p on both devices. 166 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 167 (responder). 168 3. Add Bonjour service request with service type `_ipp._tcp`. 169 4. Initiate p2p service discovery on the requester. Verify that the requester discovers 170 expected services. 171 """ 172 requester, responder = self._setup_wifi_p2p() 173 self._add_p2p_services(responder) 174 requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 175 None, # instanceName 176 '_ipp._tcp', 177 ) 178 self._search_p2p_services( 179 responder, 180 requester, 181 expected_dns_sd_sequence=constants.ServiceData.IPP_DNS_SD, 182 expected_dns_txt_sequence=(), 183 expected_upnp_sequence=(), 184 ) 185 186 def test_serv_req_dns_txt(self) -> None: 187 """Searches Bonjour services with TXT record. 188 189 Test Steps: 190 1. Initialize Wi-Fi p2p on both devices. 191 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 192 (responder). 193 3. Add Bonjour service request with instance name `MyPrinter` and 194 service type `_ipp._tcp`. 195 4. Initiate p2p service discovery on the requester. Verify that the requester discovers 196 expected services. 197 """ 198 requester, responder = self._setup_wifi_p2p() 199 self._add_p2p_services(responder) 200 requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 201 'MyPrinter', 202 '_ipp._tcp', 203 ) 204 self._search_p2p_services( 205 responder, 206 requester, 207 expected_dns_sd_sequence=(), 208 expected_dns_txt_sequence=constants.ServiceData.IPP_DNS_TXT, 209 expected_upnp_sequence=(), 210 ) 211 212 def test_serv_req_upnp_all(self) -> None: 213 """Searches all UPnP services with service type `ssdp:all`. 214 215 Test Steps: 216 1. Initialize Wi-Fi p2p on both devices. 217 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 218 (responder). 219 3. Add UPnP service request with service type `ssdp:all`. 220 4. Initiate p2p service discovery on the requester. Verify that the requester discovers 221 expected services. 222 """ 223 requester, responder = self._setup_wifi_p2p() 224 self._add_p2p_services(responder) 225 requester.ad.wifi.wifiP2pAddUpnpServiceRequest('ssdp:all') 226 self._search_p2p_services( 227 responder, 228 requester, 229 expected_dns_sd_sequence=(), 230 expected_dns_txt_sequence=(), 231 expected_upnp_sequence=constants.ServiceData.ALL_UPNP_SERVICES, 232 ) 233 234 def test_serv_req_upnp_root_device(self) -> None: 235 """Searches UPnP root devices. 236 237 Test Steps: 238 1. Initialize Wi-Fi p2p on both devices. 239 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 240 (responder). 241 3. Add UPnP service request with service type `upnp:rootdevice`. 242 4. Initiate p2p service discovery on the requester. Verify that the requester discovers 243 expected services. 244 """ 245 requester, responder = self._setup_wifi_p2p() 246 self._add_p2p_services(responder) 247 requester.ad.wifi.wifiP2pAddUpnpServiceRequest('upnp:rootdevice') 248 self._search_p2p_services( 249 responder, 250 requester, 251 expected_dns_sd_sequence=(), 252 expected_dns_txt_sequence=(), 253 expected_upnp_sequence=constants.ServiceData.UPNP_ROOT_DEVICE, 254 ) 255 256 def test_serv_req_remove_request(self) -> None: 257 """Checks that API `WifiP2pManager#removeServiceRequest` works well. 258 259 Test Steps: 260 1. Initialize Wi-Fi p2p on both devices. 261 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 262 (responder). 263 3. Add 2 UPnP service requests and 2 Bonjour service requests on the 264 requester. 265 4. Removes 3 of the 4 added requests. 266 5. Initiate p2p service discovery on the requester. Verify that the requester 267 only discovers services corresponds to the remaining request. 268 """ 269 requester, responder = self._setup_wifi_p2p() 270 self._add_p2p_services(responder) 271 272 # Add requests 273 requester.ad.log.info('Adding service requests.') 274 upnp_req_1_id = requester.ad.wifi.wifiP2pAddUpnpServiceRequest() 275 requester.ad.wifi.wifiP2pAddUpnpServiceRequest('ssdp:all') 276 bonjour_req_1_id = requester.ad.wifi.wifiP2pAddBonjourServiceRequest() 277 bonjour_req_2_id = requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 278 None, # instanceName 279 '_ipp._tcp', 280 ) 281 282 # Remove 3 of the 4 added requests except for ssdp:all 283 requester.ad.log.info('Removing service requests.') 284 requester.ad.wifi.wifiP2pRemoveServiceRequest(upnp_req_1_id) 285 requester.ad.wifi.wifiP2pRemoveServiceRequest(bonjour_req_1_id) 286 requester.ad.wifi.wifiP2pRemoveServiceRequest(bonjour_req_2_id) 287 288 # Initialize test listener. 289 p2p_utils.set_upnp_response_listener(requester) 290 p2p_utils.set_dns_sd_response_listeners(requester) 291 # Search service 292 requester.ad.wifi.wifiP2pDiscoverServices() 293 294 # Initiates service discovery and check expected services. 295 p2p_utils.check_discovered_services( 296 requester, 297 responder.p2p_device.device_address, 298 expected_dns_sd_sequence=(), 299 expected_dns_txt_sequence=(), 300 expected_upnp_sequence=constants.ServiceData.ALL_UPNP_SERVICES 301 ) 302 303 def test_serv_req_clear_request(self) -> None: 304 """Checks that API `WifiP2pManager#clearServiceRequests` works well. 305 306 Test Steps: 307 1. Initialize Wi-Fi p2p on both devices. 308 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 309 (responder). 310 3. Add 2 UPnP service requests and 2 Bonjour service requests on the 311 requester. 312 4. Clears all added requests. 313 5. Initiate p2p service discovery on the requester. Verify that the service 314 discovery should fail due to no service request. 315 """ 316 requester, responder = self._setup_wifi_p2p() 317 self._add_p2p_services(responder) 318 319 # Add requests 320 requester.ad.log.info('Adding service requests.') 321 requester.ad.wifi.wifiP2pAddUpnpServiceRequest() 322 requester.ad.wifi.wifiP2pAddUpnpServiceRequest('ssdp:all') 323 requester.ad.wifi.wifiP2pAddBonjourServiceRequest() 324 requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 325 None, # instanceName 326 '_ipp._tcp', 327 ) 328 329 # Clear requests 330 requester.ad.log.info('Clearing all service requests.') 331 requester.ad.wifi.wifiP2pClearServiceRequests() 332 333 # Search services, but NO_SERVICE_REQUESTS is returned. 334 requester.ad.log.info('Initiating service discovery.') 335 expect_error_code = constants.WifiP2pManagerConstants.NO_SERVICE_REQUESTS 336 with asserts.assert_raises_regex( 337 Exception, 338 f'reason_code={str(expect_error_code)}', 339 extras='Service discovery should fail due to no service request.', 340 ): 341 requester.ad.wifi.wifiP2pDiscoverServices() 342 343 def test_serv_req_multi_channel_01(self) -> None: 344 """Searches all UPnP services on channel1 and all Bonjour services on 345 channel2. 346 347 Test Steps: 348 1. Initialize Wi-Fi p2p on both devices. This initializes p2p channel 349 channel1. 350 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 351 (responder). 352 3. Initialize an extra p2p channel channel2 on another device (requester). 353 4. Add a UPnP service request to channel1. 354 5. Add a Bonjour request to channel2. 355 6. Initiate p2p service discovery on channel1. Verify that the requester 356 discovers UPnP services on channel1 and Bonjour services on channel2. 357 """ 358 requester, responder = self._setup_wifi_p2p() 359 self._add_p2p_services(responder) 360 channel1 = requester.channel_ids[0] 361 channel2 = p2p_utils.init_extra_channel(requester) 362 363 requester.ad.log.info('Adding service requests.') 364 # Add UPnP request to the channel 1. 365 requester.ad.wifi.wifiP2pAddUpnpServiceRequest( 366 None, # serviceType 367 channel1, 368 ) 369 # Add UPnP request to channel 2. 370 requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 371 None, # instanceName 372 None, # serviceType 373 channel2, 374 ) 375 376 # Set service listener. 377 requester.ad.log.info('Setting service listeners.') 378 p2p_utils.set_upnp_response_listener(requester, channel1) 379 p2p_utils.set_dns_sd_response_listeners(requester, channel1) 380 p2p_utils.set_upnp_response_listener(requester, channel2) 381 p2p_utils.set_dns_sd_response_listeners(requester, channel2) 382 383 # Discover services 384 requester.ad.log.info('Initiating service discovery.') 385 requester.ad.wifi.wifiP2pDiscoverServices(channel1) 386 responder_address = responder.p2p_device.device_address 387 388 # Check discovered services 389 # Channel1 receive only UPnP service. 390 requester.ad.log.info('Checking services on channel %d.', channel1) 391 p2p_utils.check_discovered_services( 392 requester, 393 responder.p2p_device.device_address, 394 expected_dns_sd_sequence=(), 395 expected_dns_txt_sequence=(), 396 expected_upnp_sequence=constants.ServiceData.ALL_UPNP_SERVICES, 397 channel_id=channel1, 398 ) 399 # Channel2 receive only Bonjour service. 400 requester.ad.log.info('Checking services on channel %d.', channel2) 401 p2p_utils.check_discovered_services( 402 requester, 403 responder.p2p_device.device_address, 404 expected_dns_sd_sequence=constants.ServiceData.ALL_DNS_SD, 405 expected_dns_txt_sequence=constants.ServiceData.ALL_DNS_TXT, 406 expected_upnp_sequence=(), 407 channel_id=channel2, 408 ) 409 410 # Clean up. 411 p2p_utils.reset_p2p_service_state(requester.ad, channel1) 412 p2p_utils.reset_p2p_service_state(requester.ad, channel2) 413 414 def test_serv_req_multi_channel_02(self) -> None: 415 """Searches Bonjour IPP PTR service on channel1 and AFP TXT service on channel2. 416 417 Test Steps: 418 1. Initialize Wi-Fi p2p on both devices. This initializes p2p channel 419 channel1. 420 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 421 (responder). 422 3. Initialize an extra p2p channel channel2 on another device (requester). 423 4. Add a Bonjour IPP PTR request to channel1. 424 5. Add a Bonjour AFP TXT request to channel2. 425 6. Initiate p2p service discovery on channel1. Verify that the requester 426 discovers IPP PTR services on channel1 and AFP TXT services on channel2. 427 """ 428 requester, responder = self._setup_wifi_p2p() 429 self._add_p2p_services(responder) 430 channel1 = requester.channel_ids[0] 431 channel2 = p2p_utils.init_extra_channel(requester) 432 433 # Add Bonjour IPP PRT request to channel1. 434 requester.ad.log.info('Adding service requests.') 435 requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 436 None, # instanceName 437 '_ipp._tcp', 438 channel1, 439 ) 440 441 # Add Bonjour AFP TXT request to channel2. 442 requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 443 'Example', 444 '_afpovertcp._tcp', 445 channel2, 446 ) 447 448 # Initialize listener test objects. 449 requester.ad.log.info('Setting service listeners.') 450 p2p_utils.set_upnp_response_listener(requester, channel1) 451 p2p_utils.set_dns_sd_response_listeners(requester, channel1) 452 p2p_utils.set_upnp_response_listener(requester, channel2) 453 p2p_utils.set_dns_sd_response_listeners(requester, channel2) 454 455 # Discover services 456 requester.ad.log.info('Initiating service discovery.') 457 requester.ad.wifi.wifiP2pDiscoverServices(channel1) 458 responder_address = responder.p2p_device.device_address 459 460 # Check discovered services 461 # Channel1 receive only Bonjour IPP PTR. 462 requester.ad.log.info('Checking services on channel %d.', channel1) 463 p2p_utils.check_discovered_services( 464 requester, 465 responder.p2p_device.device_address, 466 expected_dns_sd_sequence=constants.ServiceData.IPP_DNS_SD, 467 expected_dns_txt_sequence=(), 468 expected_upnp_sequence=(), 469 channel_id=channel1, 470 ) 471 # Channel2 receive only Bonjour AFP TXT. 472 requester.ad.log.info('Checking services on channel %d.', channel2) 473 p2p_utils.check_discovered_services( 474 requester, 475 responder.p2p_device.device_address, 476 expected_dns_sd_sequence=(), 477 expected_dns_txt_sequence=constants.ServiceData.AFP_DNS_TXT, 478 expected_upnp_sequence=(), 479 channel_id=channel2, 480 ) 481 482 # Clean up. 483 p2p_utils.reset_p2p_service_state(requester.ad, channel1) 484 p2p_utils.reset_p2p_service_state(requester.ad, channel2) 485 486 def test_serv_req_multi_channel_03(self) -> None: 487 """Checks that `removeServiceRequest` and `clearServiceRequests` have no 488 effect against another channel. 489 490 Test Steps: 491 1. Initialize Wi-Fi p2p on both devices. This initializes p2p channel 492 channel1. 493 2. Add local services UPnP and Bonjour and initiate peer discovery on one device 494 (responder). 495 3. Initialize an extra p2p channel channel2 on another device (requester). 496 4. Add a Bonjour request to channel1. 497 5. Try to remove the request of channel1 on channel2. This should 498 not have effect. 499 6. Try to clear service requests on channel2. This should not have 500 effect. 501 4. Add a Bonjour request to channel2. 502 5. Initiate p2p service discovery on channel1. Verify that the requester 503 discovers Bonjour services but not UPnP services on channel1. 504 """ 505 requester, responder = self._setup_wifi_p2p() 506 self._add_p2p_services(responder) 507 channel1 = requester.channel_ids[0] 508 channel2 = p2p_utils.init_extra_channel(requester) 509 510 # Add Bonjour request to channel1. 511 requester.ad.log.info('Adding service request to channel %d', channel1) 512 bonjour_req_id = requester.ad.wifi.wifiP2pAddBonjourServiceRequest( 513 None, # instanceName 514 None, # serviceType 515 channel1, 516 ) 517 requester.ad.log.info('Added request %d', bonjour_req_id) 518 519 # Try to remove the Bonjour request of channel1 on channel2. 520 # However, it should silently succeed but have no effect 521 requester.ad.log.info('Removing the request %d from channel %d', bonjour_req_id, channel1) 522 requester.ad.wifi.wifiP2pRemoveServiceRequest(bonjour_req_id, channel2) 523 524 # Clear the all requests on channel2. 525 # However, it should silently succeed but have no effect 526 requester.ad.log.info('Clearing service requests on channel %d', channel2) 527 requester.ad.wifi.wifiP2pClearServiceRequests(channel2) 528 529 # Initialize service listeners. 530 requester.ad.log.info('Setting service listeners on both channels.') 531 p2p_utils.set_upnp_response_listener(requester, channel1) 532 p2p_utils.set_dns_sd_response_listeners(requester, channel1) 533 p2p_utils.set_upnp_response_listener(requester, channel2) 534 p2p_utils.set_dns_sd_response_listeners(requester, channel2) 535 536 # Discover services 537 requester.ad.log.info('Initiating service discovery.') 538 requester.ad.wifi.wifiP2pDiscoverServices(channel1) 539 responder_address = responder.p2p_device.device_address 540 541 # Check that Bonjour response can be received on channel1 542 requester.ad.log.info('Checking Bonjour services on channel %d.', channel1) 543 p2p_utils.check_discovered_services( 544 requester, 545 responder.p2p_device.device_address, 546 expected_dns_sd_sequence=constants.ServiceData.ALL_DNS_SD, 547 expected_dns_txt_sequence=constants.ServiceData.AFP_DNS_TXT, 548 expected_upnp_sequence=(), 549 channel_id=channel1, 550 ) 551 552 # Clean up. 553 p2p_utils.reset_p2p_service_state(requester.ad, channel1) 554 p2p_utils.reset_p2p_service_state(requester.ad, channel2) 555 556 def _search_p2p_services( 557 self, 558 responder: p2p_utils.DeviceState, 559 requester: p2p_utils.DeviceState, 560 expected_dns_sd_sequence: Sequence[Sequence[str, dict[str, str]]], 561 expected_dns_txt_sequence: Sequence[Sequence[str, str]], 562 expected_upnp_sequence: Sequence[str], 563 ) -> None: 564 """Initiate service discovery and assert expected p2p services are discovered. 565 566 Args: 567 responder: The responder device state. 568 requester: The requester device state. 569 expected_dns_sd_sequence: Expected DNS-SD responses. 570 expected_dns_txt_sequence: Expected DNS TXT records. 571 expected_upnp_sequence: Expected UPNP services. 572 """ 573 requester.ad.log.info('Initiating service discovery.') 574 p2p_utils.set_upnp_response_listener(requester) 575 p2p_utils.set_dns_sd_response_listeners(requester) 576 requester.ad.wifi.wifiP2pDiscoverServices() 577 578 requester.ad.log.info('Checking discovered services.') 579 p2p_utils.check_discovered_services( 580 requester, 581 responder.p2p_device.device_address, 582 expected_dns_sd_sequence=expected_dns_sd_sequence, 583 expected_dns_txt_sequence=expected_dns_txt_sequence, 584 expected_upnp_sequence=expected_upnp_sequence, 585 ) 586 587 def _add_p2p_services(self, responder: p2p_utils.DeviceState): 588 """Sets up P2P services on the responder device. 589 590 This method adds local UPNP and Bonjour services to the responder device and 591 initiates peer discovery. 592 """ 593 responder.ad.log.info('Setting up p2p local services UpnP and Bonjour.') 594 p2p_utils.add_upnp_local_service( 595 responder, constants.ServiceData.DEFAULT_UPNP_SERVICE_CONF 596 ) 597 p2p_utils.add_bonjour_local_service( 598 responder, constants.ServiceData.DEFAULT_IPP_SERVICE_CONF 599 ) 600 p2p_utils.add_bonjour_local_service( 601 responder, constants.ServiceData.DEFAULT_AFP_SERVICE_CONF 602 ) 603 responder.ad.wifi.wifiP2pDiscoverPeers() 604 605 def _setup_wifi_p2p(self): 606 logging.info('Initializing Wi-Fi p2p.') 607 responder = p2p_utils.setup_wifi_p2p(self.responder_ad) 608 requester = p2p_utils.setup_wifi_p2p(self.requester_ad) 609 return requester, responder 610 611 def _teardown_wifi_p2p(self, ad: android_device.AndroidDevice): 612 try: 613 p2p_utils.teardown_wifi_p2p(ad) 614 finally: 615 ad.services.create_output_excerpts_all(self.current_test_info) 616 617 def teardown_test(self) -> None: 618 utils.concurrent_exec( 619 self._teardown_wifi_p2p, 620 param_list=[[ad] for ad in self.ads], 621 raise_on_exception=False, 622 ) 623 624 def on_fail(self, record: records.TestResult) -> None: 625 logging.info('Collecting bugreports...') 626 android_device.take_bug_reports( 627 self.ads, destination=self.current_test_info.output_path 628 ) 629 630 631if __name__ == '__main__': 632 test_runner.main() 633