fix: resolve ty type diagnostics and add CI quality gates

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)
+

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

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

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"):

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:

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):

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)

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,15 +514,15 @@ 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()
 
     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()