285 lines
11 KiB
Python
285 lines
11 KiB
Python
# Lint as: python2, python3
|
|
# Copyright 2022 The Chromium OS Authors. All rights reserved.
|
|
# Use of this source code is governed by a BSD-style license that can be
|
|
# found in the LICENSE file.
|
|
|
|
import unittest
|
|
import tempfile
|
|
import shutil
|
|
import stat
|
|
|
|
from unittest.mock import Mock, ANY, patch
|
|
from autotest_lib.client.common_lib import error
|
|
from autotest_lib.server.cros.tradefed import tradefed_test
|
|
|
|
|
|
class TradefedTestTest(unittest.TestCase):
|
|
"""Tests for TradefedTest class."""
|
|
|
|
def setUp(self):
|
|
self._mockjob_tmpdirs = []
|
|
self._bindir = tempfile.mkdtemp()
|
|
self._outputdir = tempfile.mkdtemp()
|
|
self.mock_adb = Mock()
|
|
self.tradefed = tradefed_test.TradefedTest(self.create_mock_job(),
|
|
self._bindir,
|
|
self._outputdir,
|
|
adb=self.mock_adb)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self._bindir)
|
|
shutil.rmtree(self._outputdir)
|
|
for tmpdir in self._mockjob_tmpdirs:
|
|
shutil.rmtree(tmpdir)
|
|
|
|
def create_mock_job(self):
|
|
"""Creates a mock necessary for constructing tradefed_test instance."""
|
|
mock_job = Mock()
|
|
mock_job.pkgmgr = None
|
|
mock_job.autodir = None
|
|
mock_job.tmpdir = tempfile.mkdtemp()
|
|
self._mockjob_tmpdirs.append(mock_job.tmpdir)
|
|
return mock_job
|
|
|
|
# Verify that try_adb_connect fails when run_adb_cmd fails.
|
|
@patch('autotest_lib.server.cros.tradefed.adb.get_adb_target')
|
|
def test_try_adb_connect_run_adb_fail(self, mock_get_adb_target):
|
|
mock_run_adb_cmd = self.mock_adb.run
|
|
|
|
# Exit status is set to non-0 to exit _try_adb_connect() early.
|
|
mock_run_adb_cmd.return_value.exit_status = 1
|
|
mock_get_adb_target.return_value = '123.76.0.29:3467'
|
|
|
|
self.assertFalse(self.tradefed._try_adb_connect(Mock()))
|
|
mock_run_adb_cmd.assert_called_with(ANY,
|
|
args=('connect',
|
|
'123.76.0.29:3467'),
|
|
verbose=ANY,
|
|
env=ANY,
|
|
ignore_status=ANY,
|
|
timeout=ANY)
|
|
|
|
# Verify that _run_tradefed_with_timeout works.
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._run')
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._tradefed_cmd_path'
|
|
)
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_utils.adb_keepalive')
|
|
@patch('autotest_lib.server.cros.tradefed.adb.get_adb_targets')
|
|
def test_run_tradefed_with_timeout(self, mock_get_adb_targets, _,
|
|
mock_tradefed_cmd_path, mock_run):
|
|
self.tradefed._install_paths = '/any/install/path'
|
|
|
|
mock_host1 = Mock()
|
|
mock_host2 = Mock()
|
|
self.tradefed._hosts = [mock_host1, mock_host2]
|
|
|
|
mock_get_adb_targets.return_value = ['host1:4321', 'host2:22']
|
|
|
|
mock_tradefed_cmd_path.return_value = '/any/path'
|
|
|
|
self.tradefed._run_tradefed_with_timeout(['command'], 1234)
|
|
mock_get_adb_targets.assert_called_with(self.tradefed._hosts)
|
|
|
|
def test_kill_adb_server(self):
|
|
mock_run = self.mock_adb.run
|
|
self.tradefed._kill_adb_server()
|
|
mock_run.assert_called_with(None,
|
|
args=('kill-server', ),
|
|
timeout=ANY,
|
|
verbose=ANY)
|
|
|
|
def test_verify_arc_hosts_single_host(self):
|
|
mock_run = self.mock_adb.run
|
|
mock_host = Mock()
|
|
self.tradefed._hosts = [mock_host]
|
|
|
|
self.tradefed._verify_arc_hosts()
|
|
|
|
mock_run.assert_called_with(mock_host,
|
|
args=('shell', 'getprop',
|
|
'ro.build.fingerprint'))
|
|
|
|
# Verify that multiple hosts with differet fingerprints fail.
|
|
def test_verify_arc_hosts_different_fingerprints(self):
|
|
mock_run = self.mock_adb.run
|
|
mock_host1 = Mock()
|
|
mock_host2 = Mock()
|
|
self.tradefed._hosts = [mock_host1, mock_host2]
|
|
|
|
side_effects = [Mock(), Mock()]
|
|
side_effects[0].stdout = 'fingerprint1'
|
|
side_effects[1].stdout = 'fingerprint2'
|
|
mock_run.side_effect = side_effects
|
|
|
|
self.assertRaises(error.TestFail, self.tradefed._verify_arc_hosts)
|
|
|
|
mock_run.assert_any_call(mock_host1,
|
|
args=('shell', 'getprop',
|
|
'ro.build.fingerprint'))
|
|
mock_run.assert_any_call(mock_host2,
|
|
args=('shell', 'getprop',
|
|
'ro.build.fingerprint'))
|
|
|
|
# Verify that wait for arc boot uses polling with adb.
|
|
@patch('autotest_lib.server.utils.poll_for_condition')
|
|
def test_wait_for_arc_boot(self, mock_poll_for_condition):
|
|
mock_run = self.mock_adb.run
|
|
|
|
# stdout just has to be something that evaluates to True.
|
|
mock_run.return_value.stdout = 'anything'
|
|
|
|
mock_host = Mock()
|
|
self.tradefed._wait_for_arc_boot(mock_host)
|
|
|
|
self.assertEqual(mock_run.call_count, 0)
|
|
|
|
# Verify that the condition function uses the expected adb command.
|
|
self.assertEqual(mock_poll_for_condition.call_count, 1)
|
|
args = mock_poll_for_condition.call_args[0]
|
|
condition_func = args[0]
|
|
self.assertTrue(condition_func())
|
|
|
|
mock_run.assert_called_with(mock_host,
|
|
args=('shell', 'pgrep', '-f',
|
|
'org.chromium.arc.intent_helper'),
|
|
ignore_status=True)
|
|
|
|
def test_disable_adb_install_dialog_android_version_over_29(self):
|
|
mock_run = self.mock_adb.run
|
|
mock_run.return_value.stdout = 'disabled'
|
|
|
|
self.tradefed._android_version = 30
|
|
mock_host = Mock()
|
|
self.tradefed._disable_adb_install_dialog(mock_host)
|
|
|
|
mock_run.assert_called_with(mock_host,
|
|
args=('shell', 'settings', 'put', 'global',
|
|
'verifier_verify_adb_installs', '0'),
|
|
verbose=ANY)
|
|
|
|
def test_disable_adb_install_dialog_android_version_under_29(self):
|
|
mock_run = self.mock_adb.run
|
|
|
|
mock_run.return_value.stdout = 'disabled'
|
|
|
|
self.tradefed._android_version = 28
|
|
mock_host = Mock()
|
|
self.tradefed._disable_adb_install_dialog(mock_host)
|
|
|
|
mock_run.assert_called_with(mock_host,
|
|
args=('shell', 'settings', 'put', 'global',
|
|
'verifier_verify_adb_installs', '0'),
|
|
verbose=ANY)
|
|
|
|
mock_host.run.assert_called_with(
|
|
'android-sh -c \'setprop persist.sys.disable_rescue true\'')
|
|
|
|
def test_fetch_helpers_from_dut(self):
|
|
mock_run = self.mock_adb.run
|
|
self.tradefed._repository = '/repo/path'
|
|
|
|
mock_host = Mock()
|
|
self.tradefed._hosts = [mock_host]
|
|
|
|
# '::' is intentional and should be skipped.
|
|
mock_run.return_value.stdout = 'package1:package2::package3'
|
|
|
|
self.tradefed._fetch_helpers_from_dut()
|
|
|
|
mock_run.assert_called_with(
|
|
mock_host,
|
|
args=('shell', 'getprop',
|
|
'ro.vendor.cts_interaction_helper_packages'))
|
|
|
|
self.assertEqual(mock_host.get_file.call_count, 3)
|
|
|
|
mock_host.get_file.assert_any_call(
|
|
'/usr/local/opt/google/vms/android/package1.apk',
|
|
'/repo/path/testcases',
|
|
)
|
|
|
|
mock_host.get_file.assert_any_call(
|
|
'/usr/local/opt/google/vms/android/package2.apk',
|
|
'/repo/path/testcases',
|
|
)
|
|
|
|
mock_host.get_file.assert_any_call(
|
|
'/usr/local/opt/google/vms/android/package3.apk',
|
|
'/repo/path/testcases',
|
|
)
|
|
|
|
def test_get_abilist(self):
|
|
mock_run = self.mock_adb.run
|
|
mock_host = Mock()
|
|
self.tradefed._hosts = [mock_host]
|
|
|
|
mock_run.return_value.stdout = 'arm,x86,my_awesome_architecture'
|
|
|
|
self.assertEqual(['arm', 'x86', 'my_awesome_architecture'],
|
|
self.tradefed._get_abilist())
|
|
|
|
mock_run.assert_called_with(mock_host,
|
|
args=('shell', 'getprop',
|
|
'ro.product.cpu.abilist'))
|
|
|
|
def test_copy_extra_artifacts_dut(self):
|
|
mock_run = self.mock_adb.run
|
|
mock_host = Mock()
|
|
|
|
extra_artifacts = ['artifacts', '/path/to/some/file']
|
|
self.tradefed._copy_extra_artifacts_dut(extra_artifacts, mock_host,
|
|
self._outputdir)
|
|
|
|
self.assertEqual(mock_run.call_count, 2)
|
|
|
|
mock_run.assert_any_call(
|
|
mock_host,
|
|
args=('pull', 'artifacts', self._outputdir),
|
|
verbose=ANY,
|
|
timeout=ANY,
|
|
)
|
|
|
|
mock_run.assert_any_call(
|
|
mock_host,
|
|
args=('pull', '/path/to/some/file', self._outputdir),
|
|
verbose=ANY,
|
|
timeout=ANY,
|
|
)
|
|
|
|
# TODO(rkuroiwa): This test was added to test Adb.add_path.
|
|
# So most of these tradefed_test functions are mocked because
|
|
# they are not ncecessarily ready to be tested.
|
|
# Once the rest of the modules are tested, reevaluate unmocking them.
|
|
@patch('os.chmod')
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._instance_copyfile'
|
|
)
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._validate_download_cache'
|
|
)
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._download_to_cache'
|
|
)
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_test.TradefedTest._invalidate_download_cache'
|
|
)
|
|
@patch('autotest_lib.server.cros.tradefed.tradefed_utils.lock')
|
|
def test_install_files(self, mock_lock, mock_invalidate_download_cache,
|
|
mock_download_to_cache,
|
|
mock_validate_download_cache, mock_instance_copy,
|
|
mock_chmod):
|
|
mock_add_path = self.mock_adb.add_path
|
|
self.tradefed._tradefed_cache_lock = '/lock/lock_file'
|
|
self.tradefed._install_paths = []
|
|
|
|
mock_download_to_cache.return_value = '/path/to/downloaded/file'
|
|
mock_instance_copy.return_value = '/path/to/local/downloaded_file'
|
|
|
|
self.tradefed._install_files('gs://mybucket/path/to/dir', ['anyfile'],
|
|
stat.S_IRWXU)
|
|
|
|
mock_lock.assert_called_with('/lock/lock_file')
|
|
mock_invalidate_download_cache.assert_called()
|
|
mock_validate_download_cache.assert_called()
|
|
mock_chmod.assert_called_with('/path/to/local/downloaded_file',
|
|
stat.S_IRWXU)
|
|
mock_add_path.assert_called_with('/path/to/local')
|
|
|
|
self.assertEqual(self.tradefed._install_paths, ['/path/to/local'])
|