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