diff options
author | Benjamin Klimczak <benjamin.klimczak@arm.com> | 2022-07-11 12:33:42 +0100 |
---|---|---|
committer | Benjamin Klimczak <benjamin.klimczak@arm.com> | 2022-07-26 14:08:21 +0100 |
commit | 5d81f37de09efe10f90512e50252be9c36925fcf (patch) | |
tree | b4d7cdfd051da0a6e882bdfcf280fd7ca7b39e57 /tests/mlia/test_utils_proc.py | |
parent | 7899b908c1fe6d86b92a80f3827ddd0ac05b674b (diff) | |
download | mlia-5d81f37de09efe10f90512e50252be9c36925fcf.tar.gz |
MLIA-551 Rework remains of AIET architecture
Re-factoring the code base to further merge the old AIET code into MLIA.
- Remove last traces of the backend type 'tool'
- Controlled systems removed, including SSH protocol, controller,
RunningCommand, locks etc.
- Build command / build dir and deploy functionality removed from
Applications and Systems
- Moving working_dir()
- Replace module 'output_parser' with new module 'output_consumer' and
merge Base64 parsing into it
- Change the output consumption to optionally remove (i.e. actually
consume) lines
- Use Base64 parsing in GenericInferenceOutputParser, replacing the
regex-based parsing and remove the now unused regex parsing
- Remove AIET reporting
- Pre-install applications by moving them to src/mlia/resources/backends
- Rename aiet-config.json to backend-config.json
- Move tests from tests/mlia/ to tests/
- Adapt unit tests to code changes
- Dependencies removed: paramiko, filelock, psutil
- Fix bug in corstone.py: The wrong resource directory was used which
broke the functionality to download backends.
- Use f-string formatting.
- Use logging instead of print.
Change-Id: I768bc3bb6b2eda57d219ad01be4a8e0a74167d76
Diffstat (limited to 'tests/mlia/test_utils_proc.py')
-rw-r--r-- | tests/mlia/test_utils_proc.py | 149 |
1 files changed, 0 insertions, 149 deletions
diff --git a/tests/mlia/test_utils_proc.py b/tests/mlia/test_utils_proc.py deleted file mode 100644 index 8316ca5..0000000 --- a/tests/mlia/test_utils_proc.py +++ /dev/null @@ -1,149 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Tests for the module utils/proc.""" -import signal -import subprocess -import time -from pathlib import Path -from unittest.mock import MagicMock - -import pytest - -from mlia.utils.proc import CommandExecutor -from mlia.utils.proc import working_directory - - -class TestCommandExecutor: - """Tests for class CommandExecutor.""" - - @staticmethod - def test_execute() -> None: - """Test command execution.""" - executor = CommandExecutor() - - retcode, stdout, stderr = executor.execute(["echo", "hello world!"]) - assert retcode == 0 - assert stdout.decode().strip() == "hello world!" - assert stderr.decode() == "" - - @staticmethod - def test_submit() -> None: - """Test command submittion.""" - executor = CommandExecutor() - - running_command = executor.submit(["sleep", "10"]) - assert running_command.is_alive() is True - assert running_command.exit_code() is None - - running_command.kill() - for _ in range(3): - time.sleep(0.5) - if not running_command.is_alive(): - break - - assert running_command.is_alive() is False - assert running_command.exit_code() == -9 - - with pytest.raises(subprocess.CalledProcessError): - executor.execute(["sleep", "-1"]) - - @staticmethod - @pytest.mark.parametrize("wait", [True, False]) - def test_stop(wait: bool) -> None: - """Test command termination.""" - executor = CommandExecutor() - - running_command = executor.submit(["sleep", "10"]) - running_command.stop(wait=wait) - - if wait: - assert running_command.is_alive() is False - - @staticmethod - def test_unable_to_stop(monkeypatch: pytest.MonkeyPatch) -> None: - """Test when command could not be stopped.""" - running_command_mock = MagicMock() - running_command_mock.poll.return_value = None - - monkeypatch.setattr( - "mlia.utils.proc.subprocess.Popen", - MagicMock(return_value=running_command_mock), - ) - - with pytest.raises(Exception, match="Unable to stop running command"): - executor = CommandExecutor() - running_command = executor.submit(["sleep", "10"]) - - running_command.stop(num_of_attempts=1, interval=0.1) - - running_command_mock.send_signal.assert_called_once_with(signal.SIGINT) - - @staticmethod - def test_stop_after_several_attempts(monkeypatch: pytest.MonkeyPatch) -> None: - """Test when command could be stopped after several attempts.""" - running_command_mock = MagicMock() - running_command_mock.poll.side_effect = [None, 0] - - monkeypatch.setattr( - "mlia.utils.proc.subprocess.Popen", - MagicMock(return_value=running_command_mock), - ) - - executor = CommandExecutor() - running_command = executor.submit(["sleep", "10"]) - - running_command.stop(num_of_attempts=1, interval=0.1) - running_command_mock.send_signal.assert_called_once_with(signal.SIGINT) - - @staticmethod - def test_send_signal() -> None: - """Test sending signal.""" - executor = CommandExecutor() - running_command = executor.submit(["sleep", "10"]) - running_command.send_signal(signal.SIGINT) - - # wait a bit for a signal processing - time.sleep(1) - - assert running_command.is_alive() is False - assert running_command.exit_code() == -2 - - @staticmethod - @pytest.mark.parametrize( - "redirect_output, expected_output", [[True, "hello\n"], [False, ""]] - ) - def test_wait( - capsys: pytest.CaptureFixture, redirect_output: bool, expected_output: str - ) -> None: - """Test wait completion functionality.""" - executor = CommandExecutor() - - running_command = executor.submit(["echo", "hello"]) - running_command.wait(redirect_output=redirect_output) - - out, _ = capsys.readouterr() - assert out == expected_output - - -@pytest.mark.parametrize( - "should_exist, create_dir", - [ - [True, False], - [False, True], - ], -) -def test_working_directory_context_manager( - tmp_path: Path, should_exist: bool, create_dir: bool -) -> None: - """Test working_directory context manager.""" - prev_wd = Path.cwd() - - working_dir = tmp_path / "work_dir" - if should_exist: - working_dir.mkdir() - - with working_directory(working_dir, create_dir=create_dir) as current_working_dir: - assert current_working_dir.is_dir() - assert Path.cwd() == current_working_dir - - assert Path.cwd() == prev_wd |