Skip to content

Commit f0312ba

Browse files
author
yennj12
committed
add
1 parent 1ecd110 commit f0312ba

File tree

2 files changed

+197
-0
lines changed

2 files changed

+197
-0
lines changed

‎mobly/cdm_transport_test.py

+31
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
# https://cs.android.com/android/platform/superproject/main/+/main:frameworks/base/tests/CompanionDeviceMultiDeviceTests/host/cdm_transport_test.py?hl=zh-tw
2+
3+
#!/usr/bin/env python3
4+
# Lint as: python3
5+
"""
6+
Test E2E CDM functions on mobly.
7+
"""
8+
9+
import cdm_base_test
10+
import sys
11+
12+
from mobly import asserts
13+
from mobly import test_runner
14+
15+
CDM_SNIPPET_PACKAGE = 'android.companion.multidevices'
16+
17+
18+
class TransportTestClass(cdm_base_test.BaseTestClass):
19+
20+
def test_permissions_sync(self):
21+
"""This tests permissions sync from one device to another."""
22+
23+
# associate and attach transports
24+
self.attach_transports()
25+
26+
# start permissions sync
27+
self.sender.cdm.startPermissionsSync(self.receiver_id)
28+
29+
30+
if __name__ == '__main__':
31+
test_runner.main()

‎mobly/wifi_direct_test.py

