| #!/usr/bin/env vpython3 |
| # Copyright 2022 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import unittest.mock as mock |
| import sys |
| import unittest |
| import subprocess |
| import os |
| import signal |
| import glob |
| import copy |
| |
| # if the current directory is in scripts (pwd), then we need to |
| # add plugin in order to import from that directory |
| if os.path.split(os.path.dirname(__file__))[1] != 'plugin': |
| sys.path.append( |
| os.path.join(os.path.abspath(os.path.dirname(__file__)), 'plugin')) |
| |
| # if executing from plugin directory, pull in scripts |
| else: |
| sys.path.append( |
| os.path.join(os.path.abspath(os.path.dirname(__file__)), '..')) |
| from plugin_constants import PLUGIN_PROTOS_PATH, MAX_RECORDED_COUNT |
| from test_plugins import VideoRecorderPlugin, BasePlugin, FileCopyPlugin |
| import iossim_util |
| |
| sys.path.append(PLUGIN_PROTOS_PATH) |
| import test_plugin_service_pb2 |
| import test_plugin_service_pb2_grpc |
| |
| TEST_DEVICE_ID = '123' |
| TEST_DEVICE_NAME = 'simulator_x_y' |
| TEST_DEVICE_PATH = '/root/dir' |
| TEST_CASE_NAME = '[AAA_BBB]' |
| TEST_CASE_INFO = test_plugin_service_pb2.TestCaseInfo(name=TEST_CASE_NAME) |
| TEST_DEVICE_INFO = test_plugin_service_pb2.DeviceInfo(name=TEST_DEVICE_NAME) |
| OUT_DIR = 'out/dir' |
| TEST_DEVICE_CACHE = { |
| TEST_DEVICE_NAME: { |
| 'UDID': TEST_DEVICE_ID, |
| 'path': TEST_DEVICE_PATH |
| } |
| } |
| |
| |
| class BasePluginTest(unittest.TestCase): |
| |
| @mock.patch("iossim_util.get_simulator_list") |
| def test_get_udid_and_path_for_device_name_no_cache(self, mock_get_list): |
| mock_get_list.return_value = { |
| 'devices': { |
| 'RUNTIME': [{ |
| 'name': TEST_DEVICE_NAME, |
| 'udid': TEST_DEVICE_ID |
| }] |
| } |
| } |
| cache = {} |
| base_plugin = BasePlugin(cache, 'OUT_DIR') |
| |
| self.assertEqual( |
| base_plugin.get_udid_and_path_for_device_name(TEST_DEVICE_NAME, |
| [TEST_DEVICE_PATH]), |
| (TEST_DEVICE_ID, TEST_DEVICE_PATH)) |
| mock_get_list.assert_called_once_with(TEST_DEVICE_PATH) |
| self.assertEqual( |
| base_plugin.device_info_cache.get(TEST_DEVICE_NAME), { |
| 'UDID': TEST_DEVICE_ID, |
| 'path': TEST_DEVICE_PATH, |
| }) |
| |
| @mock.patch('iossim_util.get_simulator_list') |
| def test_get_udid_and_path_for_device_name_with_cache(self, mock_get_list): |
| base_plugin = BasePlugin(TEST_DEVICE_CACHE, 'OUT_DIR') |
| |
| self.assertEqual( |
| base_plugin.get_udid_and_path_for_device_name(TEST_DEVICE_NAME), |
| (TEST_DEVICE_ID, TEST_DEVICE_PATH)) |
| mock_get_list.assert_not_called() |
| |
| |
| class VideoRecorderPluginTest(unittest.TestCase): |
| |
| @mock.patch("subprocess.Popen") |
| def test_test_case_will_start_succeed(self, mock_popen): |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_will_start(request) |
| file_name = video_recorder_plugin.get_video_file_name(TEST_CASE_NAME, 0) |
| file_dir = os.path.join(OUT_DIR, file_name) |
| cmd = [ |
| 'xcrun', 'simctl', '--set', TEST_DEVICE_PATH, 'io', TEST_DEVICE_ID, |
| 'recordVideo', '--codec=h264', '-f', file_dir |
| ] |
| mock_popen.assert_called_once_with(cmd) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).test_case_name == TEST_CASE_NAME) |
| |
| @mock.patch("subprocess.Popen") |
| def test_test_case_will_start_exceedMaxRecordedCount(self, mock_popen): |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.testcase_recorded_count[ |
| TEST_CASE_NAME] = MAX_RECORDED_COUNT |
| video_recorder_plugin.test_case_will_start(request) |
| mock_popen.assert_not_called() |
| |
| @mock.patch("subprocess.Popen") |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_test_case_will_start_previousProcessNotTerminated( |
| self, mock_os_remove, mock_os_kill, mock_popen): |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_will_start(request) |
| video_recorder_plugin.test_case_will_start(request) |
| mock_os_kill.assert_called_once_with(mock.ANY, signal.SIGTERM) |
| file_name = video_recorder_plugin.get_video_file_name(TEST_CASE_NAME, 0) |
| file_dir = os.path.join(OUT_DIR, file_name) |
| mock_os_remove.assert_called_once_with(file_dir) |
| cmd = [ |
| 'xcrun', 'simctl', '--set', TEST_DEVICE_PATH, 'io', TEST_DEVICE_ID, |
| 'recordVideo', '--codec=h264', '-f', file_dir |
| ] |
| mock_popen.assert_called_with(cmd) |
| |
| @mock.patch("subprocess.Popen") |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_test_case_did_fail_succeed(self, mock_os_remove, mock_os_kill, |
| mock_popen): |
| # first, start recording |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_will_start(request) |
| |
| # then test case fails |
| request = test_plugin_service_pb2.TestCaseDidFailRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_did_fail(request) |
| mock_os_kill.assert_called_once_with(mock.ANY, signal.SIGINT) |
| mock_os_remove.assert_not_called() |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).process == None) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).test_case_name == None) |
| self.assertTrue( |
| video_recorder_plugin.testcase_recorded_count[TEST_CASE_NAME] == 1) |
| |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_test_case_did_fail_noRecordingRunning(self, mock_os_remove, |
| mock_os_kill): |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseDidFailRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_did_fail(request) |
| mock_os_kill.assert_not_called() |
| mock_os_remove.assert_not_called() |
| |
| @mock.patch("subprocess.Popen") |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_test_case_did_finish_succeed(self, mock_os_remove, mock_os_kill, |
| mock_popen): |
| # first, start recording |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_will_start(request) |
| |
| # then test case finishes |
| request = test_plugin_service_pb2.TestCaseDidFinishRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_did_finish(request) |
| mock_os_kill.assert_called_once_with(mock.ANY, signal.SIGTERM) |
| file_name = video_recorder_plugin.get_video_file_name(TEST_CASE_NAME, 0) |
| file_dir = os.path.join(OUT_DIR, file_name) |
| mock_os_remove.assert_called_once_with(file_dir) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).process == None) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).test_case_name == None) |
| self.assertTrue( |
| TEST_CASE_NAME not in video_recorder_plugin.testcase_recorded_count) |
| |
| @mock.patch("subprocess.Popen") |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_test_case_did_finish_remove_file_failed(self, mock_os_remove, |
| mock_os_kill, mock_popen): |
| # first, start recording |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_will_start(request) |
| |
| # then test case finishes |
| mock_os_remove.side_effect = FileNotFoundError |
| request = test_plugin_service_pb2.TestCaseDidFinishRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| # this should not throw exception because it's caught |
| video_recorder_plugin.test_case_did_finish(request) |
| mock_os_kill.assert_called_once_with(mock.ANY, signal.SIGTERM) |
| file_name = video_recorder_plugin.get_video_file_name(TEST_CASE_NAME, 0) |
| file_dir = os.path.join(OUT_DIR, file_name) |
| mock_os_remove.assert_called_once_with(file_dir) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).process == None) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).test_case_name == None) |
| self.assertTrue( |
| TEST_CASE_NAME not in video_recorder_plugin.testcase_recorded_count) |
| |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_test_case_did_finish_noRecordingRunning(self, mock_os_remove, |
| mock_os_kill): |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseDidFinishRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_did_finish(request) |
| mock_os_kill.assert_not_called() |
| mock_os_remove.assert_not_called() |
| |
| @mock.patch("subprocess.Popen") |
| @mock.patch("os.kill") |
| @mock.patch("os.remove") |
| def test_reset_succeed(self, mock_os_remove, mock_os_kill, mock_popen): |
| # first, start recording |
| video_recorder_plugin = VideoRecorderPlugin( |
| copy.deepcopy(TEST_DEVICE_CACHE), OUT_DIR) |
| request = test_plugin_service_pb2.TestCaseWillStartRequest( |
| test_case_info=TEST_CASE_INFO, device_info=TEST_DEVICE_INFO) |
| video_recorder_plugin.test_case_will_start(request) |
| |
| # reset |
| video_recorder_plugin.reset() |
| mock_os_kill.assert_called_once_with(mock.ANY, signal.SIGTERM) |
| file_name = video_recorder_plugin.get_video_file_name(TEST_CASE_NAME, 0) |
| file_dir = os.path.join(OUT_DIR, file_name) |
| mock_os_remove.assert_called_once_with(file_dir) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).process == None) |
| self.assertTrue( |
| video_recorder_plugin.recording_process_for_device_name( |
| TEST_DEVICE_NAME).test_case_name == None) |
| |
| # reset again to make sure no exception is thrown |
| video_recorder_plugin.reset() |
| |
| |
| class FileCopyPluginTest(unittest.TestCase): |
| |
| @mock.patch("os.path.exists") |
| @mock.patch("os.mkdir") |
| @mock.patch("glob.glob") |
| @mock.patch("shutil.move") |
| def testOutputPathExists(self, move_mock: mock.MagicMock, |
| glob_mock: mock.MagicMock, |
| mkdir_mock: mock.MagicMock, |
| path_mock: mock.MagicMock): |
| path_mock.return_value = True |
| glob_mock.return_value = ["glob_return_value"] |
| |
| file_copy_plugin = FileCopyPlugin('GLOB_PATTERN', OUT_DIR, |
| copy.deepcopy(TEST_DEVICE_CACHE)) |
| request = test_plugin_service_pb2.TestBundleWillFinishRequest( |
| device_info=TEST_DEVICE_INFO) |
| |
| file_copy_plugin.test_bundle_will_finish(request) |
| |
| mkdir_mock.assert_not_called() |
| path_mock.assert_called_once_with(OUT_DIR) |
| glob_mock.assert_called_once_with( |
| os.path.join(TEST_DEVICE_PATH, TEST_DEVICE_ID, "GLOB_PATTERN")) |
| move_mock.assert_called_once_with("glob_return_value", OUT_DIR) |
| |
| @mock.patch("os.path.exists") |
| @mock.patch("os.mkdir") |
| @mock.patch("glob.glob") |
| @mock.patch("shutil.move") |
| def testOutputPathDoesNotExist(self, move_mock: mock.MagicMock, |
| glob_mock: mock.MagicMock, |
| mkdir_mock: mock.MagicMock, |
| path_mock: mock.MagicMock): |
| path_mock.return_value = False |
| glob_mock.return_value = ["glob_return_value"] |
| |
| file_copy_plugin = FileCopyPlugin('GLOB_PATTERN', OUT_DIR, |
| copy.deepcopy(TEST_DEVICE_CACHE)) |
| request = test_plugin_service_pb2.TestBundleWillFinishRequest( |
| device_info=TEST_DEVICE_INFO) |
| |
| file_copy_plugin.test_bundle_will_finish(request) |
| |
| mkdir_mock.assert_called_once_with(OUT_DIR) |
| path_mock.assert_called_once_with(OUT_DIR) |
| glob_mock.assert_called_once_with( |
| os.path.join(TEST_DEVICE_PATH, TEST_DEVICE_ID, "GLOB_PATTERN")) |
| move_mock.assert_called_once_with("glob_return_value", OUT_DIR) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
