xref: /aosp_15_r20/external/autotest/server/cros/tradefed/tradefed_test_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7) 
1# Lint as: python2, python3
2# Copyright 2022 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import unittest
7import tempfile
8import shutil
9import stat
10
11from unittest.mock import Mock, ANY, patch
12from autotest_lib.client.common_lib import error
13from autotest_lib.server.cros.tradefed import tradefed_test
14
15
16class TradefedTestTest(unittest.TestCase):
17    """Tests for TradefedTest class."""
18
19    def setUp(self):
20        self._mockjob_tmpdirs = []
21        self._bindir = tempfile.mkdtemp()
22        self._outputdir = tempfile.mkdtemp()
23        self.mock_adb = Mock()
24        self.tradefed = tradefed_test.TradefedTest(self.create_mock_job(),
25                                                   self._bindir,
26                                                   self._outputdir,
27                                                   adb=self.mock_adb)
28
29    def tearDown(self):
30        shutil.rmtree(self._bindir)
31        shutil.rmtree(self._outputdir)
32        for tmpdir in self._mockjob_tmpdirs:
33            shutil.rmtree(tmpdir)
34
35    def create_mock_job(self):
36        """Creates a mock necessary for constructing tradefed_test instance."""
37        mock_job = Mock()
38        mock_job.pkgmgr = None
39        mock_job.autodir = None
40        mock_job.tmpdir = tempfile.mkdtemp()
41        self._mockjob_tmpdirs.append(mock_job.tmpdir)
42        return mock_job
43
44    # Verify that try_adb_connect fails when run_adb_cmd fails.
45    @patch('autotest_lib.server.cros.tradefed.adb.get_adb_target')
46    def test_try_adb_connect_run_adb_fail(self, mock_get_adb_target):
47        mock_run_adb_cmd = self.mock_adb.run
48
49        # Exit status is set to non-0 to exit _try_adb_connect() early.
50        mock_run_adb_cmd.return_value.exit_status = 1
51        mock_get_adb_target.return_value = '123.76.0.29:3467'
52
53        self.assertFalse(self.tradefed._try_adb_connect(Mock()))
54        mock_run_adb_cmd.assert_called_with(ANY,
55                                            args=('connect',
56                                                  '123.76.0.29:3467'),
57                                            verbose=ANY,
58                                            env=ANY,
59                                            ignore_status=ANY,
60                                            timeout=ANY)
61
62    # Verify that _run_tradefed_with_timeout works.
63    @patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._run')
64    @patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._tradefed_cmd_path'
65           )
66    @patch('autotest_lib.server.cros.tradefed.tradefed_utils.adb_keepalive')
67    @patch('autotest_lib.server.cros.tradefed.adb.get_adb_targets')
68    def test_run_tradefed_with_timeout(self, mock_get_adb_targets, _,
69                                       mock_tradefed_cmd_path, mock_run):
70        self.tradefed._install_paths = '/any/install/path'
71
72        mock_host1 = Mock()
73        mock_host2 = Mock()
74        self.tradefed._hosts = [mock_host1, mock_host2]
75
76        mock_get_adb_targets.return_value = ['host1:4321', 'host2:22']
77
78        mock_tradefed_cmd_path.return_value = '/any/path'
79
80        self.tradefed._run_tradefed_with_timeout(['command'], 1234)
81        mock_get_adb_targets.assert_called_with(self.tradefed._hosts)
82
83    def test_kill_adb_server(self):
84        mock_run = self.mock_adb.run
85        self.tradefed._kill_adb_server()
86        mock_run.assert_called_with(None,
87                                    args=('kill-server', ),
88                                    timeout=ANY,
89                                    verbose=ANY)
90
91    def test_verify_arc_hosts_single_host(self):
92        mock_run = self.mock_adb.run
93        mock_host = Mock()
94        self.tradefed._hosts = [mock_host]
95
96        self.tradefed._verify_arc_hosts()
97
98        mock_run.assert_called_with(mock_host,
99                                    args=('shell', 'getprop',
100                                          'ro.build.fingerprint'))
101
102    # Verify that multiple hosts with differet fingerprints fail.
103    def test_verify_arc_hosts_different_fingerprints(self):
104        mock_run = self.mock_adb.run
105        mock_host1 = Mock()
106        mock_host2 = Mock()
107        self.tradefed._hosts = [mock_host1, mock_host2]
108
109        side_effects = [Mock(), Mock()]
110        side_effects[0].stdout = 'fingerprint1'
111        side_effects[1].stdout = 'fingerprint2'
112        mock_run.side_effect = side_effects
113
114        self.assertRaises(error.TestFail, self.tradefed._verify_arc_hosts)
115
116        mock_run.assert_any_call(mock_host1,
117                                 args=('shell', 'getprop',
118                                       'ro.build.fingerprint'))
119        mock_run.assert_any_call(mock_host2,
120                                 args=('shell', 'getprop',
121                                       'ro.build.fingerprint'))
122
123    # Verify that wait for arc boot uses polling with adb.
124    @patch('autotest_lib.server.utils.poll_for_condition')
125    def test_wait_for_arc_boot(self, mock_poll_for_condition):
126        mock_run = self.mock_adb.run
127
128        # stdout just has to be something that evaluates to True.
129        mock_run.return_value.stdout = 'anything'
130
131        mock_host = Mock()
132        self.tradefed._wait_for_arc_boot(mock_host)
133
134        self.assertEqual(mock_run.call_count, 0)
135
136        # Verify that the condition function uses the expected adb command.
137        self.assertEqual(mock_poll_for_condition.call_count, 1)
138        args = mock_poll_for_condition.call_args[0]
139        condition_func = args[0]
140        self.assertTrue(condition_func())
141
142        mock_run.assert_called_with(mock_host,
143                                    args=('shell', 'pgrep', '-f',
144                                          'org.chromium.arc.intent_helper'),
145                                    ignore_status=True)
146
147    def test_disable_adb_install_dialog_android_version_over_29(self):
148        mock_run = self.mock_adb.run
149        mock_run.return_value.stdout = 'disabled'
150
151        self.tradefed._android_version = 30
152        mock_host = Mock()
153        self.tradefed._disable_adb_install_dialog(mock_host)
154
155        mock_run.assert_called_with(mock_host,
156                                    args=('shell', 'settings', 'put', 'global',
157                                          'verifier_verify_adb_installs', '0'),
158                                    verbose=ANY)
159
160    def test_disable_adb_install_dialog_android_version_under_29(self):
161        mock_run = self.mock_adb.run
162
163        mock_run.return_value.stdout = 'disabled'
164
165        self.tradefed._android_version = 28
166        mock_host = Mock()
167        self.tradefed._disable_adb_install_dialog(mock_host)
168
169        mock_run.assert_called_with(mock_host,
170                                    args=('shell', 'settings', 'put', 'global',
171                                          'verifier_verify_adb_installs', '0'),
172                                    verbose=ANY)
173
174        mock_host.run.assert_called_with(
175                'android-sh -c \'setprop persist.sys.disable_rescue true\'')
176
177    def test_fetch_helpers_from_dut(self):
178        mock_run = self.mock_adb.run
179        self.tradefed._repository = '/repo/path'
180
181        mock_host = Mock()
182        self.tradefed._hosts = [mock_host]
183
184        # '::' is intentional and should be skipped.
185        mock_run.return_value.stdout = 'package1:package2::package3'
186
187        self.tradefed._fetch_helpers_from_dut()
188
189        mock_run.assert_called_with(
190                mock_host,
191                args=('shell', 'getprop',
192                      'ro.vendor.cts_interaction_helper_packages'))
193
194        self.assertEqual(mock_host.get_file.call_count, 3)
195
196        mock_host.get_file.assert_any_call(
197                '/usr/local/opt/google/vms/android/package1.apk',
198                '/repo/path/testcases',
199        )
200
201        mock_host.get_file.assert_any_call(
202                '/usr/local/opt/google/vms/android/package2.apk',
203                '/repo/path/testcases',
204        )
205
206        mock_host.get_file.assert_any_call(
207                '/usr/local/opt/google/vms/android/package3.apk',
208                '/repo/path/testcases',
209        )
210
211    def test_get_abilist(self):
212        mock_run = self.mock_adb.run
213        mock_host = Mock()
214        self.tradefed._hosts = [mock_host]
215
216        mock_run.return_value.stdout = 'arm,x86,my_awesome_architecture'
217
218        self.assertEqual(['arm', 'x86', 'my_awesome_architecture'],
219                         self.tradefed._get_abilist())
220
221        mock_run.assert_called_with(mock_host,
222                                    args=('shell', 'getprop',
223                                          'ro.product.cpu.abilist'))
224
225    def test_copy_extra_artifacts_dut(self):
226        mock_run = self.mock_adb.run
227        mock_host = Mock()
228
229        extra_artifacts = ['artifacts', '/path/to/some/file']
230        self.tradefed._copy_extra_artifacts_dut(extra_artifacts, mock_host,
231                                                self._outputdir)
232
233        self.assertEqual(mock_run.call_count, 2)
234
235        mock_run.assert_any_call(
236                mock_host,
237                args=('pull', 'artifacts', self._outputdir),
238                verbose=ANY,
239                timeout=ANY,
240        )
241
242        mock_run.assert_any_call(
243                mock_host,
244                args=('pull', '/path/to/some/file', self._outputdir),
245                verbose=ANY,
246                timeout=ANY,
247        )
248
249    # TODO(rkuroiwa): This test was added to test Adb.add_path.
250    # So most of these tradefed_test functions are mocked because
251    # they are not ncecessarily ready to be tested.
252    # Once the rest of the modules are tested, reevaluate unmocking them.
253    @patch('os.chmod')
254    @patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._instance_copyfile'
255           )
256    @patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._validate_download_cache'
257           )
258    @patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._download_to_cache'
259           )
260    @patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._invalidate_download_cache'
261           )
262    @patch('autotest_lib.server.cros.tradefed.tradefed_utils.lock')
263    def test_install_files(self, mock_lock, mock_invalidate_download_cache,
264                           mock_download_to_cache,
265                           mock_validate_download_cache, mock_instance_copy,
266                           mock_chmod):
267        mock_add_path = self.mock_adb.add_path
268        self.tradefed._tradefed_cache_lock = '/lock/lock_file'
269        self.tradefed._install_paths = []
270
271        mock_download_to_cache.return_value = '/path/to/downloaded/file'
272        mock_instance_copy.return_value = '/path/to/local/downloaded_file'
273
274        self.tradefed._install_files('gs://mybucket/path/to/dir', ['anyfile'],
275                                     stat.S_IRWXU)
276
277        mock_lock.assert_called_with('/lock/lock_file')
278        mock_invalidate_download_cache.assert_called()
279        mock_validate_download_cache.assert_called()
280        mock_chmod.assert_called_with('/path/to/local/downloaded_file',
281                                      stat.S_IRWXU)
282        mock_add_path.assert_called_with('/path/to/local')
283
284        self.assertEqual(self.tradefed._install_paths, ['/path/to/local'])
285