+166
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
# https://cs.android.com/android/platform/superproject/main/+/main:packages/modules/Wifi/tests/hostsidetests/multidevices/test/direct/wifi_direct_test.py;l=20?hl=zh-tw
2+
3+
4+
# Copyright (C) 2023 The Android Open Source Project
5+
#
6+
# Licensed under the Apache License, Version 2.0 (the "License");
7+
# you may not use this file except in compliance with the License.
8+
# You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing, software
13+
# distributed under the License is distributed on an "AS IS" BASIS,
14+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
# See the License for the specific language governing permissions and
16+
# limitations under the License.
17+
#
18+
# Licensed under the Apache License, Version 2.0 (the "License");
19+
# you may not use this file except in compliance with the License.
20+
# You may obtain a copy of the License at
21+
#
22+
# http://www.apache.org/licenses/LICENSE-2.0
23+
#
24+
# Unless required by applicable law or agreed to in writing, software
25+
# distributed under the License is distributed on an "AS IS" BASIS,
26+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
27+
# See the License for the specific language governing permissions and
28+
# limitations under the License.
29+
30+
# Lint as: python3
31+
32+
from collections.abc import Sequence
33+
import datetime
34+
35+
from mobly import asserts
36+
from mobly import base_test
37+
from mobly import test_runner
38+
from mobly import utils
39+
from mobly.controllers import android_device
40+
from mobly.controllers.android_device_lib import callback_handler_v2
41+
from mobly.snippet import errors
42+
43+
from direct import constants
44+
45+
_DEFAULT_TIMEOUT = datetime.timedelta(seconds=30)
46+
47+
_WIFI_DIRECT_SNIPPET_KEY = 'wifi_direct_mobly_snippet'
48+
49+
50+
def _init_wifi_p2p(
51+
ad: android_device.AndroidDevice,
52+
) -> callback_handler_v2.CallbackHandlerV2:
53+
"""Registers the application with the Wi-Fi framework."""
54+
state_handler = ad.wifi.p2pInitialize()
55+
init_event = state_handler.waitAndGet(
56+
event_name=constants.WIFI_P2P_STATE_CHANGED_ACTION,
57+
timeout=_DEFAULT_TIMEOUT.total_seconds(),
58+
)
59+
state = constants.ExtraWifiState(init_event.data[constants.EXTRA_WIFI_STATE])
60+
asserts.assert_equal(
61+
state,
62+
constants.ExtraWifiState.WIFI_P2P_STATE_ENABLED,
63+
f'Failed to initialize Wi-Fi P2P, state: {state}',
64+
)
65+
return state_handler
66+
67+
68+
class WifiDirectTest(base_test.BaseTestClass):
69+
"""Tests Wi-Fi Direct between 2 Android devices."""
70+
71+
ads: Sequence[android_device.AndroidDevice]
72+
73+
def _setup_device(self, ad: android_device.AndroidDevice) -> None:
74+
tag = 'files' if 'files' in self.user_params else 'mh_files'
75+
asserts.assert_in(
76+
_WIFI_DIRECT_SNIPPET_KEY,
77+
self.user_params[tag],
78+
'Wi-Fi direct snippet not found',
79+
)
80+
ad.adb.install(
81+
['-d', '-g', '-r', self.user_params[tag][_WIFI_DIRECT_SNIPPET_KEY][0]]
82+
)
83+
ad.load_snippet('wifi', constants.WIFI_DIRECT_SNIPPET_PACKAGE_NAME)
84+
85+
def setup_class(self) -> None:
86+
super().setup_class()
87+
self.ads = self.register_controller(android_device, min_number=2)
88+
utils.concurrent_exec(
89+
self._setup_device,
90+
param_list=[[ad] for ad in self.ads],
91+
raise_on_exception=True,
92+
)
93+
94+
def test_wifi_direct_connect(self) -> None:
95+
group_owner, client, *_ = self.ads
96+
group_owner.debug_tag = 'Group Owner'
97+
client.debug_tag = 'Client'
98+
config = constants.WifiP2pConfig(
99+
network_name='DIRECT-BeTo', passphrase='testtest'
100+
)
101+
102+
owner_state_handler = _init_wifi_p2p(group_owner)
103+
104+
owner_action_handler = group_owner.wifi.p2pCreateGroup(config.to_dict())
105+
try:
106+
owner_action_handler.waitAndGet(
107+
event_name=constants.ACTION_LISTENER_ON_SUCCESS,
108+
timeout=_DEFAULT_TIMEOUT.total_seconds(),
109+
)
110+
except errors.CallbackHandlerTimeoutError:
111+
failure_event = owner_action_handler.waitAndGet(
112+
constants.ACTION_LISTENER_ON_FAILURE
113+
)
114+
failure_reason = constants.ActionListenerOnFailure(
115+
failure_event.data[constants.ACTION_LISTENER_FAILURE_REASON]
116+
)
117+
asserts.fail(f'Failed to create a group, reason: {failure_reason.name}')
118+
owner_connected_event = owner_state_handler.waitAndGet(
119+
event_name=constants.WIFI_P2P_CREATING_GROUP,
120+
timeout=_DEFAULT_TIMEOUT.total_seconds(),
121+
)
122+
group_owner_wifi_group = owner_connected_event.data[
123+
constants.EXTRA_WIFI_P2P_GROUP
124+
]
125+
group_owner.log.info(f'Created a group: {group_owner_wifi_group}')
126+
127+
client_state_handler = _init_wifi_p2p(client)
128+
129+
client_action_callback = client.wifi.p2pConnect(config.to_dict())
130+
try:
131+
client_action_callback.waitAndGet(
132+
event_name=constants.ACTION_LISTENER_ON_SUCCESS,
133+
timeout=_DEFAULT_TIMEOUT.total_seconds(),
134+
)
135+
except errors.CallbackHandlerTimeoutError:
136+
failure_event = client_action_callback.waitAndGet(
137+
constants.ACTION_LISTENER_ON_FAILURE
138+
)
139+
failure_reason = constants.ActionListenerOnFailure(
140+
failure_event.data[constants.ACTION_LISTENER_FAILURE_REASON]
141+
)
142+
asserts.fail(
143+
f'Failed to connect to a group, reason: {failure_reason.name}'
144+
)
145+
client_connected_event = client_state_handler.waitAndGet(
146+
event_name=constants.WIFI_P2P_CREATING_GROUP,
147+
timeout=_DEFAULT_TIMEOUT.total_seconds(),
148+
)
149+
client_wifi_group = client_connected_event.data[
150+
constants.EXTRA_WIFI_P2P_GROUP
151+
]
152+
group_owner.log.info(f'Created a group: {client_wifi_group}')
153+
154+
group_owner.wifi.p2pClose()
155+
client.wifi.p2pClose()
156+
157+
def teardown_test(self) -> None:
158+
utils.concurrent_exec(
159+
lambda d: d.services.create_output_excerpts_all(self.current_test_info),
160+
param_list=[[ad] for ad in self.ads],
161+
raise_on_exception=True,
162+
)
163+
164+
165+
if __name__ == '__main__':
166+
test_runner.main()

0 commit comments

Comments
 (0)