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# Lint as: python3
16"""CTS-V Wifi test reimplemented in Mobly."""
17import logging
18import sys
19
20logging.basicConfig(filename="/tmp/wifi_aware_test_log.txt", level=logging.INFO)
21
22from mobly import asserts
23from mobly import base_test
24from mobly import test_runner
25from mobly import utils
26from mobly.controllers import android_device
27
28WIFI_AWARE_SNIPPET_PATH = 'wifi_aware_snippet'
29
30WIFI_AWARE_SNIPPET_PACKAGE = 'com.google.snippet.wifi.aware'
31
32TEST_MESSAGE = 'test message!'
33
34MESSAGE_ID = 1234
35
36def setup_aware_pairing(device, is_publish, with_password, accept):
37    if is_publish:
38        device.wifi_aware_snippet.respondToPairingSetup(with_password, accept)
39    else:
40        device.wifi_aware_snippet.initiatePairingSetup(with_password, accept)
41
42class WifiAwareTest(base_test.BaseTestClass):
43
44    def setup_class(self):
45        # Declare that two Android devices are needed.
46        self.publisher, self.subscriber = self.register_controller(
47            android_device, min_number=2)
48
49        def setup_device(device):
50            device.load_snippet('wifi_aware_snippet', WIFI_AWARE_SNIPPET_PACKAGE)
51            # Expect wifi_aware apk to be installed as it is configured to install
52            # with the module configuration AndroidTest.xml on both devices.
53            device.adb.shell([
54                'pm', 'grant', WIFI_AWARE_SNIPPET_PACKAGE,
55                'android.permission.ACCESS_FINE_LOCATION'
56            ])
57
58        # Sets up devices in parallel to save time.
59        utils.concurrent_exec(
60            setup_device, ((self.publisher,), (self.subscriber,)),
61            max_workers=2,
62            raise_on_exception=True)
63
64    def on_fail(self, record):
65        logging.info('Collecting bugreports...')
66        android_device.take_bug_reports(
67            ads=[self.publisher, self.subscriber],
68            test_name=record.test_name,
69            begin_time=record.begin_time,
70            destination=self.current_test_info.output_path
71        )
72
73    def test_discovery_base_test_case(self):
74        is_unsolicited = True
75        is_ranging_required = False
76        self.subscriber.wifi_aware_snippet.attach()
77        self.publisher.wifi_aware_snippet.attach()
78
79        self.publisher.wifi_aware_snippet.publish(is_unsolicited, is_ranging_required, False)
80        self.subscriber.wifi_aware_snippet.subscribe(is_unsolicited, is_ranging_required, False)
81
82        self.subscriber.wifi_aware_snippet.sendMessage(MESSAGE_ID, TEST_MESSAGE)
83        received_message = self.publisher.wifi_aware_snippet.receiveMessage()
84        asserts.assert_equal(
85            received_message, TEST_MESSAGE,
86            'Message received by publisher does not match the message sent by subscriber.'
87        )
88
89    def test_aware_pairing_accept_test_case(self):
90        is_pairing_supported = self.publisher.wifi_aware_snippet.checkIfPairingSupported() and \
91                               self.subscriber.wifi_aware_snippet.checkIfPairingSupported()
92        asserts.skip_if(not is_pairing_supported,
93                        "Aware pairing test skip as feature is not supported")
94        is_unsolicited = True
95        is_ranging_required = False
96        self.subscriber.wifi_aware_snippet.attach()
97        self.publisher.wifi_aware_snippet.attach()
98
99        self.publisher.wifi_aware_snippet.publish(is_unsolicited, is_ranging_required, True)
100        self.subscriber.wifi_aware_snippet.subscribe(is_unsolicited, is_ranging_required, True)
101        utils.concurrent_exec(
102            setup_aware_pairing, ((self.publisher, True, True, True),
103                                  (self.subscriber, False, True, True)),
104            max_workers=2,
105            raise_on_exception=True)
106
107    def test_aware_pairing_reject_test_case(self):
108        is_pairing_supported = self.publisher.wifi_aware_snippet.checkIfPairingSupported() and \
109                               self.subscriber.wifi_aware_snippet.checkIfPairingSupported()
110        asserts.skip_if(not is_pairing_supported,
111                        "Aware pairing test skip as feature is not supported")
112        is_unsolicited = True
113        is_ranging_required = False
114        self.subscriber.wifi_aware_snippet.attach()
115        self.publisher.wifi_aware_snippet.attach()
116
117        self.publisher.wifi_aware_snippet.publish(is_unsolicited, is_ranging_required, True)
118        self.subscriber.wifi_aware_snippet.subscribe(is_unsolicited, is_ranging_required, True)
119        utils.concurrent_exec(
120            setup_aware_pairing, ((self.publisher, True, True, False),
121                                  (self.subscriber, False, True, False)),
122            max_workers=2,
123            raise_on_exception=True)
124
125if __name__ == '__main__':
126    # Take test args
127    if '--' in sys.argv:
128        index = sys.argv.index('--')
129        sys.argv = sys.argv[:1] + sys.argv[index + 1:]
130
131    test_runner.main()
132