| # Copyright 2022 Google LLC |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| """Unit tests for suite session manager.""" |
| import os |
| import threading |
| import time |
| import unittest |
| from unittest import mock |
| |
| from local_agent import errors as agent_errors |
| from local_agent import logger as logger_module |
| from local_agent import suite_session_manager |
| |
| |
| ####################### Fake data for unit test ############################# |
| _FAKE_DEVICE_ID = 'fake-device-id' |
| _FAKE_DEVICE_ID2 = 'fake-device-id2' |
| _FAKE_TEST_SUITE_ID = 'fake-test-suite-id' |
| _FAKE_TEST_SUITE_ID2 = 'fake-test-suite-id2' |
| _FAKE_TEST_RESULT_ID = 'fake-test-result-id' |
| _START_TEST_SUITE = 'startTestSuite' |
| _END_TEST_SUITE = 'endTestSuite' |
| _THREADING_MODULE_PATH = 'threading.Thread' |
| _FAKE_OUTDATED_ARTIFACTS = 'outdated_artifacts.zip' |
| _FAKE_NOT_OUTDATED_ARTIFACTS = 'not_outdated_artifacts.zip' |
| ############################################################################## |
| |
| |
| def rpc_request(method, params): |
| """Simple wrapper for json rpc request.""" |
| return {'jsonrpc': '2.0', 'id': 0, 'method': method, 'params': params} |
| |
| |
| class SuiteSessionManagerTest(unittest.TestCase): |
| """Unit tests for local agent suite session manager.""" |
| |
| def setUp(self): |
| super().setUp() |
| fake_thread = mock.patch(_THREADING_MODULE_PATH) |
| fake_thread.start() |
| self.addCleanup(fake_thread.stop) |
| self.suite_mgr = suite_session_manager.SuiteSessionManager( |
| artifacts_fn=mock.Mock(), |
| artifact_root_dir='', |
| create_devices_fn=mock.Mock(), |
| close_devices_fn=mock.Mock()) |
| |
| def test_01_start_on_success(self): |
| """Verifies start method on success.""" |
| self.suite_mgr.start(termination_event=mock.Mock()) |
| self.suite_mgr._suite_timeout_checker.start.assert_called_once() |
| |
| @mock.patch.object( |
| suite_session_manager.SuiteSessionManager, |
| '_initialize_artifact_directory') |
| def test_02_start_test_suite_on_success(self, mock_init_dir): |
| """Verifies start_test_suite on success.""" |
| self.suite_mgr._ongoing_test_suite_id = None |
| start_suite = rpc_request( |
| _START_TEST_SUITE, |
| {'id': _FAKE_TEST_SUITE_ID, |
| 'dutDeviceIds': [_FAKE_DEVICE_ID,_FAKE_DEVICE_ID2]}) |
| expected_response = { |
| 'id': start_suite['id'], 'jsonrpc': '2.0', 'result': {}} |
| |
| rpc_response = self.suite_mgr.start_test_suite(start_suite) |
| |
| self.assertEqual(expected_response, rpc_response) |
| mock_init_dir.assert_called_once_with(_FAKE_TEST_SUITE_ID) |
| self.suite_mgr._create_devices.assert_called_once() |
| |
| @mock.patch.object( |
| suite_session_manager.SuiteSessionManager, |
| '_initialize_artifact_directory') |
| @mock.patch.object(suite_session_manager.SuiteSessionManager, 'clean_up') |
| def test_02_start_test_suite_force_on_success( |
| self, mock_force_cleanup, mock_init_dir): |
| """Verifies start_test_suite with forceStart=True on success.""" |
| self.suite_mgr._ongoing_test_suite_id = _FAKE_TEST_SUITE_ID |
| start_suite = rpc_request( |
| _START_TEST_SUITE, |
| {'id': _FAKE_TEST_SUITE_ID2, |
| 'forceStart': True, |
| 'dutDeviceIds': [_FAKE_DEVICE_ID]}) |
| expected_response = { |
| 'id': start_suite['id'], 'jsonrpc': '2.0', 'result': {}} |
| |
| rpc_response = self.suite_mgr.start_test_suite(start_suite) |
| |
| self.assertEqual(expected_response, rpc_response) |
| mock_force_cleanup.assert_called_once() |
| mock_init_dir.assert_called_once_with(_FAKE_TEST_SUITE_ID2) |
| |
| def test_02_start_test_suite_on_failure_no_device_ids(self): |
| """Verifies start_test_suite on failure with no device ids.""" |
| no_device_ids = rpc_request( |
| _START_TEST_SUITE, {'id': _FAKE_TEST_SUITE_ID}) |
| error_msg = 'Invalid rpc command, no dutDeviceIds' |
| with self.assertRaisesRegex(agent_errors.InvalidRPCError, error_msg): |
| self.suite_mgr.start_test_suite(no_device_ids) |
| |
| def test_02_start_test_suite_on_failure_invalid_session(self): |
| """Verifies start_test_suite on failure with incomplete session.""" |
| self.suite_mgr._ongoing_test_suite_id = _FAKE_TEST_SUITE_ID |
| invalid_start = rpc_request( |
| _START_TEST_SUITE, |
| {'id': _FAKE_TEST_SUITE_ID, 'dutDeviceIds': [_FAKE_DEVICE_ID]}) |
| error_msg = f'{_FAKE_TEST_SUITE_ID} has not ended yet.' |
| with self.assertRaisesRegex( |
| agent_errors.InvalidTestSuiteSessionError, error_msg): |
| self.suite_mgr.start_test_suite(invalid_start) |
| |
| @mock.patch.object(suite_session_manager.SuiteSessionManager, 'clean_up') |
| def test_03_end_test_suite_on_success(self, mock_cleanup): |
| """Verifies end test suite on success.""" |
| self.suite_mgr._ongoing_test_suite_id = _FAKE_TEST_SUITE_ID |
| self.suite_mgr._ongoing_test_suite_start_time = 0 |
| end_suite = rpc_request(_END_TEST_SUITE, {'id': _FAKE_TEST_SUITE_ID}) |
| expected_response = { |
| 'id': end_suite['id'], 'jsonrpc': '2.0', 'result': {}} |
| |
| rpc_response = self.suite_mgr.end_test_suite(end_suite) |
| |
| self.assertEqual(expected_response, rpc_response) |
| mock_cleanup.assert_called_once() |
| |
| def test_03_end_test_suite_on_failure_invalid_session(self): |
| """Verifies end_test_suite on failure with invalid session.""" |
| self.suite_mgr._ongoing_test_suite_id = None |
| invalid_end = rpc_request(_END_TEST_SUITE, {'id': _FAKE_TEST_SUITE_ID}) |
| error_msg = f'Session {_FAKE_TEST_SUITE_ID} has never started before.' |
| with self.assertRaisesRegex( |
| agent_errors.InvalidTestSuiteSessionError, error_msg): |
| self.suite_mgr.end_test_suite(invalid_end) |
| |
| @mock.patch.object( |
| suite_session_manager.SuiteSessionManager, '_remove_outdated_artifacts') |
| def test_04_clean_up_on_success(self, mock_remove): |
| """Verifies clean_up method on success.""" |
| self.suite_mgr._ongoing_test_suite_id = _FAKE_TEST_SUITE_ID |
| self.suite_mgr._ongoing_test_suite_start_time = 0 |
| |
| self.suite_mgr.clean_up(test_result_id=_FAKE_TEST_RESULT_ID) |
| |
| self.suite_mgr._close_devices.assert_called_once() |
| self.suite_mgr._compress_artifacts_and_upload.assert_called_once_with( |
| test_suite_id=_FAKE_TEST_SUITE_ID, |
| test_result_id=_FAKE_TEST_RESULT_ID) |
| mock_remove.assert_called_once() |
| |
| @mock.patch.object(logger_module, 'add_file_handler') |
| @mock.patch.object(os, 'makedirs') |
| @mock.patch.object(os.path, 'exists', return_value=False) |
| def test_05_initialize_artifact_directory_on_success( |
| self, mock_exists, mock_mk, mock_add): |
| """Verifies initialize_artifact_directory on success.""" |
| self.suite_mgr._initialize_artifact_directory(_FAKE_TEST_SUITE_ID) |
| mock_mk.assert_called_once() |
| mock_add.assert_called_once() |
| |
| @mock.patch.object(logger_module, 'logger') |
| @mock.patch.object(suite_session_manager, 'os') |
| def test_06_remove_outdated_artifacts_on_success( |
| self, mock_os, mock_logger): |
| """Verifies _remove_outdated_artifacts removes outdated files only.""" |
| mock_os.listdir.return_value = [ |
| _FAKE_OUTDATED_ARTIFACTS, |
| _FAKE_NOT_OUTDATED_ARTIFACTS |
| ] |
| mock_os.stat.side_effect = [ |
| mock.Mock(st_ctime=0), |
| mock.Mock(st_ctime=time.time()), |
| ] |
| mock_os.path.isfile.return_value = True |
| mock_os.path.exists.return_value = True |
| mock_os.path.join.return_value = _FAKE_OUTDATED_ARTIFACTS |
| |
| self.suite_mgr._remove_outdated_artifacts() |
| |
| mock_os.path.exists.assert_called_once() |
| mock_os.remove.assert_called_once() |
| self.assertEqual(2, mock_os.path.isfile.call_count) |
| self.assertEqual(_FAKE_OUTDATED_ARTIFACTS, |
| mock_os.remove.call_args.args[0]) |
| |
| @mock.patch.object(suite_session_manager, 'os') |
| def test_06_remove_outdated_artifacts_will_suppress_oserror( |
| self, mock_os): |
| """ |
| Verifies _remove_outdated_artifacts suppresses OSError |
| when cannot remove. |
| """ |
| mock_os.listdir.return_value = [_FAKE_OUTDATED_ARTIFACTS] |
| mock_os.path.isfile.return_value = True |
| mock_os.stat.return_value.st_ctime = 0 |
| mock_os.remove.side_effect = OSError |
| |
| self.suite_mgr._remove_outdated_artifacts() |
| |
| mock_os.path.exists.assert_called_once() |
| mock_os.remove.assert_called_once() |
| |
| @mock.patch.object(os, 'remove') |
| @mock.patch.object(os.path, 'exists', return_value=False) |
| def test_06_remove_outdated_artifacts_not_exists( |
| self, mock_exists, mock_rm): |
| """Verifies _remove_outdated_artifacts without existing artifacts.""" |
| self.suite_mgr._remove_outdated_artifacts() |
| mock_exists.assert_called_once() |
| self.assertEqual(0, mock_rm.call_count) |
| |
| @mock.patch.object(suite_session_manager.SuiteSessionManager, 'clean_up') |
| @mock.patch.object(suite_session_manager.time, 'sleep') |
| def test_07_check_suite_timeout_no_test_suite(self, mock_sleep, mock_cleanup): |
| """Verifies _initialize_suite_tracker no ongoing suite.""" |
| fake_terminiation_event = mock.Mock() |
| fake_terminiation_event.wait.side_effect = [False, True] |
| self.suite_mgr._termination_event = fake_terminiation_event |
| self.suite_mgr._ongoing_test_suite_id = None |
| |
| self.suite_mgr._check_suite_timeout() |
| |
| self.assertEqual(0, mock_cleanup.call_count) |
| |
| @mock.patch.object(suite_session_manager.SuiteSessionManager, 'clean_up') |
| @mock.patch.object(suite_session_manager.time, 'time') |
| @mock.patch.object(suite_session_manager.time, 'sleep') |
| def test_07_check_suite_timeout_on_success( |
| self, mock_sleep, mock_time, mock_cleanup): |
| """Verifies _initialize_suite_tracker on success.""" |
| fake_terminiation_event = mock.Mock() |
| fake_terminiation_event.wait.side_effect = [False, True] |
| self.suite_mgr._termination_event = fake_terminiation_event |
| self.suite_mgr._ongoing_test_suite_id = _FAKE_TEST_SUITE_ID |
| self.suite_mgr._ongoing_test_suite_start_time = 0 |
| self.suite_mgr._busy_devices = set() |
| mock_time.return_value = suite_session_manager._SUITE_SESSION_TIME_OUT |
| |
| self.suite_mgr._check_suite_timeout() |
| |
| mock_cleanup.assert_called_once() |
| |
| |
| if __name__ == '__main__': |
| unittest.main(failfast=True) |
