• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3#   Copyright 2016 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import logging
18import time
19import unittest
20
21import mock
22
23from acts import utils
24from acts import signals
25from acts.controllers.adb import AdbError
26
27PROVISIONED_STATE_GOOD = 1
28
29
30class ByPassSetupWizardTests(unittest.TestCase):
31    """This test class for unit testing acts.utils.bypass_setup_wizard."""
32
33    def test_start_standing_subproc(self):
34        with self.assertRaisesRegex(utils.ActsUtilsError,
35                                    'Process .* has terminated'):
36            utils.start_standing_subprocess('sleep 0', check_health_delay=0.1)
37
38    def test_stop_standing_subproc(self):
39        p = utils.start_standing_subprocess('sleep 0')
40        time.sleep(0.1)
41        with self.assertRaisesRegex(utils.ActsUtilsError,
42                                    'Process .* has terminated'):
43            utils.stop_standing_subprocess(p)
44
45    @mock.patch('time.sleep')
46    def test_bypass_setup_wizard_no_complications(self, _):
47        ad = mock.Mock()
48        ad.adb.shell.side_effect = [
49            # Return value for SetupWizardExitActivity
50            BypassSetupWizardReturn.NO_COMPLICATIONS,
51            # Return value for device_provisioned
52            PROVISIONED_STATE_GOOD,
53        ]
54        ad.adb.return_state = BypassSetupWizardReturn.NO_COMPLICATIONS
55        self.assertTrue(utils.bypass_setup_wizard(ad))
56        self.assertFalse(
57            ad.adb.root_adb.called,
58            'The root command should not be called if there are no '
59            'complications.')
60
61    @mock.patch('time.sleep')
62    def test_bypass_setup_wizard_unrecognized_error(self, _):
63        ad = mock.Mock()
64        ad.adb.shell.side_effect = [
65            # Return value for SetupWizardExitActivity
66            BypassSetupWizardReturn.UNRECOGNIZED_ERR,
67            # Return value for device_provisioned
68            PROVISIONED_STATE_GOOD,
69        ]
70        with self.assertRaises(AdbError):
71            utils.bypass_setup_wizard(ad)
72        self.assertFalse(
73            ad.adb.root_adb.called,
74            'The root command should not be called if we do not have a '
75            'codepath for recovering from the failure.')
76
77    @mock.patch('time.sleep')
78    def test_bypass_setup_wizard_need_root_access(self, _):
79        ad = mock.Mock()
80        ad.adb.shell.side_effect = [
81            # Return value for SetupWizardExitActivity
82            BypassSetupWizardReturn.ROOT_ADB_NO_COMP,
83            # Return value for rooting the device
84            BypassSetupWizardReturn.NO_COMPLICATIONS,
85            # Return value for device_provisioned
86            PROVISIONED_STATE_GOOD
87        ]
88
89        utils.bypass_setup_wizard(ad)
90
91        self.assertTrue(
92            ad.adb.root_adb_called,
93            'The command required root access, but the device was never '
94            'rooted.')
95
96    @mock.patch('time.sleep')
97    def test_bypass_setup_wizard_need_root_already_skipped(self, _):
98        ad = mock.Mock()
99        ad.adb.shell.side_effect = [
100            # Return value for SetupWizardExitActivity
101            BypassSetupWizardReturn.ROOT_ADB_SKIPPED,
102            # Return value for SetupWizardExitActivity after root
103            BypassSetupWizardReturn.ALREADY_BYPASSED,
104            # Return value for device_provisioned
105            PROVISIONED_STATE_GOOD
106        ]
107        self.assertTrue(utils.bypass_setup_wizard(ad))
108        self.assertTrue(ad.adb.root_adb_called)
109
110    @mock.patch('time.sleep')
111    def test_bypass_setup_wizard_root_access_still_fails(self, _):
112        ad = mock.Mock()
113        ad.adb.shell.side_effect = [
114            # Return value for SetupWizardExitActivity
115            BypassSetupWizardReturn.ROOT_ADB_FAILS,
116            # Return value for SetupWizardExitActivity after root
117            BypassSetupWizardReturn.UNRECOGNIZED_ERR,
118            # Return value for device_provisioned
119            PROVISIONED_STATE_GOOD
120        ]
121
122        with self.assertRaises(AdbError):
123            utils.bypass_setup_wizard(ad)
124        self.assertTrue(ad.adb.root_adb_called)
125
126
127class BypassSetupWizardReturn:
128    # No complications. Bypass works the first time without issues.
129    NO_COMPLICATIONS = ('Starting: Intent { cmp=com.google.android.setupwizard/'
130                        '.SetupWizardExitActivity }')
131
132    # Fail with doesn't need to be skipped/was skipped already.
133    ALREADY_BYPASSED = AdbError('', 'ADB_CMD_OUTPUT:0', 'Error type 3\n'
134                                                        'Error: Activity class',
135                                1)
136    # Fail with different error.
137    UNRECOGNIZED_ERR = AdbError('', 'ADB_CMD_OUTPUT:0', 'Error type 4\n'
138                                                        'Error: Activity class',
139                                0)
140    # Fail, get root access, then no complications arise.
141    ROOT_ADB_NO_COMP = AdbError('', 'ADB_CMD_OUTPUT:255',
142                                'Security exception: Permission Denial: '
143                                'starting Intent { flg=0x10000000 '
144                                'cmp=com.google.android.setupwizard/'
145                                '.SetupWizardExitActivity } from null '
146                                '(pid=5045, uid=2000) not exported from uid '
147                                '10000', 0)
148    # Even with root access, the bypass setup wizard doesn't need to be skipped.
149    ROOT_ADB_SKIPPED = AdbError('', 'ADB_CMD_OUTPUT:255',
150                                'Security exception: Permission Denial: '
151                                'starting Intent { flg=0x10000000 '
152                                'cmp=com.google.android.setupwizard/'
153                                '.SetupWizardExitActivity } from null '
154                                '(pid=5045, uid=2000) not exported from '
155                                'uid 10000', 0)
156    # Even with root access, the bypass setup wizard fails
157    ROOT_ADB_FAILS = AdbError(
158        '', 'ADB_CMD_OUTPUT:255',
159        'Security exception: Permission Denial: starting Intent { '
160        'flg=0x10000000 cmp=com.google.android.setupwizard/'
161        '.SetupWizardExitActivity } from null (pid=5045, uid=2000) not '
162        'exported from uid 10000', 0)
163
164
165class ConcurrentActionsTest(unittest.TestCase):
166    """Tests acts.utils.run_concurrent_actions and related functions."""
167
168    @staticmethod
169    def function_returns_passed_in_arg(arg):
170        return arg
171
172    @staticmethod
173    def function_raises_passed_in_exception_type(exception_type):
174        raise exception_type
175
176    def test_run_concurrent_actions_no_raise_returns_proper_return_values(self):
177        """Tests run_concurrent_actions_no_raise returns in the correct order.
178
179        Each function passed into run_concurrent_actions_no_raise returns the
180        values returned from each individual callable in the order passed in.
181        """
182        ret_values = utils.run_concurrent_actions_no_raise(
183            lambda: self.function_returns_passed_in_arg('ARG1'),
184            lambda: self.function_returns_passed_in_arg('ARG2'),
185            lambda: self.function_returns_passed_in_arg('ARG3')
186        )
187
188        self.assertEqual(len(ret_values), 3)
189        self.assertEqual(ret_values[0], 'ARG1')
190        self.assertEqual(ret_values[1], 'ARG2')
191        self.assertEqual(ret_values[2], 'ARG3')
192
193    def test_run_concurrent_actions_no_raise_returns_raised_exceptions(self):
194        """Tests run_concurrent_actions_no_raise returns raised exceptions.
195
196        Instead of allowing raised exceptions to be raised in the main thread,
197        this function should capture the exception and return them in the slot
198        the return value should have been returned in.
199        """
200        ret_values = utils.run_concurrent_actions_no_raise(
201            lambda: self.function_raises_passed_in_exception_type(IndexError),
202            lambda: self.function_raises_passed_in_exception_type(KeyError)
203        )
204
205        self.assertEqual(len(ret_values), 2)
206        self.assertEqual(ret_values[0].__class__, IndexError)
207        self.assertEqual(ret_values[1].__class__, KeyError)
208
209    def test_run_concurrent_actions_returns_proper_return_values(self):
210        """Tests run_concurrent_actions returns in the correct order.
211
212        Each function passed into run_concurrent_actions returns the values
213        returned from each individual callable in the order passed in.
214        """
215
216        ret_values = utils.run_concurrent_actions(
217            lambda: self.function_returns_passed_in_arg('ARG1'),
218            lambda: self.function_returns_passed_in_arg('ARG2'),
219            lambda: self.function_returns_passed_in_arg('ARG3')
220        )
221
222        self.assertEqual(len(ret_values), 3)
223        self.assertEqual(ret_values[0], 'ARG1')
224        self.assertEqual(ret_values[1], 'ARG2')
225        self.assertEqual(ret_values[2], 'ARG3')
226
227    def test_run_concurrent_actions_raises_exceptions(self):
228        """Tests run_concurrent_actions raises exceptions from given actions."""
229        with self.assertRaises(KeyError):
230            utils.run_concurrent_actions(
231                lambda: self.function_returns_passed_in_arg('ARG1'),
232                lambda: self.function_raises_passed_in_exception_type(KeyError)
233            )
234
235    def test_test_concurrent_actions_raises_non_test_failure(self):
236        """Tests test_concurrent_actions raises the given exception."""
237        with self.assertRaises(KeyError):
238            utils.test_concurrent_actions(
239                lambda: self.function_raises_passed_in_exception_type(KeyError),
240                failure_exceptions=signals.TestFailure
241            )
242
243    def test_test_concurrent_actions_raises_test_failure(self):
244        """Tests test_concurrent_actions raises the given exception."""
245        with self.assertRaises(signals.TestFailure):
246            utils.test_concurrent_actions(
247                lambda: self.function_raises_passed_in_exception_type(KeyError),
248                failure_exceptions=KeyError
249            )
250
251
252class SuppressLogOutputTest(unittest.TestCase):
253    """Tests SuppressLogOutput"""
254
255    def test_suppress_log_output(self):
256        """Tests that the SuppressLogOutput context manager removes handlers
257        of the specified levels upon entry and re-adds handlers upon exit.
258        """
259        handlers = [logging.NullHandler(level=lvl) for lvl in
260                    (logging.DEBUG, logging.INFO, logging.ERROR)]
261        log = logging.getLogger('test_log')
262        for handler in handlers:
263            log.addHandler(handler)
264        with utils.SuppressLogOutput(log, [logging.INFO, logging.ERROR]):
265            self.assertTrue(
266                any(handler.level == logging.DEBUG for handler in log.handlers))
267            self.assertFalse(
268                any(handler.level in (logging.INFO, logging.ERROR)
269                    for handler in log.handlers))
270        self.assertCountEqual(handlers, log.handlers)
271
272
273class IpAddressUtilTest(unittest.TestCase):
274
275    def test_positive_ipv4_normal_address(self):
276        ip_address = "192.168.1.123"
277        self.assertTrue(utils.is_valid_ipv4_address(ip_address))
278
279    def test_positive_ipv4_any_address(self):
280        ip_address = "0.0.0.0"
281        self.assertTrue(utils.is_valid_ipv4_address(ip_address))
282
283    def test_positive_ipv4_broadcast(self):
284        ip_address = "255.255.255.0"
285        self.assertTrue(utils.is_valid_ipv4_address(ip_address))
286
287    def test_negative_ipv4_with_ipv6_address(self):
288        ip_address = "fe80::f693:9fff:fef4:1ac"
289        self.assertFalse(utils.is_valid_ipv4_address(ip_address))
290
291    def test_negative_ipv4_with_invalid_string(self):
292        ip_address = "fdsafdsafdsafdsf"
293        self.assertFalse(utils.is_valid_ipv4_address(ip_address))
294
295    def test_negative_ipv4_with_invalid_number(self):
296        ip_address = "192.168.500.123"
297        self.assertFalse(utils.is_valid_ipv4_address(ip_address))
298
299    def test_positive_ipv6(self):
300        ip_address = 'fe80::f693:9fff:fef4:1ac'
301        self.assertTrue(utils.is_valid_ipv6_address(ip_address))
302
303    def test_positive_ipv6_link_local(self):
304        ip_address = 'fe80::'
305        self.assertTrue(utils.is_valid_ipv6_address(ip_address))
306
307    def test_negative_ipv6_with_ipv4_address(self):
308        ip_address = '192.168.1.123'
309        self.assertFalse(utils.is_valid_ipv6_address(ip_address))
310
311    def test_negative_ipv6_invalid_characters(self):
312        ip_address = 'fe80:jkyr:f693:9fff:fef4:1ac'
313        self.assertFalse(utils.is_valid_ipv6_address(ip_address))
314
315    def test_negative_ipv6_invalid_string(self):
316        ip_address = 'fdsafdsafdsafdsf'
317        self.assertFalse(utils.is_valid_ipv6_address(ip_address))
318
319
320if __name__ == '__main__':
321    unittest.main()
322