diff options
author | Dmitrii Agibov <dmitrii.agibov@arm.com> | 2022-11-18 16:34:03 +0000 |
---|---|---|
committer | Dmitrii Agibov <dmitrii.agibov@arm.com> | 2022-11-29 14:44:13 +0000 |
commit | 37959522a805a5e23c930ed79aac84920c3cb208 (patch) | |
tree | 484af1240a93c955a72ce2e452432383b6704b56 /tests | |
parent | 5568f9f000d673ac53e710dcc8991fec6e8a5488 (diff) | |
download | mlia-37959522a805a5e23c930ed79aac84920c3cb208.tar.gz |
Move backends functionality into separate modules
- Move backend management/executor code into module backend_core
- Create separate module for each backend in "backend" module
- Move each backend into corresponding module
- Split Vela wrapper into several submodules
Change-Id: If01b6774aab6501951212541cc5d7f5aa7c97e95
Diffstat (limited to 'tests')
26 files changed, 1415 insertions, 1252 deletions
diff --git a/tests/conftest.py b/tests/conftest.py index b1f32dc..feb2aa0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,12 +10,12 @@ from typing import Generator import pytest import tensorflow as tf +from mlia.backend.vela.compiler import optimize_model from mlia.core.context import ExecutionContext from mlia.devices.ethosu.config import EthosUConfiguration from mlia.nn.tensorflow.utils import convert_to_tflite from mlia.nn.tensorflow.utils import save_keras_model from mlia.nn.tensorflow.utils import save_tflite_model -from mlia.tools.vela_wrapper import optimize_model @pytest.fixture(scope="session", name="test_resources_path") @@ -68,7 +68,9 @@ def test_resources(monkeypatch: pytest.MonkeyPatch, test_resources_path: Path) - """Return path to the test resources.""" return test_resources_path / "backends" - monkeypatch.setattr("mlia.backend.fs.get_backend_resources", get_test_resources) + monkeypatch.setattr( + "mlia.backend.executor.fs.get_backend_resources", get_test_resources + ) yield diff --git a/tests/test_api.py b/tests/test_api.py index 6fa15b3..b9ab8ea 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -118,12 +118,12 @@ def test_get_advisor( [ [ "ethos-u55-128", - "mlia.tools.vela_wrapper.generate_supported_operators_report", + "mlia.devices.ethosu.operators.generate_supported_operators_report", None, ], [ "ethos-u65-256", - "mlia.tools.vela_wrapper.generate_supported_operators_report", + "mlia.devices.ethosu.operators.generate_supported_operators_report", None, ], [ diff --git a/tests/test_tools_metadata_corstone.py b/tests/test_backend_corstone_install.py index a7d81f2..3b05a49 100644 --- a/tests/test_tools_metadata_corstone.py +++ b/tests/test_backend_corstone_install.py @@ -10,21 +10,21 @@ from unittest.mock import MagicMock import pytest -from mlia.backend.manager import BackendRunner -from mlia.tools.metadata.common import DownloadAndInstall -from mlia.tools.metadata.common import InstallFromPath -from mlia.tools.metadata.corstone import BackendInfo -from mlia.tools.metadata.corstone import BackendInstallation -from mlia.tools.metadata.corstone import BackendInstaller -from mlia.tools.metadata.corstone import BackendMetadata -from mlia.tools.metadata.corstone import CompoundPathChecker -from mlia.tools.metadata.corstone import Corstone300Installer -from mlia.tools.metadata.corstone import get_corstone_300_installation -from mlia.tools.metadata.corstone import get_corstone_310_installation -from mlia.tools.metadata.corstone import get_corstone_installations -from mlia.tools.metadata.corstone import PackagePathChecker -from mlia.tools.metadata.corstone import PathChecker -from mlia.tools.metadata.corstone import StaticPathChecker +from mlia.backend.corstone.install import Corstone300Installer +from mlia.backend.corstone.install import get_corstone_300_installation +from mlia.backend.corstone.install import get_corstone_310_installation +from mlia.backend.corstone.install import get_corstone_installations +from mlia.backend.corstone.install import PackagePathChecker +from mlia.backend.corstone.install import StaticPathChecker +from mlia.backend.executor.runner import BackendRunner +from mlia.backend.install import BackendInfo +from mlia.backend.install import BackendInstallation +from mlia.backend.install import BackendInstaller +from mlia.backend.install import BackendMetadata +from mlia.backend.install import CompoundPathChecker +from mlia.backend.install import DownloadAndInstall +from mlia.backend.install import InstallFromPath +from mlia.backend.install import PathChecker @pytest.fixture(name="test_mlia_resources") @@ -36,7 +36,7 @@ def fixture_test_mlia_resources( mlia_resources.mkdir() monkeypatch.setattr( - "mlia.tools.metadata.corstone.get_mlia_resources", + "mlia.backend.install.get_mlia_resources", MagicMock(return_value=mlia_resources), ) @@ -88,10 +88,12 @@ def test_could_be_installed_depends_on_platform( ) -> None: """Test that installation could not be installed on unsupported platform.""" monkeypatch.setattr( - "mlia.tools.metadata.corstone.platform.system", MagicMock(return_value=platform) + "mlia.backend.install.platform.system", + MagicMock(return_value=platform), ) monkeypatch.setattr( - "mlia.tools.metadata.corstone.all_paths_valid", MagicMock(return_value=True) + "mlia.backend.install.all_paths_valid", + MagicMock(return_value=True), ) backend_runner_mock = MagicMock(spec=BackendRunner) @@ -413,7 +415,7 @@ def test_corstone_300_installer( command_mock = MagicMock() monkeypatch.setattr( - "mlia.tools.metadata.corstone.subprocess.check_call", command_mock + "mlia.backend.corstone.install.subprocess.check_call", command_mock ) installer = Corstone300Installer() result = installer(eula_agreement, tmp_path) @@ -455,14 +457,14 @@ def test_corstone_vht_install( create_destination_and_install_mock = MagicMock() + monkeypatch.setattr("mlia.backend.install.all_files_exist", _all_files_exist) + monkeypatch.setattr( - "mlia.tools.metadata.corstone.all_files_exist", _all_files_exist + "mlia.backend.executor.system.get_available_systems", lambda: [] ) - monkeypatch.setattr("mlia.backend.system.get_available_systems", lambda: []) - monkeypatch.setattr( - "mlia.backend.system.create_destination_and_install", + "mlia.backend.executor.system.create_destination_and_install", create_destination_and_install_mock, ) @@ -478,7 +480,7 @@ def test_corstone_uninstall( remove_system_mock = MagicMock() monkeypatch.setattr( - "mlia.tools.metadata.corstone.remove_system", + "mlia.backend.install.remove_system", remove_system_mock, ) diff --git a/tests/test_backend_corstone_performance.py b/tests/test_backend_corstone_performance.py new file mode 100644 index 0000000..1734eb9 --- /dev/null +++ b/tests/test_backend_corstone_performance.py @@ -0,0 +1,519 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for module backend/manager.""" +from __future__ import annotations + +import base64 +import json +from contextlib import ExitStack as does_not_raise +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock +from unittest.mock import PropertyMock + +import pytest + +from mlia.backend.corstone.performance import BackendRunner +from mlia.backend.corstone.performance import DeviceInfo +from mlia.backend.corstone.performance import estimate_performance +from mlia.backend.corstone.performance import GenericInferenceOutputParser +from mlia.backend.corstone.performance import GenericInferenceRunnerEthosU +from mlia.backend.corstone.performance import get_generic_runner +from mlia.backend.corstone.performance import ModelInfo +from mlia.backend.corstone.performance import PerformanceMetrics +from mlia.backend.executor.application import get_application +from mlia.backend.executor.execution import ExecutionContext +from mlia.backend.executor.output_consumer import Base64OutputConsumer +from mlia.backend.executor.system import get_system +from mlia.backend.install import get_system_name +from mlia.backend.install import is_supported +from mlia.backend.install import supported_backends + + +def _mock_encode_b64(data: dict[str, int]) -> str: + """ + Encode the given data into a mock base64-encoded string of JSON. + + This reproduces the base64 encoding done in the Corstone applications. + + JSON example: + + ```json + [{'count': 1, + 'profiling_group': 'Inference', + 'samples': [{'name': 'NPU IDLE', 'value': [612]}, + {'name': 'NPU AXI0_RD_DATA_BEAT_RECEIVED', 'value': [165872]}, + {'name': 'NPU AXI0_WR_DATA_BEAT_WRITTEN', 'value': [88712]}, + {'name': 'NPU AXI1_RD_DATA_BEAT_RECEIVED', 'value': [57540]}, + {'name': 'NPU ACTIVE', 'value': [520489]}, + {'name': 'NPU TOTAL', 'value': [521101]}]}] + ``` + """ + wrapped_data = [ + { + "count": 1, + "profiling_group": "Inference", + "samples": [ + {"name": name, "value": [value]} for name, value in data.items() + ], + } + ] + json_str = json.dumps(wrapped_data) + json_bytes = bytearray(json_str, encoding="utf-8") + json_b64 = base64.b64encode(json_bytes).decode("utf-8") + tag = Base64OutputConsumer.TAG_NAME + return f"<{tag}>{json_b64}</{tag}>" + + +@pytest.mark.parametrize( + "data, is_ready, result, missed_keys", + [ + ( + [], + False, + {}, + { + "npu_active_cycles", + "npu_axi0_rd_data_beat_received", + "npu_axi0_wr_data_beat_written", + "npu_axi1_rd_data_beat_received", + "npu_idle_cycles", + "npu_total_cycles", + }, + ), + ( + ["sample text"], + False, + {}, + { + "npu_active_cycles", + "npu_axi0_rd_data_beat_received", + "npu_axi0_wr_data_beat_written", + "npu_axi1_rd_data_beat_received", + "npu_idle_cycles", + "npu_total_cycles", + }, + ), + ( + [_mock_encode_b64({"NPU AXI0_RD_DATA_BEAT_RECEIVED": 123})], + False, + {"npu_axi0_rd_data_beat_received": 123}, + { + "npu_active_cycles", + "npu_axi0_wr_data_beat_written", + "npu_axi1_rd_data_beat_received", + "npu_idle_cycles", + "npu_total_cycles", + }, + ), + ( + [ + _mock_encode_b64( + { + "NPU AXI0_RD_DATA_BEAT_RECEIVED": 1, + "NPU AXI0_WR_DATA_BEAT_WRITTEN": 2, + "NPU AXI1_RD_DATA_BEAT_RECEIVED": 3, + "NPU ACTIVE": 4, + "NPU IDLE": 5, + "NPU TOTAL": 6, + } + ) + ], + True, + { + "npu_axi0_rd_data_beat_received": 1, + "npu_axi0_wr_data_beat_written": 2, + "npu_axi1_rd_data_beat_received": 3, + "npu_active_cycles": 4, + "npu_idle_cycles": 5, + "npu_total_cycles": 6, + }, + set(), + ), + ], +) +def test_generic_inference_output_parser( + data: dict[str, int], is_ready: bool, result: dict, missed_keys: set[str] +) -> None: + """Test generic runner output parser.""" + parser = GenericInferenceOutputParser() + + for line in data: + parser.feed(line) + + assert parser.is_ready() == is_ready + assert parser.result == result + assert parser.missed_keys() == missed_keys + + +@pytest.mark.parametrize( + "device, system, application, backend, expected_error", + [ + ( + DeviceInfo(device_type="ethos-u55", mac=32), + ("Corstone-300: Cortex-M55+Ethos-U55", True), + ("Generic Inference Runner: Ethos-U55", True), + "Corstone-300", + does_not_raise(), + ), + ( + DeviceInfo(device_type="ethos-u55", mac=32), + ("Corstone-300: Cortex-M55+Ethos-U55", False), + ("Generic Inference Runner: Ethos-U55", False), + "Corstone-300", + pytest.raises( + Exception, + match=r"System Corstone-300: Cortex-M55\+Ethos-U55 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u55", mac=32), + ("Corstone-300: Cortex-M55+Ethos-U55", True), + ("Generic Inference Runner: Ethos-U55", False), + "Corstone-300", + pytest.raises( + Exception, + match=r"Application Generic Inference Runner: Ethos-U55 " + r"for the system Corstone-300: Cortex-M55\+Ethos-U55 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u55", mac=32), + ("Corstone-310: Cortex-M85+Ethos-U55", True), + ("Generic Inference Runner: Ethos-U55", True), + "Corstone-310", + does_not_raise(), + ), + ( + DeviceInfo(device_type="ethos-u55", mac=32), + ("Corstone-310: Cortex-M85+Ethos-U55", False), + ("Generic Inference Runner: Ethos-U55", False), + "Corstone-310", + pytest.raises( + Exception, + match=r"System Corstone-310: Cortex-M85\+Ethos-U55 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u55", mac=32), + ("Corstone-310: Cortex-M85+Ethos-U55", True), + ("Generic Inference Runner: Ethos-U55", False), + "Corstone-310", + pytest.raises( + Exception, + match=r"Application Generic Inference Runner: Ethos-U55 " + r"for the system Corstone-310: Cortex-M85\+Ethos-U55 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u65", mac=512), + ("Corstone-300: Cortex-M55+Ethos-U65", True), + ("Generic Inference Runner: Ethos-U65", True), + "Corstone-300", + does_not_raise(), + ), + ( + DeviceInfo(device_type="ethos-u65", mac=512), + ("Corstone-300: Cortex-M55+Ethos-U65", False), + ("Generic Inference Runner: Ethos-U65", False), + "Corstone-300", + pytest.raises( + Exception, + match=r"System Corstone-300: Cortex-M55\+Ethos-U65 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u65", mac=512), + ("Corstone-300: Cortex-M55+Ethos-U65", True), + ("Generic Inference Runner: Ethos-U65", False), + "Corstone-300", + pytest.raises( + Exception, + match=r"Application Generic Inference Runner: Ethos-U65 " + r"for the system Corstone-300: Cortex-M55\+Ethos-U65 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u65", mac=512), + ("Corstone-310: Cortex-M85+Ethos-U65", True), + ("Generic Inference Runner: Ethos-U65", True), + "Corstone-310", + does_not_raise(), + ), + ( + DeviceInfo(device_type="ethos-u65", mac=512), + ("Corstone-310: Cortex-M85+Ethos-U65", False), + ("Generic Inference Runner: Ethos-U65", False), + "Corstone-310", + pytest.raises( + Exception, + match=r"System Corstone-310: Cortex-M85\+Ethos-U65 is not installed", + ), + ), + ( + DeviceInfo(device_type="ethos-u65", mac=512), + ("Corstone-310: Cortex-M85+Ethos-U65", True), + ("Generic Inference Runner: Ethos-U65", False), + "Corstone-310", + pytest.raises( + Exception, + match=r"Application Generic Inference Runner: Ethos-U65 " + r"for the system Corstone-310: Cortex-M85\+Ethos-U65 is not installed", + ), + ), + ( + DeviceInfo( + device_type="unknown_device", # type: ignore + mac=None, # type: ignore + ), + ("some_system", False), + ("some_application", False), + "some backend", + pytest.raises(Exception, match="Unsupported device unknown_device"), + ), + ], +) +def test_estimate_performance( + device: DeviceInfo, + system: tuple[str, bool], + application: tuple[str, bool], + backend: str, + expected_error: Any, + test_tflite_model: Path, + backend_runner: MagicMock, +) -> None: + """Test getting performance estimations.""" + system_name, system_installed = system + application_name, application_installed = application + + backend_runner.is_system_installed.return_value = system_installed + backend_runner.is_application_installed.return_value = application_installed + + mock_context = create_mock_context( + [ + _mock_encode_b64( + { + "NPU AXI0_RD_DATA_BEAT_RECEIVED": 1, + "NPU AXI0_WR_DATA_BEAT_WRITTEN": 2, + "NPU AXI1_RD_DATA_BEAT_RECEIVED": 3, + "NPU ACTIVE": 4, + "NPU IDLE": 5, + "NPU TOTAL": 6, + } + ) + ] + ) + + backend_runner.run_application.return_value = mock_context + + with expected_error: + perf_metrics = estimate_performance( + ModelInfo(test_tflite_model), device, backend + ) + + assert isinstance(perf_metrics, PerformanceMetrics) + assert perf_metrics == PerformanceMetrics( + npu_axi0_rd_data_beat_received=1, + npu_axi0_wr_data_beat_written=2, + npu_axi1_rd_data_beat_received=3, + npu_active_cycles=4, + npu_idle_cycles=5, + npu_total_cycles=6, + ) + + assert backend_runner.is_system_installed.called_once_with(system_name) + assert backend_runner.is_application_installed.called_once_with( + application_name, system_name + ) + + +@pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) +def test_estimate_performance_insufficient_data( + backend_runner: MagicMock, test_tflite_model: Path, backend: str +) -> None: + """Test that performance could not be estimated when not all data presented.""" + backend_runner.is_system_installed.return_value = True + backend_runner.is_application_installed.return_value = True + + no_total_cycles_output = { + "NPU AXI0_RD_DATA_BEAT_RECEIVED": 1, + "NPU AXI0_WR_DATA_BEAT_WRITTEN": 2, + "NPU AXI1_RD_DATA_BEAT_RECEIVED": 3, + "NPU ACTIVE": 4, + "NPU IDLE": 5, + } + mock_context = create_mock_context([_mock_encode_b64(no_total_cycles_output)]) + + backend_runner.run_application.return_value = mock_context + + with pytest.raises( + Exception, match="Unable to get performance metrics, insufficient data" + ): + device = DeviceInfo(device_type="ethos-u55", mac=32) + estimate_performance(ModelInfo(test_tflite_model), device, backend) + + +def create_mock_process(stdout: list[str], stderr: list[str]) -> MagicMock: + """Mock underlying process.""" + mock_process = MagicMock() + mock_process.poll.return_value = 0 + type(mock_process).stdout = PropertyMock(return_value=iter(stdout)) + type(mock_process).stderr = PropertyMock(return_value=iter(stderr)) + return mock_process + + +def create_mock_context(stdout: list[str]) -> ExecutionContext: + """Mock ExecutionContext.""" + ctx = ExecutionContext( + app=get_application("application_1")[0], + app_params=[], + system=get_system("System 1"), + system_params=[], + ) + ctx.stdout = bytearray("\n".join(stdout).encode("utf-8")) + return ctx + + +@pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) +def test_estimate_performance_invalid_output( + test_tflite_model: Path, backend_runner: MagicMock, backend: str +) -> None: + """Test estimation could not be done if inference produces unexpected output.""" + backend_runner.is_system_installed.return_value = True + backend_runner.is_application_installed.return_value = True + + mock_context = create_mock_context(["Something", "is", "wrong"]) + backend_runner.run_application.return_value = mock_context + + with pytest.raises(Exception, match="Unable to get performance metrics"): + estimate_performance( + ModelInfo(test_tflite_model), + DeviceInfo(device_type="ethos-u55", mac=256), + backend=backend, + ) + + +@pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) +def test_get_generic_runner(backend: str) -> None: + """Test function get_generic_runner().""" + device_info = DeviceInfo("ethos-u55", 256) + + runner = get_generic_runner(device_info=device_info, backend=backend) + assert isinstance(runner, GenericInferenceRunnerEthosU) + + with pytest.raises(RuntimeError): + get_generic_runner(device_info=device_info, backend="UNKNOWN_BACKEND") + + +@pytest.mark.parametrize( + ("backend", "device_type"), + ( + ("Corstone-300", "ethos-u55"), + ("Corstone-300", "ethos-u65"), + ("Corstone-310", "ethos-u55"), + ), +) +def test_backend_support(backend: str, device_type: str) -> None: + """Test backend & device support.""" + assert is_supported(backend) + assert is_supported(backend, device_type) + + assert get_system_name(backend, device_type) + + assert backend in supported_backends() + + +class TestGenericInferenceRunnerEthosU: + """Test for the class GenericInferenceRunnerEthosU.""" + + @staticmethod + @pytest.mark.parametrize( + "device, backend, expected_system, expected_app", + [ + [ + DeviceInfo("ethos-u55", 256), + "Corstone-300", + "Corstone-300: Cortex-M55+Ethos-U55", + "Generic Inference Runner: Ethos-U55", + ], + [ + DeviceInfo("ethos-u65", 256), + "Corstone-300", + "Corstone-300: Cortex-M55+Ethos-U65", + "Generic Inference Runner: Ethos-U65", + ], + [ + DeviceInfo("ethos-u55", 256), + "Corstone-310", + "Corstone-310: Cortex-M85+Ethos-U55", + "Generic Inference Runner: Ethos-U55", + ], + [ + DeviceInfo("ethos-u65", 256), + "Corstone-310", + "Corstone-310: Cortex-M85+Ethos-U65", + "Generic Inference Runner: Ethos-U65", + ], + ], + ) + def test_artifact_resolver( + device: DeviceInfo, backend: str, expected_system: str, expected_app: str + ) -> None: + """Test artifact resolving based on the provided parameters.""" + generic_runner = get_generic_runner(device, backend) + assert isinstance(generic_runner, GenericInferenceRunnerEthosU) + + assert generic_runner.system_name == expected_system + assert generic_runner.app_name == expected_app + + @staticmethod + def test_artifact_resolver_unsupported_backend() -> None: + """Test that it should be not possible to use unsupported backends.""" + with pytest.raises( + RuntimeError, match="Unsupported device ethos-u65 for backend test_backend" + ): + get_generic_runner(DeviceInfo("ethos-u65", 256), "test_backend") + + @staticmethod + @pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) + def test_inference_should_fail_if_system_not_installed( + backend_runner: MagicMock, test_tflite_model: Path, backend: str + ) -> None: + """Test that inference should fail if system is not installed.""" + backend_runner.is_system_installed.return_value = False + + generic_runner = get_generic_runner(DeviceInfo("ethos-u55", 256), backend) + with pytest.raises( + Exception, + match=r"System Corstone-3[01]0: Cortex-M[58]5\+Ethos-U55 is not installed", + ): + generic_runner.run(ModelInfo(test_tflite_model), []) + + @staticmethod + @pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) + def test_inference_should_fail_is_apps_not_installed( + backend_runner: MagicMock, test_tflite_model: Path, backend: str + ) -> None: + """Test that inference should fail if apps are not installed.""" + backend_runner.is_system_installed.return_value = True + backend_runner.is_application_installed.return_value = False + + generic_runner = get_generic_runner(DeviceInfo("ethos-u55", 256), backend) + with pytest.raises( + Exception, + match="Application Generic Inference Runner: Ethos-U55" + r" for the system Corstone-3[01]0: Cortex-M[58]5\+Ethos-U55 is not " + r"installed", + ): + generic_runner.run(ModelInfo(test_tflite_model), []) + + +@pytest.fixture(name="backend_runner") +def fixture_backend_runner(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Mock backend runner.""" + backend_runner_mock = MagicMock(spec=BackendRunner) + monkeypatch.setattr( + "mlia.backend.corstone.performance.get_backend_runner", + MagicMock(return_value=backend_runner_mock), + ) + return backend_runner_mock diff --git a/tests/test_backend_application.py b/tests/test_backend_executor_application.py index 478658b..8962a0a 100644 --- a/tests/test_backend_application.py +++ b/tests/test_backend_executor_application.py @@ -11,20 +11,22 @@ from unittest.mock import MagicMock import pytest -from mlia.backend.application import Application -from mlia.backend.application import get_application -from mlia.backend.application import get_available_application_directory_names -from mlia.backend.application import get_available_applications -from mlia.backend.application import get_unique_application_names -from mlia.backend.application import install_application -from mlia.backend.application import load_applications -from mlia.backend.application import remove_application -from mlia.backend.common import Command -from mlia.backend.common import Param -from mlia.backend.common import UserParamConfig -from mlia.backend.config import ApplicationConfig -from mlia.backend.config import ExtendedApplicationConfig -from mlia.backend.config import NamedExecutionConfig +from mlia.backend.executor.application import Application +from mlia.backend.executor.application import get_application +from mlia.backend.executor.application import ( + get_available_application_directory_names, +) +from mlia.backend.executor.application import get_available_applications +from mlia.backend.executor.application import get_unique_application_names +from mlia.backend.executor.application import install_application +from mlia.backend.executor.application import load_applications +from mlia.backend.executor.application import remove_application +from mlia.backend.executor.common import Command +from mlia.backend.executor.common import Param +from mlia.backend.executor.common import UserParamConfig +from mlia.backend.executor.config import ApplicationConfig +from mlia.backend.executor.config import ExtendedApplicationConfig +from mlia.backend.executor.config import NamedExecutionConfig def test_get_available_application_directory_names() -> None: @@ -151,7 +153,7 @@ def test_install_application( """Test application install from archive.""" mock_create_destination_and_install = MagicMock() monkeypatch.setattr( - "mlia.backend.application.create_destination_and_install", + "mlia.backend.executor.application.create_destination_and_install", mock_create_destination_and_install, ) @@ -163,7 +165,9 @@ def test_install_application( def test_remove_application(monkeypatch: Any) -> None: """Test application removal.""" mock_remove_backend = MagicMock() - monkeypatch.setattr("mlia.backend.application.remove_backend", mock_remove_backend) + monkeypatch.setattr( + "mlia.backend.executor.application.remove_backend", mock_remove_backend + ) remove_application("some_application_directory") mock_remove_backend.assert_called_once() diff --git a/tests/test_backend_common.py b/tests/test_backend_executor_common.py index 4f4853e..e881462 100644 --- a/tests/test_backend_common.py +++ b/tests/test_backend_executor_common.py @@ -14,20 +14,20 @@ from unittest.mock import MagicMock import pytest -from mlia.backend.application import Application -from mlia.backend.common import Backend -from mlia.backend.common import BaseBackendConfig -from mlia.backend.common import Command -from mlia.backend.common import ConfigurationException -from mlia.backend.common import load_config -from mlia.backend.common import Param -from mlia.backend.common import parse_raw_parameter -from mlia.backend.common import remove_backend -from mlia.backend.config import ApplicationConfig -from mlia.backend.config import UserParamConfig -from mlia.backend.execution import ExecutionContext -from mlia.backend.execution import ParamResolver -from mlia.backend.system import System +from mlia.backend.executor.application import Application +from mlia.backend.executor.common import Backend +from mlia.backend.executor.common import BaseBackendConfig +from mlia.backend.executor.common import Command +from mlia.backend.executor.common import ConfigurationException +from mlia.backend.executor.common import load_config +from mlia.backend.executor.common import Param +from mlia.backend.executor.common import parse_raw_parameter +from mlia.backend.executor.common import remove_backend +from mlia.backend.executor.config import ApplicationConfig +from mlia.backend.executor.config import UserParamConfig +from mlia.backend.executor.execution import ExecutionContext +from mlia.backend.executor.execution import ParamResolver +from mlia.backend.executor.system import System @pytest.mark.parametrize( @@ -42,7 +42,9 @@ def test_remove_backend( ) -> None: """Test remove_backend function.""" mock_remove_resource = MagicMock() - monkeypatch.setattr("mlia.backend.common.remove_resource", mock_remove_resource) + monkeypatch.setattr( + "mlia.backend.executor.common.remove_resource", mock_remove_resource + ) with expected_exception: remove_backend(directory_name, "applications") @@ -73,7 +75,7 @@ def test_load_config( ) for config in configs: json_mock = MagicMock() - monkeypatch.setattr("mlia.backend.common.json.load", json_mock) + monkeypatch.setattr("mlia.backend.executor.common.json.load", json_mock) load_config(config) json_mock.assert_called_once() diff --git a/tests/test_backend_execution.py b/tests/test_backend_executor_execution.py index e56a1b0..6a6ea08 100644 --- a/tests/test_backend_execution.py +++ b/tests/test_backend_executor_execution.py @@ -7,16 +7,16 @@ from unittest.mock import MagicMock import pytest -from mlia.backend.application import Application -from mlia.backend.common import UserParamConfig -from mlia.backend.config import ApplicationConfig -from mlia.backend.config import SystemConfig -from mlia.backend.execution import ExecutionContext -from mlia.backend.execution import get_application_and_system -from mlia.backend.execution import get_application_by_name_and_system -from mlia.backend.execution import ParamResolver -from mlia.backend.execution import run_application -from mlia.backend.system import load_system +from mlia.backend.executor.application import Application +from mlia.backend.executor.common import UserParamConfig +from mlia.backend.executor.config import ApplicationConfig +from mlia.backend.executor.config import SystemConfig +from mlia.backend.executor.execution import ExecutionContext +from mlia.backend.executor.execution import get_application_and_system +from mlia.backend.executor.execution import get_application_by_name_and_system +from mlia.backend.executor.execution import ParamResolver +from mlia.backend.executor.execution import run_application +from mlia.backend.executor.system import load_system def test_context_param_resolver(tmpdir: Any) -> None: @@ -181,7 +181,7 @@ def test_context_param_resolver(tmpdir: Any) -> None: def test_get_application_by_name_and_system(monkeypatch: Any) -> None: """Test exceptional case for get_application_by_name_and_system.""" monkeypatch.setattr( - "mlia.backend.execution.get_application", + "mlia.backend.executor.execution.get_application", MagicMock(return_value=[MagicMock(), MagicMock()]), ) @@ -196,7 +196,7 @@ def test_get_application_by_name_and_system(monkeypatch: Any) -> None: def test_get_application_and_system(monkeypatch: Any) -> None: """Test exceptional case for get_application_and_system.""" monkeypatch.setattr( - "mlia.backend.execution.get_system", MagicMock(return_value=None) + "mlia.backend.executor.execution.get_system", MagicMock(return_value=None) ) with pytest.raises(ValueError, match="System test_system is not found"): diff --git a/tests/test_backend_fs.py b/tests/test_backend_executor_fs.py index 292a7cc..298b8db 100644 --- a/tests/test_backend_fs.py +++ b/tests/test_backend_executor_fs.py @@ -10,12 +10,12 @@ from unittest.mock import MagicMock import pytest -from mlia.backend.fs import get_backends_path -from mlia.backend.fs import recreate_directory -from mlia.backend.fs import remove_directory -from mlia.backend.fs import remove_resource -from mlia.backend.fs import ResourceType -from mlia.backend.fs import valid_for_filename +from mlia.backend.executor.fs import get_backends_path +from mlia.backend.executor.fs import recreate_directory +from mlia.backend.executor.fs import remove_directory +from mlia.backend.executor.fs import remove_resource +from mlia.backend.executor.fs import ResourceType +from mlia.backend.executor.fs import valid_for_filename @pytest.mark.parametrize( @@ -39,10 +39,12 @@ def test_remove_resource_wrong_directory( ) -> None: """Test removing resource with wrong directory.""" mock_get_resources = MagicMock(return_value=test_applications_path) - monkeypatch.setattr("mlia.backend.fs.get_backends_path", mock_get_resources) + monkeypatch.setattr( + "mlia.backend.executor.fs.get_backends_path", mock_get_resources + ) mock_shutil_rmtree = MagicMock() - monkeypatch.setattr("mlia.backend.fs.shutil.rmtree", mock_shutil_rmtree) + monkeypatch.setattr("mlia.backend.executor.fs.shutil.rmtree", mock_shutil_rmtree) with pytest.raises(Exception, match="Resource .* does not exist"): remove_resource("unknown", "applications") @@ -56,10 +58,12 @@ def test_remove_resource_wrong_directory( def test_remove_resource(monkeypatch: Any, test_applications_path: Path) -> None: """Test removing resource data.""" mock_get_resources = MagicMock(return_value=test_applications_path) - monkeypatch.setattr("mlia.backend.fs.get_backends_path", mock_get_resources) + monkeypatch.setattr( + "mlia.backend.executor.fs.get_backends_path", mock_get_resources + ) mock_shutil_rmtree = MagicMock() - monkeypatch.setattr("mlia.backend.fs.shutil.rmtree", mock_shutil_rmtree) + monkeypatch.setattr("mlia.backend.executor.fs.shutil.rmtree", mock_shutil_rmtree) remove_resource("application1", "applications") mock_shutil_rmtree.assert_called_once() diff --git a/tests/test_backend_output_consumer.py b/tests/test_backend_executor_output_consumer.py index 2a46787..537084f 100644 --- a/tests/test_backend_output_consumer.py +++ b/tests/test_backend_executor_output_consumer.py @@ -9,8 +9,8 @@ from typing import Any import pytest -from mlia.backend.output_consumer import Base64OutputConsumer -from mlia.backend.output_consumer import OutputConsumer +from mlia.backend.executor.output_consumer import Base64OutputConsumer +from mlia.backend.executor.output_consumer import OutputConsumer OUTPUT_MATCH_ALL = bytearray( diff --git a/tests/test_backend_proc.py b/tests/test_backend_executor_proc.py index d2c2cd4..e8caf8a 100644 --- a/tests/test_backend_proc.py +++ b/tests/test_backend_executor_proc.py @@ -9,14 +9,14 @@ from unittest import mock import pytest from sh import ErrorReturnCode -from mlia.backend.proc import Command -from mlia.backend.proc import CommandFailedException -from mlia.backend.proc import CommandNotFound -from mlia.backend.proc import parse_command -from mlia.backend.proc import print_command_stdout -from mlia.backend.proc import run_and_wait -from mlia.backend.proc import ShellCommand -from mlia.backend.proc import terminate_command +from mlia.backend.executor.proc import Command +from mlia.backend.executor.proc import CommandFailedException +from mlia.backend.executor.proc import CommandNotFound +from mlia.backend.executor.proc import parse_command +from mlia.backend.executor.proc import print_command_stdout +from mlia.backend.executor.proc import run_and_wait +from mlia.backend.executor.proc import ShellCommand +from mlia.backend.executor.proc import terminate_command class TestShellCommand: @@ -136,12 +136,13 @@ class TestRunAndWait: """Init test method.""" self.execute_command_mock = mock.MagicMock() monkeypatch.setattr( - "mlia.backend.proc.execute_command", self.execute_command_mock + "mlia.backend.executor.proc.execute_command", self.execute_command_mock ) self.terminate_command_mock = mock.MagicMock() monkeypatch.setattr( - "mlia.backend.proc.terminate_command", self.terminate_command_mock + "mlia.backend.executor.proc.terminate_command", + self.terminate_command_mock, ) def test_if_execute_command_raises_exception(self) -> None: diff --git a/tests/test_backend_executor_runner.py b/tests/test_backend_executor_runner.py new file mode 100644 index 0000000..36c6e5e --- /dev/null +++ b/tests/test_backend_executor_runner.py @@ -0,0 +1,254 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for module backend/manager.""" +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock +from unittest.mock import PropertyMock + +import pytest + +from mlia.backend.corstone.performance import BackendRunner +from mlia.backend.corstone.performance import ExecutionParams + + +class TestBackendRunner: + """Tests for BackendRunner class.""" + + @staticmethod + def _setup_backends( + monkeypatch: pytest.MonkeyPatch, + available_systems: list[str] | None = None, + available_apps: list[str] | None = None, + ) -> None: + """Set up backend metadata.""" + + def mock_system(system: str) -> MagicMock: + """Mock the System instance.""" + mock = MagicMock() + type(mock).name = PropertyMock(return_value=system) + return mock + + def mock_app(app: str) -> MagicMock: + """Mock the Application instance.""" + mock = MagicMock() + type(mock).name = PropertyMock(return_value=app) + mock.can_run_on.return_value = True + return mock + + system_mocks = [mock_system(name) for name in (available_systems or [])] + monkeypatch.setattr( + "mlia.backend.executor.runner.get_available_systems", + MagicMock(return_value=system_mocks), + ) + + apps_mock = [mock_app(name) for name in (available_apps or [])] + monkeypatch.setattr( + "mlia.backend.executor.runner.get_available_applications", + MagicMock(return_value=apps_mock), + ) + + @pytest.mark.parametrize( + "available_systems, system, installed", + [ + ([], "system1", False), + (["system1", "system2"], "system1", True), + ], + ) + def test_is_system_installed( + self, + available_systems: list, + system: str, + installed: bool, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test method is_system_installed.""" + backend_runner = BackendRunner() + + self._setup_backends(monkeypatch, available_systems) + + assert backend_runner.is_system_installed(system) == installed + + @pytest.mark.parametrize( + "available_systems, systems", + [ + ([], []), + (["system1"], ["system1"]), + ], + ) + def test_installed_systems( + self, + available_systems: list[str], + systems: list[str], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test method installed_systems.""" + backend_runner = BackendRunner() + + self._setup_backends(monkeypatch, available_systems) + assert backend_runner.get_installed_systems() == systems + + @staticmethod + def test_install_system(monkeypatch: pytest.MonkeyPatch) -> None: + """Test system installation.""" + install_system_mock = MagicMock() + monkeypatch.setattr( + "mlia.backend.executor.runner.install_system", install_system_mock + ) + + backend_runner = BackendRunner() + backend_runner.install_system(Path("test_system_path")) + + install_system_mock.assert_called_once_with(Path("test_system_path")) + + @pytest.mark.parametrize( + "available_systems, systems, expected_result", + [ + ([], [], False), + (["system1"], [], False), + (["system1"], ["system1"], True), + (["system1", "system2"], ["system1", "system3"], False), + (["system1", "system2"], ["system1", "system2"], True), + ], + ) + def test_systems_installed( + self, + available_systems: list[str], + systems: list[str], + expected_result: bool, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test method systems_installed.""" + self._setup_backends(monkeypatch, available_systems) + + backend_runner = BackendRunner() + + assert backend_runner.systems_installed(systems) is expected_result + + @pytest.mark.parametrize( + "available_apps, applications, expected_result", + [ + ([], [], False), + (["app1"], [], False), + (["app1"], ["app1"], True), + (["app1", "app2"], ["app1", "app3"], False), + (["app1", "app2"], ["app1", "app2"], True), + ], + ) + def test_applications_installed( + self, + available_apps: list[str], + applications: list[str], + expected_result: bool, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test method applications_installed.""" + self._setup_backends(monkeypatch, [], available_apps) + backend_runner = BackendRunner() + + assert backend_runner.applications_installed(applications) is expected_result + + @pytest.mark.parametrize( + "available_apps, applications", + [ + ([], []), + ( + ["application1", "application2"], + ["application1", "application2"], + ), + ], + ) + def test_get_installed_applications( + self, + available_apps: list[str], + applications: list[str], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test method get_installed_applications.""" + self._setup_backends(monkeypatch, [], available_apps) + + backend_runner = BackendRunner() + assert applications == backend_runner.get_installed_applications() + + @staticmethod + def test_install_application(monkeypatch: pytest.MonkeyPatch) -> None: + """Test application installation.""" + mock_install_application = MagicMock() + monkeypatch.setattr( + "mlia.backend.executor.runner.install_application", + mock_install_application, + ) + + backend_runner = BackendRunner() + backend_runner.install_application(Path("test_application_path")) + mock_install_application.assert_called_once_with(Path("test_application_path")) + + @pytest.mark.parametrize( + "available_apps, application, installed", + [ + ([], "system1", False), + ( + ["application1", "application2"], + "application1", + True, + ), + ( + [], + "application1", + False, + ), + ], + ) + def test_is_application_installed( + self, + available_apps: list[str], + application: str, + installed: bool, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Test method is_application_installed.""" + self._setup_backends(monkeypatch, [], available_apps) + + backend_runner = BackendRunner() + assert installed == backend_runner.is_application_installed( + application, "system1" + ) + + @staticmethod + @pytest.mark.parametrize( + "execution_params, expected_command", + [ + ( + ExecutionParams("application_4", "System 4", [], []), + ["application_4", [], "System 4", []], + ), + ( + ExecutionParams( + "application_6", + "System 6", + ["param1=value2"], + ["sys-param1=value2"], + ), + [ + "application_6", + ["param1=value2"], + "System 6", + ["sys-param1=value2"], + ], + ), + ], + ) + def test_run_application_local( + monkeypatch: pytest.MonkeyPatch, + execution_params: ExecutionParams, + expected_command: list[str], + ) -> None: + """Test method run_application with local systems.""" + run_app = MagicMock() + monkeypatch.setattr("mlia.backend.executor.runner.run_application", run_app) + + backend_runner = BackendRunner() + backend_runner.run_application(execution_params) + + run_app.assert_called_once_with(*expected_command) diff --git a/tests/test_backend_source.py b/tests/test_backend_executor_source.py index c6ef26f..3aa336e 100644 --- a/tests/test_backend_source.py +++ b/tests/test_backend_executor_source.py @@ -10,11 +10,11 @@ from unittest.mock import patch import pytest -from mlia.backend.common import ConfigurationException -from mlia.backend.source import create_destination_and_install -from mlia.backend.source import DirectorySource -from mlia.backend.source import get_source -from mlia.backend.source import TarArchiveSource +from mlia.backend.executor.common import ConfigurationException +from mlia.backend.executor.source import create_destination_and_install +from mlia.backend.executor.source import DirectorySource +from mlia.backend.executor.source import get_source +from mlia.backend.executor.source import TarArchiveSource def test_create_destination_and_install(test_systems_path: Path, tmpdir: Any) -> None: @@ -27,7 +27,10 @@ def test_create_destination_and_install(test_systems_path: Path, tmpdir: Any) -> assert (resources / "system1").is_dir() -@patch("mlia.backend.source.DirectorySource.create_destination", return_value=False) +@patch( + "mlia.backend.executor.source.DirectorySource.create_destination", + return_value=False, +) def test_create_destination_and_install_if_dest_creation_not_required( mock_ds_create_destination: Any, tmpdir: Any ) -> None: diff --git a/tests/test_backend_system.py b/tests/test_backend_executor_system.py index ecc149d..c94ef30 100644 --- a/tests/test_backend_system.py +++ b/tests/test_backend_executor_system.py @@ -10,17 +10,17 @@ from unittest.mock import MagicMock import pytest -from mlia.backend.common import Command -from mlia.backend.common import ConfigurationException -from mlia.backend.common import Param -from mlia.backend.common import UserParamConfig -from mlia.backend.config import SystemConfig -from mlia.backend.system import get_available_systems -from mlia.backend.system import get_system -from mlia.backend.system import install_system -from mlia.backend.system import load_system -from mlia.backend.system import remove_system -from mlia.backend.system import System +from mlia.backend.executor.common import Command +from mlia.backend.executor.common import ConfigurationException +from mlia.backend.executor.common import Param +from mlia.backend.executor.common import UserParamConfig +from mlia.backend.executor.config import SystemConfig +from mlia.backend.executor.system import get_available_systems +from mlia.backend.executor.system import get_system +from mlia.backend.executor.system import install_system +from mlia.backend.executor.system import load_system +from mlia.backend.executor.system import remove_system +from mlia.backend.executor.system import System def test_get_available_systems() -> None: @@ -95,7 +95,7 @@ def test_install_system( """Test system installation from archive.""" mock_create_destination_and_install = MagicMock() monkeypatch.setattr( - "mlia.backend.system.create_destination_and_install", + "mlia.backend.executor.system.create_destination_and_install", mock_create_destination_and_install, ) @@ -108,7 +108,9 @@ def test_install_system( def test_remove_system(monkeypatch: Any) -> None: """Test system removal.""" mock_remove_backend = MagicMock() - monkeypatch.setattr("mlia.backend.system.remove_backend", mock_remove_backend) + monkeypatch.setattr( + "mlia.backend.executor.system.remove_backend", mock_remove_backend + ) remove_system("some_system_dir") mock_remove_backend.assert_called_once() diff --git a/tests/test_backend_install.py b/tests/test_backend_install.py new file mode 100644 index 0000000..024a833 --- /dev/null +++ b/tests/test_backend_install.py @@ -0,0 +1,124 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for common management functionality.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from mlia.backend.install import BackendInfo +from mlia.backend.install import get_all_application_names +from mlia.backend.install import get_all_system_names +from mlia.backend.install import get_system_name +from mlia.backend.install import is_supported +from mlia.backend.install import StaticPathChecker +from mlia.backend.install import supported_backends + + +@pytest.mark.parametrize( + "copy_source, system_config", + [ + (True, "system_config.json"), + (True, None), + (False, "system_config.json"), + (False, None), + ], +) +def test_static_path_checker( + tmp_path: Path, copy_source: bool, system_config: str +) -> None: + """Test static path checker.""" + checker = StaticPathChecker(tmp_path, ["file1.txt"], copy_source, system_config) + tmp_path.joinpath("file1.txt").touch() + + result = checker(tmp_path) + assert result == BackendInfo(tmp_path, copy_source, system_config) + + +def test_static_path_checker_invalid_path(tmp_path: Path) -> None: + """Test static path checker with invalid path.""" + checker = StaticPathChecker(tmp_path, ["file1.txt"]) + + result = checker(tmp_path) + assert result is None + + result = checker(tmp_path / "unknown_directory") + assert result is None + + +def test_supported_backends() -> None: + """Test function supported backends.""" + assert supported_backends() == ["Corstone-300", "Corstone-310"] + + +@pytest.mark.parametrize( + "backend, expected_result", + [ + ["unknown_backend", False], + ["Corstone-300", True], + ["Corstone-310", True], + ], +) +def test_is_supported(backend: str, expected_result: bool) -> None: + """Test function is_supported.""" + assert is_supported(backend) == expected_result + + +@pytest.mark.parametrize( + "backend, expected_result", + [ + [ + "Corstone-300", + [ + "Corstone-300: Cortex-M55+Ethos-U55", + "Corstone-300: Cortex-M55+Ethos-U65", + ], + ], + [ + "Corstone-310", + [ + "Corstone-310: Cortex-M85+Ethos-U55", + "Corstone-310: Cortex-M85+Ethos-U65", + ], + ], + ], +) +def test_get_all_system_names(backend: str, expected_result: list[str]) -> None: + """Test function get_all_system_names.""" + assert sorted(get_all_system_names(backend)) == expected_result + + +@pytest.mark.parametrize( + "backend, expected_result", + [ + [ + "Corstone-300", + [ + "Generic Inference Runner: Ethos-U55", + "Generic Inference Runner: Ethos-U65", + ], + ], + [ + "Corstone-310", + [ + "Generic Inference Runner: Ethos-U55", + "Generic Inference Runner: Ethos-U65", + ], + ], + ], +) +def test_get_all_application_names(backend: str, expected_result: list[str]) -> None: + """Test function get_all_application_names.""" + assert sorted(get_all_application_names(backend)) == expected_result + + +def test_get_system_name() -> None: + """Test function get_system_name.""" + assert ( + get_system_name("Corstone-300", "ethos-u55") + == "Corstone-300: Cortex-M55+Ethos-U55" + ) + + with pytest.raises(KeyError): + get_system_name("some_backend", "some_type") diff --git a/tests/test_backend_manager.py b/tests/test_backend_manager.py index dfbcdaa..19cb357 100644 --- a/tests/test_backend_manager.py +++ b/tests/test_backend_manager.py @@ -1,758 +1,282 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 -"""Tests for module backend/manager.""" +"""Tests for installation manager.""" from __future__ import annotations -import base64 -import json -from contextlib import ExitStack as does_not_raise from pathlib import Path from typing import Any +from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import PropertyMock import pytest -from mlia.backend.application import get_application -from mlia.backend.execution import ExecutionContext -from mlia.backend.manager import BackendRunner -from mlia.backend.manager import DeviceInfo -from mlia.backend.manager import estimate_performance -from mlia.backend.manager import ExecutionParams -from mlia.backend.manager import GenericInferenceOutputParser -from mlia.backend.manager import GenericInferenceRunnerEthosU -from mlia.backend.manager import get_generic_runner -from mlia.backend.manager import get_system_name -from mlia.backend.manager import is_supported -from mlia.backend.manager import ModelInfo -from mlia.backend.manager import PerformanceMetrics -from mlia.backend.manager import supported_backends -from mlia.backend.output_consumer import Base64OutputConsumer -from mlia.backend.system import get_system - - -def _mock_encode_b64(data: dict[str, int]) -> str: - """ - Encode the given data into a mock base64-encoded string of JSON. - - This reproduces the base64 encoding done in the Corstone applications. - - JSON example: - - ```json - [{'count': 1, - 'profiling_group': 'Inference', - 'samples': [{'name': 'NPU IDLE', 'value': [612]}, - {'name': 'NPU AXI0_RD_DATA_BEAT_RECEIVED', 'value': [165872]}, - {'name': 'NPU AXI0_WR_DATA_BEAT_WRITTEN', 'value': [88712]}, - {'name': 'NPU AXI1_RD_DATA_BEAT_RECEIVED', 'value': [57540]}, - {'name': 'NPU ACTIVE', 'value': [520489]}, - {'name': 'NPU TOTAL', 'value': [521101]}]}] - ``` - """ - wrapped_data = [ - { - "count": 1, - "profiling_group": "Inference", - "samples": [ - {"name": name, "value": [value]} for name, value in data.items() - ], - } - ] - json_str = json.dumps(wrapped_data) - json_bytes = bytearray(json_str, encoding="utf-8") - json_b64 = base64.b64encode(json_bytes).decode("utf-8") - tag = Base64OutputConsumer.TAG_NAME - return f"<{tag}>{json_b64}</{tag}>" +from mlia.backend.install import DownloadAndInstall +from mlia.backend.install import Installation +from mlia.backend.install import InstallationType +from mlia.backend.install import InstallFromPath +from mlia.backend.manager import DefaultInstallationManager -@pytest.mark.parametrize( - "data, is_ready, result, missed_keys", - [ - ( - [], - False, - {}, - { - "npu_active_cycles", - "npu_axi0_rd_data_beat_received", - "npu_axi0_wr_data_beat_written", - "npu_axi1_rd_data_beat_received", - "npu_idle_cycles", - "npu_total_cycles", - }, - ), - ( - ["sample text"], - False, - {}, - { - "npu_active_cycles", - "npu_axi0_rd_data_beat_received", - "npu_axi0_wr_data_beat_written", - "npu_axi1_rd_data_beat_received", - "npu_idle_cycles", - "npu_total_cycles", - }, - ), - ( - [_mock_encode_b64({"NPU AXI0_RD_DATA_BEAT_RECEIVED": 123})], - False, - {"npu_axi0_rd_data_beat_received": 123}, - { - "npu_active_cycles", - "npu_axi0_wr_data_beat_written", - "npu_axi1_rd_data_beat_received", - "npu_idle_cycles", - "npu_total_cycles", - }, - ), - ( - [ - _mock_encode_b64( - { - "NPU AXI0_RD_DATA_BEAT_RECEIVED": 1, - "NPU AXI0_WR_DATA_BEAT_WRITTEN": 2, - "NPU AXI1_RD_DATA_BEAT_RECEIVED": 3, - "NPU ACTIVE": 4, - "NPU IDLE": 5, - "NPU TOTAL": 6, - } - ) - ], - True, - { - "npu_axi0_rd_data_beat_received": 1, - "npu_axi0_wr_data_beat_written": 2, - "npu_axi1_rd_data_beat_received": 3, - "npu_active_cycles": 4, - "npu_idle_cycles": 5, - "npu_total_cycles": 6, - }, - set(), - ), - ], -) -def test_generic_inference_output_parser( - data: dict[str, int], is_ready: bool, result: dict, missed_keys: set[str] -) -> None: - """Test generic runner output parser.""" - parser = GenericInferenceOutputParser() - - for line in data: - parser.feed(line) - - assert parser.is_ready() == is_ready - assert parser.result == result - assert parser.missed_keys() == missed_keys - - -class TestBackendRunner: - """Tests for BackendRunner class.""" - - @staticmethod - def _setup_backends( - monkeypatch: pytest.MonkeyPatch, - available_systems: list[str] | None = None, - available_apps: list[str] | None = None, - ) -> None: - """Set up backend metadata.""" - - def mock_system(system: str) -> MagicMock: - """Mock the System instance.""" - mock = MagicMock() - type(mock).name = PropertyMock(return_value=system) - return mock - - def mock_app(app: str) -> MagicMock: - """Mock the Application instance.""" - mock = MagicMock() - type(mock).name = PropertyMock(return_value=app) - mock.can_run_on.return_value = True - return mock - - system_mocks = [mock_system(name) for name in (available_systems or [])] - monkeypatch.setattr( - "mlia.backend.manager.get_available_systems", - MagicMock(return_value=system_mocks), - ) +def get_default_installation_manager_mock( + name: str, + already_installed: bool = False, +) -> MagicMock: + """Get mock instance for DefaultInstallationManager.""" + mock = MagicMock(spec=DefaultInstallationManager) - apps_mock = [mock_app(name) for name in (available_apps or [])] - monkeypatch.setattr( - "mlia.backend.manager.get_available_applications", - MagicMock(return_value=apps_mock), - ) + props = { + "name": name, + "already_installed": already_installed, + } + for prop, value in props.items(): + setattr(type(mock), prop, PropertyMock(return_value=value)) - @pytest.mark.parametrize( - "available_systems, system, installed", - [ - ([], "system1", False), - (["system1", "system2"], "system1", True), - ], - ) - def test_is_system_installed( - self, - available_systems: list, - system: str, - installed: bool, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Test method is_system_installed.""" - backend_runner = BackendRunner() - - self._setup_backends(monkeypatch, available_systems) - - assert backend_runner.is_system_installed(system) == installed - - @pytest.mark.parametrize( - "available_systems, systems", - [ - ([], []), - (["system1"], ["system1"]), - ], - ) - def test_installed_systems( - self, - available_systems: list[str], - systems: list[str], - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Test method installed_systems.""" - backend_runner = BackendRunner() - - self._setup_backends(monkeypatch, available_systems) - assert backend_runner.get_installed_systems() == systems - - @staticmethod - def test_install_system(monkeypatch: pytest.MonkeyPatch) -> None: - """Test system installation.""" - install_system_mock = MagicMock() - monkeypatch.setattr("mlia.backend.manager.install_system", install_system_mock) - - backend_runner = BackendRunner() - backend_runner.install_system(Path("test_system_path")) - - install_system_mock.assert_called_once_with(Path("test_system_path")) - - @pytest.mark.parametrize( - "available_systems, systems, expected_result", - [ - ([], [], False), - (["system1"], [], False), - (["system1"], ["system1"], True), - (["system1", "system2"], ["system1", "system3"], False), - (["system1", "system2"], ["system1", "system2"], True), - ], - ) - def test_systems_installed( - self, - available_systems: list[str], - systems: list[str], - expected_result: bool, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Test method systems_installed.""" - self._setup_backends(monkeypatch, available_systems) - - backend_runner = BackendRunner() - - assert backend_runner.systems_installed(systems) is expected_result - - @pytest.mark.parametrize( - "available_apps, applications, expected_result", - [ - ([], [], False), - (["app1"], [], False), - (["app1"], ["app1"], True), - (["app1", "app2"], ["app1", "app3"], False), - (["app1", "app2"], ["app1", "app2"], True), - ], - ) - def test_applications_installed( - self, - available_apps: list[str], - applications: list[str], - expected_result: bool, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Test method applications_installed.""" - self._setup_backends(monkeypatch, [], available_apps) - backend_runner = BackendRunner() - - assert backend_runner.applications_installed(applications) is expected_result - - @pytest.mark.parametrize( - "available_apps, applications", - [ - ([], []), - ( - ["application1", "application2"], - ["application1", "application2"], - ), - ], - ) - def test_get_installed_applications( - self, - available_apps: list[str], - applications: list[str], - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Test method get_installed_applications.""" - self._setup_backends(monkeypatch, [], available_apps) - - backend_runner = BackendRunner() - assert applications == backend_runner.get_installed_applications() - - @staticmethod - def test_install_application(monkeypatch: pytest.MonkeyPatch) -> None: - """Test application installation.""" - mock_install_application = MagicMock() - monkeypatch.setattr( - "mlia.backend.manager.install_application", mock_install_application - ) + return mock - backend_runner = BackendRunner() - backend_runner.install_application(Path("test_application_path")) - mock_install_application.assert_called_once_with(Path("test_application_path")) - @pytest.mark.parametrize( - "available_apps, application, installed", - [ - ([], "system1", False), - ( - ["application1", "application2"], - "application1", - True, - ), - ( - [], - "application1", - False, - ), - ], +def _ready_for_uninstall_mock() -> MagicMock: + return get_default_installation_manager_mock( + name="already_installed", + already_installed=True, ) - def test_is_application_installed( - self, - available_apps: list[str], - application: str, - installed: bool, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - """Test method is_application_installed.""" - self._setup_backends(monkeypatch, [], available_apps) - - backend_runner = BackendRunner() - assert installed == backend_runner.is_application_installed( - application, "system1" - ) - @staticmethod - @pytest.mark.parametrize( - "execution_params, expected_command", - [ - ( - ExecutionParams("application_4", "System 4", [], []), - ["application_4", [], "System 4", []], - ), - ( - ExecutionParams( - "application_6", - "System 6", - ["param1=value2"], - ["sys-param1=value2"], - ), - [ - "application_6", - ["param1=value2"], - "System 6", - ["sys-param1=value2"], - ], - ), - ], - ) - def test_run_application_local( - monkeypatch: pytest.MonkeyPatch, - execution_params: ExecutionParams, - expected_command: list[str], - ) -> None: - """Test method run_application with local systems.""" - run_app = MagicMock() - monkeypatch.setattr("mlia.backend.manager.run_application", run_app) - backend_runner = BackendRunner() - backend_runner.run_application(execution_params) +def get_installation_mock( + name: str, + already_installed: bool = False, + could_be_installed: bool = False, + supported_install_type: type | tuple | None = None, +) -> MagicMock: + """Get mock instance for the installation.""" + mock = MagicMock(spec=Installation) - run_app.assert_called_once_with(*expected_command) + def supports(install_type: InstallationType) -> bool: + if supported_install_type is None: + return False + return isinstance(install_type, supported_install_type) -@pytest.mark.parametrize( - "device, system, application, backend, expected_error", - [ - ( - DeviceInfo(device_type="ethos-u55", mac=32), - ("Corstone-300: Cortex-M55+Ethos-U55", True), - ("Generic Inference Runner: Ethos-U55", True), - "Corstone-300", - does_not_raise(), - ), - ( - DeviceInfo(device_type="ethos-u55", mac=32), - ("Corstone-300: Cortex-M55+Ethos-U55", False), - ("Generic Inference Runner: Ethos-U55", False), - "Corstone-300", - pytest.raises( - Exception, - match=r"System Corstone-300: Cortex-M55\+Ethos-U55 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u55", mac=32), - ("Corstone-300: Cortex-M55+Ethos-U55", True), - ("Generic Inference Runner: Ethos-U55", False), - "Corstone-300", - pytest.raises( - Exception, - match=r"Application Generic Inference Runner: Ethos-U55 " - r"for the system Corstone-300: Cortex-M55\+Ethos-U55 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u55", mac=32), - ("Corstone-310: Cortex-M85+Ethos-U55", True), - ("Generic Inference Runner: Ethos-U55", True), - "Corstone-310", - does_not_raise(), - ), - ( - DeviceInfo(device_type="ethos-u55", mac=32), - ("Corstone-310: Cortex-M85+Ethos-U55", False), - ("Generic Inference Runner: Ethos-U55", False), - "Corstone-310", - pytest.raises( - Exception, - match=r"System Corstone-310: Cortex-M85\+Ethos-U55 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u55", mac=32), - ("Corstone-310: Cortex-M85+Ethos-U55", True), - ("Generic Inference Runner: Ethos-U55", False), - "Corstone-310", - pytest.raises( - Exception, - match=r"Application Generic Inference Runner: Ethos-U55 " - r"for the system Corstone-310: Cortex-M85\+Ethos-U55 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u65", mac=512), - ("Corstone-300: Cortex-M55+Ethos-U65", True), - ("Generic Inference Runner: Ethos-U65", True), - "Corstone-300", - does_not_raise(), - ), - ( - DeviceInfo(device_type="ethos-u65", mac=512), - ("Corstone-300: Cortex-M55+Ethos-U65", False), - ("Generic Inference Runner: Ethos-U65", False), - "Corstone-300", - pytest.raises( - Exception, - match=r"System Corstone-300: Cortex-M55\+Ethos-U65 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u65", mac=512), - ("Corstone-300: Cortex-M55+Ethos-U65", True), - ("Generic Inference Runner: Ethos-U65", False), - "Corstone-300", - pytest.raises( - Exception, - match=r"Application Generic Inference Runner: Ethos-U65 " - r"for the system Corstone-300: Cortex-M55\+Ethos-U65 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u65", mac=512), - ("Corstone-310: Cortex-M85+Ethos-U65", True), - ("Generic Inference Runner: Ethos-U65", True), - "Corstone-310", - does_not_raise(), - ), - ( - DeviceInfo(device_type="ethos-u65", mac=512), - ("Corstone-310: Cortex-M85+Ethos-U65", False), - ("Generic Inference Runner: Ethos-U65", False), - "Corstone-310", - pytest.raises( - Exception, - match=r"System Corstone-310: Cortex-M85\+Ethos-U65 is not installed", - ), - ), - ( - DeviceInfo(device_type="ethos-u65", mac=512), - ("Corstone-310: Cortex-M85+Ethos-U65", True), - ("Generic Inference Runner: Ethos-U65", False), - "Corstone-310", - pytest.raises( - Exception, - match=r"Application Generic Inference Runner: Ethos-U65 " - r"for the system Corstone-310: Cortex-M85\+Ethos-U65 is not installed", - ), - ), - ( - DeviceInfo( - device_type="unknown_device", # type: ignore - mac=None, # type: ignore - ), - ("some_system", False), - ("some_application", False), - "some backend", - pytest.raises(Exception, match="Unsupported device unknown_device"), - ), - ], -) -def test_estimate_performance( - device: DeviceInfo, - system: tuple[str, bool], - application: tuple[str, bool], - backend: str, - expected_error: Any, - test_tflite_model: Path, - backend_runner: MagicMock, -) -> None: - """Test getting performance estimations.""" - system_name, system_installed = system - application_name, application_installed = application + mock.supports.side_effect = supports - backend_runner.is_system_installed.return_value = system_installed - backend_runner.is_application_installed.return_value = application_installed + props = { + "name": name, + "already_installed": already_installed, + "could_be_installed": could_be_installed, + } + for prop, value in props.items(): + setattr(type(mock), prop, PropertyMock(return_value=value)) - mock_context = create_mock_context( - [ - _mock_encode_b64( - { - "NPU AXI0_RD_DATA_BEAT_RECEIVED": 1, - "NPU AXI0_WR_DATA_BEAT_WRITTEN": 2, - "NPU AXI1_RD_DATA_BEAT_RECEIVED": 3, - "NPU ACTIVE": 4, - "NPU IDLE": 5, - "NPU TOTAL": 6, - } - ) - ] - ) + return mock - backend_runner.run_application.return_value = mock_context - with expected_error: - perf_metrics = estimate_performance( - ModelInfo(test_tflite_model), device, backend - ) +def _already_installed_mock() -> MagicMock: + return get_installation_mock( + name="already_installed", + already_installed=True, + supported_install_type=(DownloadAndInstall, InstallFromPath), + ) - assert isinstance(perf_metrics, PerformanceMetrics) - assert perf_metrics == PerformanceMetrics( - npu_axi0_rd_data_beat_received=1, - npu_axi0_wr_data_beat_written=2, - npu_axi1_rd_data_beat_received=3, - npu_active_cycles=4, - npu_idle_cycles=5, - npu_total_cycles=6, - ) - assert backend_runner.is_system_installed.called_once_with(system_name) - assert backend_runner.is_application_installed.called_once_with( - application_name, system_name - ) +def _ready_for_installation_mock() -> MagicMock: + return get_installation_mock( + name="ready_for_installation", + already_installed=False, + could_be_installed=True, + ) -@pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) -def test_estimate_performance_insufficient_data( - backend_runner: MagicMock, test_tflite_model: Path, backend: str -) -> None: - """Test that performance could not be estimated when not all data presented.""" - backend_runner.is_system_installed.return_value = True - backend_runner.is_application_installed.return_value = True - - no_total_cycles_output = { - "NPU AXI0_RD_DATA_BEAT_RECEIVED": 1, - "NPU AXI0_WR_DATA_BEAT_WRITTEN": 2, - "NPU AXI1_RD_DATA_BEAT_RECEIVED": 3, - "NPU ACTIVE": 4, - "NPU IDLE": 5, - } - mock_context = create_mock_context([_mock_encode_b64(no_total_cycles_output)]) +def _could_be_downloaded_and_installed_mock() -> MagicMock: + return get_installation_mock( + name="could_be_downloaded_and_installed", + already_installed=False, + could_be_installed=True, + supported_install_type=DownloadAndInstall, + ) - backend_runner.run_application.return_value = mock_context - with pytest.raises( - Exception, match="Unable to get performance metrics, insufficient data" - ): - device = DeviceInfo(device_type="ethos-u55", mac=32) - estimate_performance(ModelInfo(test_tflite_model), device, backend) +def _could_be_installed_from_mock() -> MagicMock: + return get_installation_mock( + name="could_be_installed_from", + already_installed=False, + could_be_installed=True, + supported_install_type=InstallFromPath, + ) -@pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) -def test_estimate_performance_invalid_output( - test_tflite_model: Path, backend_runner: MagicMock, backend: str -) -> None: - """Test estimation could not be done if inference produces unexpected output.""" - backend_runner.is_system_installed.return_value = True - backend_runner.is_application_installed.return_value = True - - mock_context = create_mock_context(["Something", "is", "wrong"]) - backend_runner.run_application.return_value = mock_context - - with pytest.raises(Exception, match="Unable to get performance metrics"): - estimate_performance( - ModelInfo(test_tflite_model), - DeviceInfo(device_type="ethos-u55", mac=256), - backend=backend, +def get_installation_manager( + noninteractive: bool, + installations: list[Any], + monkeypatch: pytest.MonkeyPatch, + yes_response: bool = True, +) -> DefaultInstallationManager: + """Get installation manager instance.""" + if not noninteractive: + monkeypatch.setattr( + "mlia.backend.manager.yes", MagicMock(return_value=yes_response) ) + return DefaultInstallationManager(installations, noninteractive=noninteractive) -def create_mock_process(stdout: list[str], stderr: list[str]) -> MagicMock: - """Mock underlying process.""" - mock_process = MagicMock() - mock_process.poll.return_value = 0 - type(mock_process).stdout = PropertyMock(return_value=iter(stdout)) - type(mock_process).stderr = PropertyMock(return_value=iter(stderr)) - return mock_process +def test_installation_manager_filtering() -> None: + """Test default installation manager.""" + already_installed = _already_installed_mock() + ready_for_installation = _ready_for_installation_mock() + could_be_downloaded_and_installed = _could_be_downloaded_and_installed_mock() -def create_mock_context(stdout: list[str]) -> ExecutionContext: - """Mock ExecutionContext.""" - ctx = ExecutionContext( - app=get_application("application_1")[0], - app_params=[], - system=get_system("System 1"), - system_params=[], + manager = DefaultInstallationManager( + [ + already_installed, + ready_for_installation, + could_be_downloaded_and_installed, + ] ) - ctx.stdout = bytearray("\n".join(stdout).encode("utf-8")) - return ctx + assert manager.already_installed("already_installed") == [already_installed] + assert manager.ready_for_installation() == [ + ready_for_installation, + could_be_downloaded_and_installed, + ] -@pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) -def test_get_generic_runner(backend: str) -> None: - """Test function get_generic_runner().""" - device_info = DeviceInfo("ethos-u55", 256) +@pytest.mark.parametrize("noninteractive", [True, False]) +@pytest.mark.parametrize( + "install_mock, eula_agreement, backend_name, force, expected_call", + [ + [ + _could_be_downloaded_and_installed_mock(), + True, + "could_be_downloaded_and_installed", + False, + [call(DownloadAndInstall(eula_agreement=True))], + ], + [ + _could_be_downloaded_and_installed_mock(), + False, + "could_be_downloaded_and_installed", + True, + [call(DownloadAndInstall(eula_agreement=False))], + ], + [ + _already_installed_mock(), + False, + "already_installed", + True, + [call(DownloadAndInstall(eula_agreement=False))], + ], + [ + _could_be_downloaded_and_installed_mock(), + False, + "unknown", + True, + [], + ], + ], +) +def test_installation_manager_download_and_install( + install_mock: MagicMock, + noninteractive: bool, + eula_agreement: bool, + backend_name: str, + force: bool, + expected_call: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test installation process.""" + install_mock.reset_mock() - runner = get_generic_runner(device_info=device_info, backend=backend) - assert isinstance(runner, GenericInferenceRunnerEthosU) + manager = get_installation_manager(noninteractive, [install_mock], monkeypatch) - with pytest.raises(RuntimeError): - get_generic_runner(device_info=device_info, backend="UNKNOWN_BACKEND") + manager.download_and_install( + backend_name, eula_agreement=eula_agreement, force=force + ) + assert install_mock.install.mock_calls == expected_call + if force and install_mock.already_installed: + install_mock.uninstall.assert_called_once() + else: + install_mock.uninstall.assert_not_called() + +@pytest.mark.parametrize("noninteractive", [True, False]) @pytest.mark.parametrize( - ("backend", "device_type"), - ( - ("Corstone-300", "ethos-u55"), - ("Corstone-300", "ethos-u65"), - ("Corstone-310", "ethos-u55"), - ), + "install_mock, backend_name, force, expected_call", + [ + [ + _could_be_installed_from_mock(), + "could_be_installed_from", + False, + [call(InstallFromPath(Path("some_path")))], + ], + [ + _could_be_installed_from_mock(), + "unknown", + False, + [], + ], + [ + _could_be_installed_from_mock(), + "unknown", + True, + [], + ], + [ + _already_installed_mock(), + "already_installed", + False, + [], + ], + [ + _already_installed_mock(), + "already_installed", + True, + [call(InstallFromPath(Path("some_path")))], + ], + ], ) -def test_backend_support(backend: str, device_type: str) -> None: - """Test backend & device support.""" - assert is_supported(backend) - assert is_supported(backend, device_type) - - assert get_system_name(backend, device_type) +def test_installation_manager_install_from( + install_mock: MagicMock, + noninteractive: bool, + backend_name: str, + force: bool, + expected_call: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test installation process.""" + install_mock.reset_mock() - assert backend in supported_backends() + manager = get_installation_manager(noninteractive, [install_mock], monkeypatch) + manager.install_from(Path("some_path"), backend_name, force=force) + assert install_mock.install.mock_calls == expected_call + if force and install_mock.already_installed: + install_mock.uninstall.assert_called_once() + else: + install_mock.uninstall.assert_not_called() -class TestGenericInferenceRunnerEthosU: - """Test for the class GenericInferenceRunnerEthosU.""" - @staticmethod - @pytest.mark.parametrize( - "device, backend, expected_system, expected_app", +@pytest.mark.parametrize("noninteractive", [True, False]) +@pytest.mark.parametrize( + "install_mock, backend_name, expected_call", + [ [ - [ - DeviceInfo("ethos-u55", 256), - "Corstone-300", - "Corstone-300: Cortex-M55+Ethos-U55", - "Generic Inference Runner: Ethos-U55", - ], - [ - DeviceInfo("ethos-u65", 256), - "Corstone-300", - "Corstone-300: Cortex-M55+Ethos-U65", - "Generic Inference Runner: Ethos-U65", - ], - [ - DeviceInfo("ethos-u55", 256), - "Corstone-310", - "Corstone-310: Cortex-M85+Ethos-U55", - "Generic Inference Runner: Ethos-U55", - ], - [ - DeviceInfo("ethos-u65", 256), - "Corstone-310", - "Corstone-310: Cortex-M85+Ethos-U65", - "Generic Inference Runner: Ethos-U65", - ], + _ready_for_uninstall_mock(), + "already_installed", + [call()], ], - ) - def test_artifact_resolver( - device: DeviceInfo, backend: str, expected_system: str, expected_app: str - ) -> None: - """Test artifact resolving based on the provided parameters.""" - generic_runner = get_generic_runner(device, backend) - assert isinstance(generic_runner, GenericInferenceRunnerEthosU) - - assert generic_runner.system_name == expected_system - assert generic_runner.app_name == expected_app - - @staticmethod - def test_artifact_resolver_unsupported_backend() -> None: - """Test that it should be not possible to use unsupported backends.""" - with pytest.raises( - RuntimeError, match="Unsupported device ethos-u65 for backend test_backend" - ): - get_generic_runner(DeviceInfo("ethos-u65", 256), "test_backend") - - @staticmethod - @pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) - def test_inference_should_fail_if_system_not_installed( - backend_runner: MagicMock, test_tflite_model: Path, backend: str - ) -> None: - """Test that inference should fail if system is not installed.""" - backend_runner.is_system_installed.return_value = False - - generic_runner = get_generic_runner(DeviceInfo("ethos-u55", 256), backend) - with pytest.raises( - Exception, - match=r"System Corstone-3[01]0: Cortex-M[58]5\+Ethos-U55 is not installed", - ): - generic_runner.run(ModelInfo(test_tflite_model), []) - - @staticmethod - @pytest.mark.parametrize("backend", ("Corstone-300", "Corstone-310")) - def test_inference_should_fail_is_apps_not_installed( - backend_runner: MagicMock, test_tflite_model: Path, backend: str - ) -> None: - """Test that inference should fail if apps are not installed.""" - backend_runner.is_system_installed.return_value = True - backend_runner.is_application_installed.return_value = False - - generic_runner = get_generic_runner(DeviceInfo("ethos-u55", 256), backend) - with pytest.raises( - Exception, - match="Application Generic Inference Runner: Ethos-U55" - r" for the system Corstone-3[01]0: Cortex-M[58]5\+Ethos-U55 is not " - r"installed", - ): - generic_runner.run(ModelInfo(test_tflite_model), []) - - -@pytest.fixture(name="backend_runner") -def fixture_backend_runner(monkeypatch: pytest.MonkeyPatch) -> MagicMock: - """Mock backend runner.""" - backend_runner_mock = MagicMock(spec=BackendRunner) - monkeypatch.setattr( - "mlia.backend.manager.get_backend_runner", - MagicMock(return_value=backend_runner_mock), - ) - return backend_runner_mock + ], +) +def test_installation_manager_uninstall( + install_mock: MagicMock, + noninteractive: bool, + backend_name: str, + expected_call: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test uninstallation.""" + install_mock.reset_mock() + + manager = get_installation_manager(noninteractive, [install_mock], monkeypatch) + manager.uninstall(backend_name) + + assert install_mock.uninstall.mock_calls == expected_call diff --git a/tests/test_tools_metadata_py_package.py b/tests/test_backend_tosa_checker_install.py index 8b93e33..0393f0b 100644 --- a/tests/test_tools_metadata_py_package.py +++ b/tests/test_backend_tosa_checker_install.py @@ -6,22 +6,10 @@ from unittest.mock import MagicMock import pytest -from mlia.tools.metadata.common import DownloadAndInstall -from mlia.tools.metadata.common import InstallFromPath -from mlia.tools.metadata.py_package import get_pypackage_backend_installations -from mlia.tools.metadata.py_package import get_tosa_backend_installation -from mlia.tools.metadata.py_package import PyPackageBackendInstallation - - -def test_get_pypackage_backends() -> None: - """Test function get_pypackage_backends.""" - backend_installs = get_pypackage_backend_installations() - - assert isinstance(backend_installs, list) - assert len(backend_installs) == 1 - - tosa_installation = backend_installs[0] - assert isinstance(tosa_installation, PyPackageBackendInstallation) +from mlia.backend.install import DownloadAndInstall +from mlia.backend.install import InstallFromPath +from mlia.backend.install import PyPackageBackendInstallation +from mlia.backend.tosa_checker.install import get_tosa_backend_installation def test_get_tosa_backend_installation( @@ -30,7 +18,7 @@ def test_get_tosa_backend_installation( """Test function get_tosa_backend_installation.""" mock_package_manager = MagicMock() monkeypatch.setattr( - "mlia.tools.metadata.py_package.get_package_manager", + "mlia.backend.install.get_package_manager", lambda: mock_package_manager, ) diff --git a/tests/test_backend_vela_compat.py b/tests/test_backend_vela_compat.py new file mode 100644 index 0000000..6f7a41c --- /dev/null +++ b/tests/test_backend_vela_compat.py @@ -0,0 +1,74 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for module vela/compat.""" +from pathlib import Path + +import pytest + +from mlia.backend.vela.compat import generate_supported_operators_report +from mlia.backend.vela.compat import NpuSupported +from mlia.backend.vela.compat import Operator +from mlia.backend.vela.compat import Operators +from mlia.backend.vela.compat import supported_operators +from mlia.devices.ethosu.config import EthosUConfiguration +from mlia.utils.filesystem import working_directory + + +@pytest.mark.parametrize( + "model, expected_ops", + [ + ( + "test_model.tflite", + Operators( + ops=[ + Operator( + name="sequential/conv1/Relu;sequential/conv1/BiasAdd;" + "sequential/conv2/Conv2D;sequential/conv1/Conv2D", + op_type="CONV_2D", + run_on_npu=NpuSupported(supported=True, reasons=[]), + ), + Operator( + name="sequential/conv2/Relu;sequential/conv2/BiasAdd;" + "sequential/conv2/Conv2D", + op_type="CONV_2D", + run_on_npu=NpuSupported(supported=True, reasons=[]), + ), + Operator( + name="sequential/max_pooling2d/MaxPool", + op_type="MAX_POOL_2D", + run_on_npu=NpuSupported(supported=True, reasons=[]), + ), + Operator( + name="sequential/flatten/Reshape", + op_type="RESHAPE", + run_on_npu=NpuSupported(supported=True, reasons=[]), + ), + Operator( + name="Identity", + op_type="FULLY_CONNECTED", + run_on_npu=NpuSupported(supported=True, reasons=[]), + ), + ] + ), + ) + ], +) +def test_operators(test_models_path: Path, model: str, expected_ops: Operators) -> None: + """Test operators function.""" + device = EthosUConfiguration("ethos-u55-256") + + operators = supported_operators(test_models_path / model, device.compiler_options) + for expected, actual in zip(expected_ops.ops, operators.ops): + # do not compare names as they could be different on each model generation + assert expected.op_type == actual.op_type + assert expected.run_on_npu == actual.run_on_npu + + +def test_generate_supported_operators_report(tmp_path: Path) -> None: + """Test generating supported operators report.""" + with working_directory(tmp_path): + generate_supported_operators_report() + + md_file = tmp_path / "SUPPORTED_OPS.md" + assert md_file.is_file() + assert md_file.stat().st_size > 0 diff --git a/tests/test_tools_vela_wrapper.py b/tests/test_backend_vela_compiler.py index 0efcb0f..40268ae 100644 --- a/tests/test_tools_vela_wrapper.py +++ b/tests/test_backend_vela_compiler.py @@ -1,26 +1,16 @@ # SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. # SPDX-License-Identifier: Apache-2.0 -"""Tests for module tools/vela_wrapper.""" +"""Tests for module vela/compiler.""" from pathlib import Path -from unittest.mock import MagicMock -import pytest from ethosu.vela.compiler_driver import TensorAllocator from ethosu.vela.scheduler import OptimizationStrategy +from mlia.backend.vela.compiler import optimize_model +from mlia.backend.vela.compiler import OptimizedModel +from mlia.backend.vela.compiler import VelaCompiler +from mlia.backend.vela.compiler import VelaCompilerOptions from mlia.devices.ethosu.config import EthosUConfiguration -from mlia.tools.vela_wrapper import estimate_performance -from mlia.tools.vela_wrapper import generate_supported_operators_report -from mlia.tools.vela_wrapper import NpuSupported -from mlia.tools.vela_wrapper import Operator -from mlia.tools.vela_wrapper import Operators -from mlia.tools.vela_wrapper import optimize_model -from mlia.tools.vela_wrapper import OptimizedModel -from mlia.tools.vela_wrapper import PerformanceMetrics -from mlia.tools.vela_wrapper import supported_operators -from mlia.tools.vela_wrapper import VelaCompiler -from mlia.tools.vela_wrapper import VelaCompilerOptions -from mlia.utils.filesystem import working_directory def test_default_vela_compiler() -> None: @@ -171,115 +161,3 @@ def test_optimize_model(tmp_path: Path, test_tflite_model: Path) -> None: assert tmp_file.is_file() assert tmp_file.stat().st_size > 0 - - -@pytest.mark.parametrize( - "model, expected_ops", - [ - ( - "test_model.tflite", - Operators( - ops=[ - Operator( - name="sequential/conv1/Relu;sequential/conv1/BiasAdd;" - "sequential/conv2/Conv2D;sequential/conv1/Conv2D", - op_type="CONV_2D", - run_on_npu=NpuSupported(supported=True, reasons=[]), - ), - Operator( - name="sequential/conv2/Relu;sequential/conv2/BiasAdd;" - "sequential/conv2/Conv2D", - op_type="CONV_2D", - run_on_npu=NpuSupported(supported=True, reasons=[]), - ), - Operator( - name="sequential/max_pooling2d/MaxPool", - op_type="MAX_POOL_2D", - run_on_npu=NpuSupported(supported=True, reasons=[]), - ), - Operator( - name="sequential/flatten/Reshape", - op_type="RESHAPE", - run_on_npu=NpuSupported(supported=True, reasons=[]), - ), - Operator( - name="Identity", - op_type="FULLY_CONNECTED", - run_on_npu=NpuSupported(supported=True, reasons=[]), - ), - ] - ), - ) - ], -) -def test_operators(test_models_path: Path, model: str, expected_ops: Operators) -> None: - """Test operators function.""" - device = EthosUConfiguration("ethos-u55-256") - - operators = supported_operators(test_models_path / model, device.compiler_options) - for expected, actual in zip(expected_ops.ops, operators.ops): - # do not compare names as they could be different on each model generation - assert expected.op_type == actual.op_type - assert expected.run_on_npu == actual.run_on_npu - - -def test_estimate_performance(test_tflite_model: Path) -> None: - """Test getting performance estimations.""" - device = EthosUConfiguration("ethos-u55-256") - perf_metrics = estimate_performance(test_tflite_model, device.compiler_options) - - assert isinstance(perf_metrics, PerformanceMetrics) - - -def test_estimate_performance_already_optimized( - tmp_path: Path, test_tflite_model: Path -) -> None: - """Test that performance estimation should fail for already optimized model.""" - device = EthosUConfiguration("ethos-u55-256") - - optimized_model_path = tmp_path / "optimized_model.tflite" - - optimize_model(test_tflite_model, device.compiler_options, optimized_model_path) - - with pytest.raises( - Exception, match="Unable to estimate performance for the given optimized model" - ): - estimate_performance(optimized_model_path, device.compiler_options) - - -def test_generate_supported_operators_report(tmp_path: Path) -> None: - """Test generating supported operators report.""" - with working_directory(tmp_path): - generate_supported_operators_report() - - md_file = tmp_path / "SUPPORTED_OPS.md" - assert md_file.is_file() - assert md_file.stat().st_size > 0 - - -def test_read_invalid_model(test_tflite_invalid_model: Path) -> None: - """Test that reading invalid model should fail with exception.""" - with pytest.raises( - Exception, match=f"Unable to read model {test_tflite_invalid_model}" - ): - device = EthosUConfiguration("ethos-u55-256") - estimate_performance(test_tflite_invalid_model, device.compiler_options) - - -def test_compile_invalid_model( - test_tflite_model: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - """Test that if model could not be compiled then correct exception raised.""" - mock_compiler = MagicMock() - mock_compiler.side_effect = Exception("Bad model!") - - monkeypatch.setattr("mlia.tools.vela_wrapper.compiler_driver", mock_compiler) - - model_path = tmp_path / "optimized_model.tflite" - with pytest.raises( - Exception, match="Model could not be optimized with Vela compiler" - ): - device = EthosUConfiguration("ethos-u55-256") - optimize_model(test_tflite_model, device.compiler_options, model_path) - - assert not model_path.exists() diff --git a/tests/test_backend_vela_performance.py b/tests/test_backend_vela_performance.py new file mode 100644 index 0000000..a1c806c --- /dev/null +++ b/tests/test_backend_vela_performance.py @@ -0,0 +1,64 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-License-Identifier: Apache-2.0 +"""Tests for module vela/performance.""" +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from mlia.backend.vela.compiler import optimize_model +from mlia.backend.vela.performance import estimate_performance +from mlia.backend.vela.performance import PerformanceMetrics +from mlia.devices.ethosu.config import EthosUConfiguration + + +def test_estimate_performance(test_tflite_model: Path) -> None: + """Test getting performance estimations.""" + device = EthosUConfiguration("ethos-u55-256") + perf_metrics = estimate_performance(test_tflite_model, device.compiler_options) + + assert isinstance(perf_metrics, PerformanceMetrics) + + +def test_estimate_performance_already_optimized( + tmp_path: Path, test_tflite_model: Path +) -> None: + """Test that performance estimation should fail for already optimized model.""" + device = EthosUConfiguration("ethos-u55-256") + + optimized_model_path = tmp_path / "optimized_model.tflite" + + optimize_model(test_tflite_model, device.compiler_options, optimized_model_path) + + with pytest.raises( + Exception, match="Unable to estimate performance for the given optimized model" + ): + estimate_performance(optimized_model_path, device.compiler_options) + + +def test_read_invalid_model(test_tflite_invalid_model: Path) -> None: + """Test that reading invalid model should fail with exception.""" + with pytest.raises( + Exception, match=f"Unable to read model {test_tflite_invalid_model}" + ): + device = EthosUConfiguration("ethos-u55-256") + estimate_performance(test_tflite_invalid_model, device.compiler_options) + + +def test_compile_invalid_model( + test_tflite_model: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Test that if model could not be compiled then correct exception raised.""" + mock_compiler = MagicMock() + mock_compiler.side_effect = Exception("Bad model!") + + monkeypatch.setattr("mlia.backend.vela.compiler.compiler_driver", mock_compiler) + + model_path = tmp_path / "optimized_model.tflite" + with pytest.raises( + Exception, match="Model could not be optimized with Vela compiler" + ): + device = EthosUConfiguration("ethos-u55-256") + optimize_model(test_tflite_model, device.compiler_options, model_path) + + assert not model_path.exists() diff --git a/tests/test_cli_commands.py b/tests/test_cli_commands.py index 3a01f78..77e1f88 100644 --- a/tests/test_cli_commands.py +++ b/tests/test_cli_commands.py @@ -10,6 +10,7 @@ from unittest.mock import MagicMock import pytest +from mlia.backend.manager import DefaultInstallationManager from mlia.cli.commands import backend_install from mlia.cli.commands import backend_list from mlia.cli.commands import backend_uninstall @@ -21,7 +22,6 @@ from mlia.devices.ethosu.config import EthosUConfiguration from mlia.devices.ethosu.performance import MemoryUsage from mlia.devices.ethosu.performance import NPUCycles from mlia.devices.ethosu.performance import PerformanceMetrics -from mlia.tools.metadata.common import DefaultInstallationManager def test_operators_expected_parameters(sample_context: ExecutionContext) -> None: diff --git a/tests/test_devices_ethosu_config.py b/tests/test_devices_ethosu_config.py index d4e043f..2fec0d5 100644 --- a/tests/test_devices_ethosu_config.py +++ b/tests/test_devices_ethosu_config.py @@ -9,9 +9,9 @@ from unittest.mock import MagicMock import pytest +from mlia.backend.vela.compiler import VelaCompilerOptions from mlia.devices.ethosu.config import EthosUConfiguration from mlia.devices.ethosu.config import get_target -from mlia.tools.vela_wrapper import VelaCompilerOptions from mlia.utils.filesystem import get_vela_config diff --git a/tests/test_devices_ethosu_data_analysis.py b/tests/test_devices_ethosu_data_analysis.py index 26aae76..8184c70 100644 --- a/tests/test_devices_ethosu_data_analysis.py +++ b/tests/test_devices_ethosu_data_analysis.py @@ -5,6 +5,9 @@ from __future__ import annotations import pytest +from mlia.backend.vela.compat import NpuSupported +from mlia.backend.vela.compat import Operator +from mlia.backend.vela.compat import Operators from mlia.core.common import DataItem from mlia.core.data_analysis import Fact from mlia.devices.ethosu.config import EthosUConfiguration @@ -20,9 +23,6 @@ from mlia.devices.ethosu.performance import NPUCycles from mlia.devices.ethosu.performance import OptimizationPerformanceMetrics from mlia.devices.ethosu.performance import PerformanceMetrics from mlia.nn.tensorflow.optimizations.select import OptimizationSettings -from mlia.tools.vela_wrapper import NpuSupported -from mlia.tools.vela_wrapper import Operator -from mlia.tools.vela_wrapper import Operators def test_perf_metrics_diff() -> None: diff --git a/tests/test_devices_ethosu_data_collection.py b/tests/test_devices_ethosu_data_collection.py index a4f37aa..84b9424 100644 --- a/tests/test_devices_ethosu_data_collection.py +++ b/tests/test_devices_ethosu_data_collection.py @@ -6,6 +6,7 @@ from unittest.mock import MagicMock import pytest +from mlia.backend.vela.compat import Operators from mlia.core.context import Context from mlia.core.data_collection import DataCollector from mlia.core.errors import FunctionalityNotSupportedError @@ -18,7 +19,6 @@ from mlia.devices.ethosu.performance import NPUCycles from mlia.devices.ethosu.performance import OptimizationPerformanceMetrics from mlia.devices.ethosu.performance import PerformanceMetrics from mlia.nn.tensorflow.optimizations.select import OptimizationSettings -from mlia.tools.vela_wrapper import Operators @pytest.mark.parametrize( diff --git a/tests/test_devices_ethosu_performance.py b/tests/test_devices_ethosu_performance.py index b3e5298..3ff73d8 100644 --- a/tests/test_devices_ethosu_performance.py +++ b/tests/test_devices_ethosu_performance.py @@ -23,6 +23,6 @@ def test_memory_usage_conversion() -> None: def mock_performance_estimation(monkeypatch: pytest.MonkeyPatch) -> None: """Mock performance estimation.""" monkeypatch.setattr( - "mlia.backend.manager.estimate_performance", + "mlia.backend.corstone.performance.estimate_performance", MagicMock(return_value=MagicMock()), ) diff --git a/tests/test_devices_ethosu_reporters.py b/tests/test_devices_ethosu_reporters.py index f04270c..926c4c3 100644 --- a/tests/test_devices_ethosu_reporters.py +++ b/tests/test_devices_ethosu_reporters.py @@ -13,6 +13,9 @@ from typing import Literal import pytest +from mlia.backend.vela.compat import NpuSupported +from mlia.backend.vela.compat import Operator +from mlia.backend.vela.compat import Operators from mlia.core.reporting import get_reporter from mlia.core.reporting import produce_report from mlia.core.reporting import Report @@ -26,9 +29,6 @@ from mlia.devices.ethosu.reporters import ethos_u_formatters from mlia.devices.ethosu.reporters import report_device_details from mlia.devices.ethosu.reporters import report_operators from mlia.devices.ethosu.reporters import report_perf_metrics -from mlia.tools.vela_wrapper import NpuSupported -from mlia.tools.vela_wrapper import Operator -from mlia.tools.vela_wrapper import Operators from mlia.utils.console import remove_ascii_codes diff --git a/tests/test_tools_metadata_common.py b/tests/test_tools_metadata_common.py deleted file mode 100644 index 9811852..0000000 --- a/tests/test_tools_metadata_common.py +++ /dev/null @@ -1,282 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Tests for commmon installation related functions.""" -from __future__ import annotations - -from pathlib import Path -from typing import Any -from unittest.mock import call -from unittest.mock import MagicMock -from unittest.mock import PropertyMock - -import pytest - -from mlia.tools.metadata.common import DefaultInstallationManager -from mlia.tools.metadata.common import DownloadAndInstall -from mlia.tools.metadata.common import Installation -from mlia.tools.metadata.common import InstallationType -from mlia.tools.metadata.common import InstallFromPath - - -def get_default_installation_manager_mock( - name: str, - already_installed: bool = False, -) -> MagicMock: - """Get mock instance for DefaultInstallationManager.""" - mock = MagicMock(spec=DefaultInstallationManager) - - props = { - "name": name, - "already_installed": already_installed, - } - for prop, value in props.items(): - setattr(type(mock), prop, PropertyMock(return_value=value)) - - return mock - - -def _ready_for_uninstall_mock() -> MagicMock: - return get_default_installation_manager_mock( - name="already_installed", - already_installed=True, - ) - - -def get_installation_mock( - name: str, - already_installed: bool = False, - could_be_installed: bool = False, - supported_install_type: type | tuple | None = None, -) -> MagicMock: - """Get mock instance for the installation.""" - mock = MagicMock(spec=Installation) - - def supports(install_type: InstallationType) -> bool: - if supported_install_type is None: - return False - - return isinstance(install_type, supported_install_type) - - mock.supports.side_effect = supports - - props = { - "name": name, - "already_installed": already_installed, - "could_be_installed": could_be_installed, - } - for prop, value in props.items(): - setattr(type(mock), prop, PropertyMock(return_value=value)) - - return mock - - -def _already_installed_mock() -> MagicMock: - return get_installation_mock( - name="already_installed", - already_installed=True, - supported_install_type=(DownloadAndInstall, InstallFromPath), - ) - - -def _ready_for_installation_mock() -> MagicMock: - return get_installation_mock( - name="ready_for_installation", - already_installed=False, - could_be_installed=True, - ) - - -def _could_be_downloaded_and_installed_mock() -> MagicMock: - return get_installation_mock( - name="could_be_downloaded_and_installed", - already_installed=False, - could_be_installed=True, - supported_install_type=DownloadAndInstall, - ) - - -def _could_be_installed_from_mock() -> MagicMock: - return get_installation_mock( - name="could_be_installed_from", - already_installed=False, - could_be_installed=True, - supported_install_type=InstallFromPath, - ) - - -def get_installation_manager( - noninteractive: bool, - installations: list[Any], - monkeypatch: pytest.MonkeyPatch, - yes_response: bool = True, -) -> DefaultInstallationManager: - """Get installation manager instance.""" - if not noninteractive: - monkeypatch.setattr( - "mlia.tools.metadata.common.yes", MagicMock(return_value=yes_response) - ) - - return DefaultInstallationManager(installations, noninteractive=noninteractive) - - -def test_installation_manager_filtering() -> None: - """Test default installation manager.""" - already_installed = _already_installed_mock() - ready_for_installation = _ready_for_installation_mock() - could_be_downloaded_and_installed = _could_be_downloaded_and_installed_mock() - - manager = DefaultInstallationManager( - [ - already_installed, - ready_for_installation, - could_be_downloaded_and_installed, - ] - ) - assert manager.already_installed("already_installed") == [already_installed] - assert manager.ready_for_installation() == [ - ready_for_installation, - could_be_downloaded_and_installed, - ] - - -@pytest.mark.parametrize("noninteractive", [True, False]) -@pytest.mark.parametrize( - "install_mock, eula_agreement, backend_name, force, expected_call", - [ - [ - _could_be_downloaded_and_installed_mock(), - True, - "could_be_downloaded_and_installed", - False, - [call(DownloadAndInstall(eula_agreement=True))], - ], - [ - _could_be_downloaded_and_installed_mock(), - False, - "could_be_downloaded_and_installed", - True, - [call(DownloadAndInstall(eula_agreement=False))], - ], - [ - _already_installed_mock(), - False, - "already_installed", - True, - [call(DownloadAndInstall(eula_agreement=False))], - ], - [ - _could_be_downloaded_and_installed_mock(), - False, - "unknown", - True, - [], - ], - ], -) -def test_installation_manager_download_and_install( - install_mock: MagicMock, - noninteractive: bool, - eula_agreement: bool, - backend_name: str, - force: bool, - expected_call: Any, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Test installation process.""" - install_mock.reset_mock() - - manager = get_installation_manager(noninteractive, [install_mock], monkeypatch) - - manager.download_and_install( - backend_name, eula_agreement=eula_agreement, force=force - ) - - assert install_mock.install.mock_calls == expected_call - if force and install_mock.already_installed: - install_mock.uninstall.assert_called_once() - else: - install_mock.uninstall.assert_not_called() - - -@pytest.mark.parametrize("noninteractive", [True, False]) -@pytest.mark.parametrize( - "install_mock, backend_name, force, expected_call", - [ - [ - _could_be_installed_from_mock(), - "could_be_installed_from", - False, - [call(InstallFromPath(Path("some_path")))], - ], - [ - _could_be_installed_from_mock(), - "unknown", - False, - [], - ], - [ - _could_be_installed_from_mock(), - "unknown", - True, - [], - ], - [ - _already_installed_mock(), - "already_installed", - False, - [], - ], - [ - _already_installed_mock(), - "already_installed", - True, - [call(InstallFromPath(Path("some_path")))], - ], - ], -) -def test_installation_manager_install_from( - install_mock: MagicMock, - noninteractive: bool, - backend_name: str, - force: bool, - expected_call: Any, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Test installation process.""" - install_mock.reset_mock() - - manager = get_installation_manager(noninteractive, [install_mock], monkeypatch) - manager.install_from(Path("some_path"), backend_name, force=force) - - assert install_mock.install.mock_calls == expected_call - if force and install_mock.already_installed: - install_mock.uninstall.assert_called_once() - else: - install_mock.uninstall.assert_not_called() - - -@pytest.mark.parametrize("noninteractive", [True, False]) -@pytest.mark.parametrize( - "install_mock, backend_name, expected_call", - [ - [ - _ready_for_uninstall_mock(), - "already_installed", - [call()], - ], - ], -) -def test_installation_manager_uninstall( - install_mock: MagicMock, - noninteractive: bool, - backend_name: str, - expected_call: Any, - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Test uninstallation.""" - install_mock.reset_mock() - - manager = get_installation_manager(noninteractive, [install_mock], monkeypatch) - manager.uninstall(backend_name) - - assert install_mock.uninstall.mock_calls == expected_call |