diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/common_test.py b/python/packages/jumpstarter-cli/jumpstarter_cli/common_test.py index 0e94b5d55..1b85fc177 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/common_test.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/common_test.py @@ -88,3 +88,12 @@ def test_invalid_duration_raises_click_exception(self): with pytest.raises(click.BadParameter, match="is not a valid duration"): param_type.convert("not-a-duration", None, None) + def test_integer_value_as_seconds(self): + td = DURATION.convert(42, None, None) + assert td == timedelta(seconds=42) + + def test_unsupported_type_raises_click_exception(self): + param_type = DurationParamType() + with pytest.raises(click.BadParameter, match="is not a valid duration"): + param_type.convert(object(), None, None) + diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/completion_test.py b/python/packages/jumpstarter-cli/jumpstarter_cli/completion_test.py index 100517be4..f8d74ae05 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/completion_test.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/completion_test.py @@ -1,3 +1,5 @@ +from unittest.mock import patch + from click.testing import CliRunner from .jmp import jmp @@ -41,3 +43,11 @@ def test_completion_unsupported_shell(): result = runner.invoke(jmp, ["completion", "powershell"]) assert result.exit_code == 2 assert "Invalid value" in result.output or "powershell" in result.output + + +def test_completion_raises_when_get_completion_class_returns_none(): + with patch("jumpstarter_cli.completion.get_completion_class", return_value=None): + runner = CliRunner() + result = runner.invoke(jmp, ["completion", "bash"]) + assert result.exit_code == 1 + assert "Unsupported shell" in result.output diff --git a/python/packages/jumpstarter-cli/jumpstarter_cli/login.py b/python/packages/jumpstarter-cli/jumpstarter_cli/login.py index 960a8296d..9c0f0da75 100644 --- a/python/packages/jumpstarter-cli/jumpstarter_cli/login.py +++ b/python/packages/jumpstarter-cli/jumpstarter_cli/login.py @@ -228,11 +228,13 @@ async def login( # noqa: C901 match config: # we are updating an existing config case ClientConfigV1Alpha1(): - assert config.token is not None + if config.token is None: + raise click.ClickException("No token set in client config. Please login again.") issuer = decode_jwt_issuer(config.token) config_kind = "client" case ExporterConfigV1Alpha1(): - assert config.token is not None + if config.token is None: + raise click.ClickException("No token set in exporter config. Please login again.") issuer = decode_jwt_issuer(config.token) config_kind = "exporter" # we are creating a new config diff --git a/python/packages/jumpstarter-driver-gpiod/jumpstarter_driver_gpiod/driver_test.py b/python/packages/jumpstarter-driver-gpiod/jumpstarter_driver_gpiod/driver_test.py index d5cf5eea0..91555b390 100644 --- a/python/packages/jumpstarter-driver-gpiod/jumpstarter_driver_gpiod/driver_test.py +++ b/python/packages/jumpstarter-driver-gpiod/jumpstarter_driver_gpiod/driver_test.py @@ -196,11 +196,11 @@ def test_digital_output_methods(self, mock_gpiod): # Test on() method driver.on() - driver._line.set_value.assert_called_with(18, mock_gpiod.line.Value.ACTIVE) + driver._line.set_value.assert_called_with(18, mock_gpiod.line.Value.ACTIVE) # ty: ignore[possibly-unbound-attribute] # Test off() method driver.off() - driver._line.set_value.assert_called_with(18, mock_gpiod.line.Value.INACTIVE) + driver._line.set_value.assert_called_with(18, mock_gpiod.line.Value.INACTIVE) # ty: ignore[possibly-unbound-attribute] # Test read_pin() method driver._line.get_value.return_value = mock_gpiod.line.Value.ACTIVE # ty: ignore[invalid-assignment] @@ -238,7 +238,7 @@ def test_digital_input_methods(self, mock_gpiod): # Test wait_for_active() when already active driver._line.get_value.return_value = mock_gpiod.line.Value.ACTIVE # ty: ignore[invalid-assignment] driver.wait_for_active() - driver._line.wait_edge_events.assert_not_called() + driver._line.wait_edge_events.assert_not_called() # ty: ignore[possibly-unbound-attribute] # Test wait_for_active() with timeout driver._line.get_value.return_value = mock_gpiod.line.Value.INACTIVE # ty: ignore[invalid-assignment] @@ -256,8 +256,8 @@ def test_digital_input_methods(self, mock_gpiod): driver._line.read_edge_events.return_value = [mock_event] # ty: ignore[invalid-assignment] driver.wait_for_edge("rising") - driver._line.wait_edge_events.assert_called() - driver._line.read_edge_events.assert_called() + driver._line.wait_edge_events.assert_called() # ty: ignore[possibly-unbound-attribute] + driver._line.read_edge_events.assert_called() # ty: ignore[possibly-unbound-attribute] # Test wait_for_edge() with invalid edge type with pytest.raises(ValueError, match="Invalid edge type: invalid"): diff --git a/python/packages/jumpstarter-driver-mitmproxy/jumpstarter_driver_mitmproxy/driver_test.py b/python/packages/jumpstarter-driver-mitmproxy/jumpstarter_driver_mitmproxy/driver_test.py index ca7553529..532b6ed1a 100644 --- a/python/packages/jumpstarter-driver-mitmproxy/jumpstarter_driver_mitmproxy/driver_test.py +++ b/python/packages/jumpstarter-driver-mitmproxy/jumpstarter_driver_mitmproxy/driver_test.py @@ -15,7 +15,12 @@ import pytest -from jumpstarter_driver_mitmproxy.driver import MitmproxyDriver +from jumpstarter_driver_mitmproxy.driver import ( + DirectoriesConfig, + ListenConfig, + MitmproxyDriver, + WebConfig, +) @pytest.fixture @@ -598,6 +603,9 @@ def test_defaults_from_data_dir(self): directories={"data": "/tmp/myproxy"}, ) try: + assert isinstance(d.directories, DirectoriesConfig) + assert isinstance(d.listen, ListenConfig) + assert isinstance(d.web, WebConfig) assert d.directories.data == "/tmp/myproxy" assert d.directories.conf == "/tmp/myproxy/conf" assert d.directories.flows == "/tmp/myproxy/flows" @@ -619,6 +627,7 @@ def test_partial_directory_override(self): }, ) try: + assert isinstance(d.directories, DirectoriesConfig) assert d.directories.conf == "/etc/mitmproxy" assert d.directories.flows == "/tmp/myproxy/flows" finally: diff --git a/python/packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/manager.py b/python/packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/manager.py index 7f2ce1570..a61de951b 100644 --- a/python/packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/manager.py +++ b/python/packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/manager.py @@ -15,6 +15,7 @@ import sys import threading from dataclasses import dataclass +from types import FrameType from typing import Callable, Optional logger = logging.getLogger(__name__) @@ -155,7 +156,7 @@ def _install_signal_handlers(self): if cls._signal_handlers_installed: return - def make_handler(sig: signal.Signals) -> Callable[[int, any], None]: # ty: ignore[invalid-type-form] + def make_handler(sig: signal.Signals) -> Callable[[int, FrameType | None], None]: """Create a signal handler that cleans up and re-raises the signal.""" def handler(signum: int, frame): diff --git a/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver.py b/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver.py index 2ce08813b..afcd4c0d6 100644 --- a/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver.py +++ b/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver.py @@ -86,7 +86,7 @@ async def flash(self, source, load_command: str | None = None): cmd = _detect_load_command(firmware_path) self.parent.set_firmware(firmware_path, cmd) - power: RenodePower = self.parent.children["power"] + power: RenodePower = self.parent.children["power"] # ty: ignore[invalid-assignment] if power.is_running: await power.send_monitor_command(f'{cmd} @"{firmware_path}"') await power.send_monitor_command("machine Reset") @@ -295,5 +295,5 @@ async def monitor_cmd(self, command: str) -> str: raise RuntimeError( "raw monitor access is disabled; set allow_raw_monitor: true in exporter config to enable" ) - power: RenodePower = self.children["power"] + power: RenodePower = self.children["power"] # ty: ignore[invalid-assignment] return await power.send_monitor_command(command) diff --git a/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver_test.py b/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver_test.py index c9f18ab4e..4c6bd5290 100644 --- a/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver_test.py +++ b/python/packages/jumpstarter-driver-renode/jumpstarter_driver_renode/driver_test.py @@ -33,8 +33,8 @@ async def test_monitor_connect_retry(self): async def mock_connect_tcp(host, port): nonlocal call_count - call_count += 1 - if call_count < 3: + call_count += 1 # ty: ignore[unresolved-reference] + if call_count < 3: # ty: ignore[unresolved-reference] raise OSError("Connection refused") stream = AsyncMock() stream.receive = AsyncMock(return_value=b"Renode v1.15\n(monitor) \n") @@ -128,7 +128,7 @@ async def test_monitor_connect_closes_stream_on_retry(self): async def mock_connect_tcp(host, port): nonlocal call_count - call_count += 1 + call_count += 1 # ty: ignore[unresolved-reference] stream = AsyncMock() streams.append(stream) if call_count < 2: @@ -254,7 +254,7 @@ def test_close_sync_no_stream(self): def _make_driver(**kwargs) -> Renode: defaults = {"platform": "platforms/boards/stm32f4_discovery-kit.repl"} defaults.update(kwargs) - return Renode(**defaults) + return Renode(**defaults) # ty: ignore[missing-argument] class TestRenodePower: @@ -264,7 +264,7 @@ async def test_power_on_command_sequence(self): driver = _make_driver(uart="sysbus.usart2") driver._firmware_path = "/tmp/test.elf" driver._load_command = "sysbus LoadELF" - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_monitor = AsyncMock(spec=RenodeMonitor) @@ -298,7 +298,7 @@ async def test_power_on_with_extra_commands(self): """Extra commands are sent between connector Connect and LoadELF.""" driver = _make_driver(extra_commands=["sysbus WriteDoubleWord 0x40090030 0x0301"]) driver._firmware_path = "/tmp/test.elf" - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_monitor = AsyncMock(spec=RenodeMonitor) with patch( @@ -327,7 +327,7 @@ async def test_power_on_with_extra_commands(self): async def test_power_on_without_firmware(self): """When no firmware is set, LoadELF is skipped but start is sent.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_monitor = AsyncMock(spec=RenodeMonitor) with patch( @@ -354,7 +354,7 @@ async def test_power_on_without_firmware(self): async def test_power_on_idempotent(self): """Second on() call logs warning and does nothing.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] power._process = MagicMock() with patch("jumpstarter_driver_renode.driver.Popen") as mock_popen: @@ -368,7 +368,7 @@ async def test_power_on_idempotent(self): async def test_power_off_terminates_process(self): """off() terminates the process, waits, then kills on timeout.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_process = MagicMock() mock_process.terminate = MagicMock() @@ -391,7 +391,7 @@ async def test_power_off_terminates_process(self): async def test_power_off_clean_shutdown(self): """off() with clean process exit does not call kill().""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_process = MagicMock() mock_process.wait = MagicMock() @@ -408,7 +408,7 @@ async def test_power_off_clean_shutdown(self): async def test_power_off_idempotent(self): """Second off() call logs warning and does nothing.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] power._process = None await power.off() @@ -420,7 +420,7 @@ async def test_power_off_idempotent(self): async def test_power_close_calls_off(self): """close() terminates the process.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_process = MagicMock() mock_process.wait = MagicMock() power._process = mock_process @@ -433,7 +433,7 @@ async def test_power_close_calls_off(self): def test_close_kills_on_timeout(self): """close() kills process when wait() times out.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_process = MagicMock() mock_process.wait = MagicMock(side_effect=TimeoutExpired("renode", 5)) power._process = mock_process @@ -447,7 +447,7 @@ def test_close_kills_on_timeout(self): def test_close_cleans_up_monitor_socket(self): """close() calls close_sync() on the monitor before terminating.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_process = MagicMock() mock_process.wait = MagicMock() power._process = mock_process @@ -464,7 +464,7 @@ def test_close_cleans_up_monitor_socket(self): async def test_power_on_cleanup_on_failure(self): """on() cleans up process when monitor setup fails.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_monitor = AsyncMock(spec=RenodeMonitor) mock_monitor.execute.side_effect = RenodeMonitorError("setup failed") @@ -498,7 +498,7 @@ async def test_power_on_cleanup_on_failure(self): async def test_off_cleans_up_on_terminate_failure(self): """off() resets _process to None even if terminate() raises.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_process = MagicMock() mock_process.terminate = MagicMock(side_effect=ProcessLookupError) @@ -514,7 +514,7 @@ async def test_off_cleans_up_on_terminate_failure(self): async def test_power_read_not_implemented(self): """read() raises NotImplementedError.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] with pytest.raises(NotImplementedError): await power.read() @@ -522,7 +522,7 @@ async def test_power_read_not_implemented(self): def test_is_running_property(self): """is_running reflects process and monitor state.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] assert power.is_running is False @@ -546,7 +546,7 @@ async def test_flash_stores_firmware_path(self, tmp_path): firmware_file = tmp_path / "test.elf" firmware_file.write_bytes(firmware_data) - flasher: RenodeFlasher = driver.children["flasher"] + flasher: RenodeFlasher = driver.children["flasher"] # ty: ignore[invalid-assignment] with patch.object(flasher, "resource") as mock_resource: mock_res = AsyncMock() @@ -564,12 +564,12 @@ async def test_flash_stores_firmware_path(self, tmp_path): async def test_flash_while_running_sends_load_and_reset(self): """When simulation is running, flash() sends load + Reset.""" driver = _make_driver() - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] power._process = MagicMock() mock_monitor = AsyncMock(spec=RenodeMonitor) power._monitor = mock_monitor - flasher: RenodeFlasher = driver.children["flasher"] + flasher: RenodeFlasher = driver.children["flasher"] # ty: ignore[invalid-assignment] elf_data = b"\x7fELF" + b"\x00" * 60 with patch.object(flasher, "resource") as mock_resource: @@ -589,7 +589,7 @@ async def test_flash_while_running_sends_load_and_reset(self): async def test_flash_custom_load_command(self): """flash() uses custom load_command when provided.""" driver = _make_driver() - flasher: RenodeFlasher = driver.children["flasher"] + flasher: RenodeFlasher = driver.children["flasher"] # ty: ignore[invalid-assignment] with patch.object(flasher, "resource") as mock_resource: mock_res = AsyncMock() @@ -609,7 +609,7 @@ async def test_flash_custom_load_command(self): async def test_flash_rejects_invalid_load_command(self): """flash() rejects load_command values not in the allowlist.""" driver = _make_driver() - flasher: RenodeFlasher = driver.children["flasher"] + flasher: RenodeFlasher = driver.children["flasher"] # ty: ignore[invalid-assignment] with pytest.raises(ValueError, match="unsupported load_command"): await flasher.flash("/some/fw.elf", load_command="logFile @/tmp/evil") @@ -618,7 +618,7 @@ async def test_flash_rejects_invalid_load_command(self): async def test_dump_not_implemented(self): """dump() raises NotImplementedError.""" driver = _make_driver() - flasher: RenodeFlasher = driver.children["flasher"] + flasher: RenodeFlasher = driver.children["flasher"] # ty: ignore[invalid-assignment] with pytest.raises(NotImplementedError, match="not supported"): await flasher.dump("/dev/null") @@ -718,7 +718,7 @@ def test_set_firmware(self): async def test_monitor_cmd_success(self): """monitor_cmd succeeds when allow_raw_monitor is True and running.""" driver = _make_driver(allow_raw_monitor=True) - power: RenodePower = driver.children["power"] + power: RenodePower = driver.children["power"] # ty: ignore[invalid-assignment] mock_monitor = AsyncMock(spec=RenodeMonitor) mock_monitor.execute = AsyncMock(return_value="OK\n") power._process = MagicMock()