ci: add quality gates for coverage and type checking

.github/workflows/lint.yaml

@@ -100,6 +100,23 @@ jobs:
           src: './python'
           version-file: python/pyproject.toml
 
+  type-check-python:
+    needs: changes
+    if: needs.changes.outputs.python == 'true'
+    runs-on: ubuntu-latest
+    continue-on-error: true
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@803947b9bd8e9f986429fa0c5a41c367cd732b41 # v7.2.1
+        with:
+          version: latest
+
+      - name: Run ty
+        working-directory: python
+        run: make ty
+
   typos:
     runs-on: ubuntu-latest
     steps:

.github/workflows/python-tests.yaml

@@ -42,6 +42,8 @@ jobs:
         python-version: ["3.11", "3.12", "3.13"]
     steps:
       - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
 
       - name: Install the latest version of uv
         uses: astral-sh/setup-uv@803947b9bd8e9f986429fa0c5a41c367cd732b41 # v7.2.1
@@ -95,9 +97,21 @@ jobs:
 
       - name: Run pytest
         working-directory: python
+        env:
+          PYTEST_ADDOPTS: "--cov --cov-report=xml"
         run: |
             make test
 
+      - name: Check coverage on changed lines
+        if: github.event_name == 'pull_request'
+        working-directory: python
+        run: |
+            coverage_files=$(find packages -name coverage.xml 2>/dev/null | sort)
+            if [ -z "$coverage_files" ]; then
+              echo "::error::No coverage.xml files found"
+              exit 1
+            fi
+            uv run diff-cover $coverage_files --compare-branch=origin/${{ github.base_ref }} --fail-under=80
 
   # https://github.com/orgs/community/discussions/26822
   pytest:

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/exceptions.py

@@ -75,15 +75,15 @@ def _extract_grpc_code_and_details(exc: BaseException) -> tuple[str | None, str]
     code = None
     details = ""
     try:
-        code_member = exc.code
+        code_member = exc.code  # ty: ignore[unresolved-attribute]
         if callable(code_member):
             grpc_code = code_member()
             code = grpc_code.name if hasattr(grpc_code, "name") else str(grpc_code)
     except Exception:
         code = None
 
     try:
-        details_member = exc.details
+        details_member = exc.details  # ty: ignore[unresolved-attribute]
         if callable(details_member):
             details = str(details_member() or "")
     except Exception:

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/opt.py

@@ -188,8 +188,8 @@ def _validate_tokens(tokens: list[str], allowed_values: set[str], ctx, param) ->
 
 
 def parse_comma_separated(
-    ctx: click.Context,
-    param: click.Parameter,
+    ctx: click.Context | None,
+    param: click.Parameter | None,
     value: str | tuple[str, ...] | None,
     allowed_values: set[str] | None = None,
     normalize_case: bool = True

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/print.py

@@ -37,7 +37,7 @@ def model_print(  # noqa: C901
             names = []
 
             try:
-                model.rich_add_names(names)
+                model.rich_add_names(names)  # ty: ignore[unresolved-attribute]
             except AttributeError as err:
                 raise NotImplementedError from err
 
@@ -47,7 +47,7 @@ def model_print(  # noqa: C901
             paths = []
 
             try:
-                model.rich_add_paths(paths)
+                model.rich_add_paths(paths)  # ty: ignore[unresolved-attribute]
             except AttributeError as err:
                 raise NotImplementedError from err
 
@@ -57,12 +57,12 @@ def model_print(  # noqa: C901
             table = Table(
                 box=None,
                 header_style=None,
-                pad_edge=None,
+                pad_edge=False,
             )
 
             try:
-                model.rich_add_columns(table, **kwargs)
-                model.rich_add_rows(table, **kwargs)
+                model.rich_add_columns(table, **kwargs)  # ty: ignore[unresolved-attribute]
+                model.rich_add_rows(table, **kwargs)  # ty: ignore[unresolved-attribute]
             except AttributeError as err:
                 raise NotImplementedError from err
 

python/packages/jumpstarter-cli-driver/jumpstarter_cli_driver/driver.py

@@ -14,7 +14,7 @@ def list_drivers():
         table = Table(
             box=None,
             header_style=None,
-            pad_edge=None,
+            pad_edge=False,
         )
 
         table.add_column("NAME", no_wrap=True)

python/packages/jumpstarter-cli/jumpstarter_cli/common.py

@@ -52,7 +52,7 @@ def convert(self, value, param, ctx):
                 td = None
                 try:
                     seconds = parse_duration(value)
-                    if seconds is not None:
+                    if seconds is not None and isinstance(seconds, (int, float)):
                         td = timedelta(seconds=seconds)
                 except (ValueError, TypeError):
                     pass

python/packages/jumpstarter-cli/jumpstarter_cli/completion.py

@@ -9,5 +9,7 @@ def completion(shell: str):
     from jumpstarter_cli.jmp import jmp
 
     comp_cls = get_completion_class(shell)
+    if comp_cls is None:
+        raise click.ClickException(f"Unsupported shell: {shell}")
     comp = comp_cls(jmp, {}, "jmp", "_JMP_COMPLETE")
     click.echo(comp.source())

python/packages/jumpstarter-cli/jumpstarter_cli/j.py

@@ -6,6 +6,7 @@
 import click
 from anyio import create_task_group, get_cancelled_exc_class, run, to_thread
 from anyio.from_thread import BlockingPortal
+from click.exceptions import Exit as ClickExit
 from jumpstarter_cli_common.exceptions import (
     ClickExceptionRed,
     async_handle_exceptions,
@@ -28,7 +29,7 @@ async def cli():
                     async with env_async(portal, stack) as client:
                         result = await to_thread.run_sync(lambda: client.cli()(standalone_mode=False))
                         if isinstance(result, int) and result != 0:
-                            raise BaseExceptionGroup("CLI exit", [click.exceptions.Exit(result)])
+                            raise BaseExceptionGroup("CLI exit", [ClickExit(result)])
         except BaseExceptionGroup as eg:
             # Handle exceptions wrapped in ExceptionGroup (e.g., from task groups)
             if exc := find_exception_in_group(eg, EnvironmentVariableNotSetError):
@@ -42,9 +43,9 @@ async def cli():
                 await cli()
             finally:
                 tg.cancel_scope.cancel()
-    except* click.exceptions.Exit as excgroup:
+    except* ClickExit as excgroup:
         for exc in leaf_exceptions(excgroup):
-            sys.exit(exc.exit_code)
+            sys.exit(cast(ClickExit, exc).exit_code)
     except* click.ClickException as excgroup:
         for exc in leaf_exceptions(excgroup):
             cast(click.ClickException, exc).show()

python/packages/jumpstarter-cli/jumpstarter_cli/login.py

@@ -228,9 +228,11 @@ async def login(  # noqa: C901
     match config:
         # we are updating an existing config
         case ClientConfigV1Alpha1():
+            assert config.token is not None
             issuer = decode_jwt_issuer(config.token)
             config_kind = "client"
         case ExporterConfigV1Alpha1():
+            assert config.token is not None
             issuer = decode_jwt_issuer(config.token)
             config_kind = "exporter"
         # we are creating a new config
@@ -313,7 +315,7 @@ def save_config() -> None:
     if stored_refresh_token and token is None and username is None and password is None:
         try:
             tokens = await oidc.refresh_token_grant(stored_refresh_token)
-            config.token = tokens["access_token"]
+            config.token = tokens["access_token"]  # ty: ignore[invalid-assignment]
             refresh_token = tokens.get("refresh_token")
             if refresh_token is not None and isinstance(config, ClientConfigV1Alpha1):
                 config.refresh_token = refresh_token
@@ -333,7 +335,7 @@ def save_config() -> None:
     else:
         tokens = await oidc.authorization_code_grant(callback_port=callback_port)
 
-    config.token = tokens["access_token"]
+    config.token = tokens["access_token"]  # ty: ignore[invalid-assignment]
     refresh_token = tokens.get("refresh_token")
 
     # only client configs support refresh_token
@@ -352,6 +354,8 @@ def save_config() -> None:
 async def relogin_client(config: ClientConfigV1Alpha1):
     """Relogin into a jumpstarter instance"""
     client_id = "jumpstarter-cli"  # TODO: store this metadata in the config
+    if config.token is None:
+        raise ReauthenticationFailed("No token set in client config")
     try:
         issuer = decode_jwt_issuer(config.token)
     except Exception as e:

python/packages/jumpstarter-cli/jumpstarter_cli/run.py

@@ -81,7 +81,7 @@ async def signal_handler():
 
             with open_signal_receiver(signal.SIGINT, signal.SIGTERM, signal.SIGHUP, signal.SIGQUIT) as signals:
                 async for sig in signals:
-                    if signal_handled:
+                    if signal_handled:  # ty: ignore[unresolved-reference]
                         continue  # Ignore duplicate signals
                     received_signal = sig
                     logger.info("CHILD: Received %d (%s)", received_signal, signal.Signals(received_signal).name)

python/packages/jumpstarter-cli/jumpstarter_cli/shell.py

@@ -6,8 +6,11 @@
 from types import SimpleNamespace
 
 import anyio
+import anyio.from_thread
+import anyio.to_thread
 import click
 import grpc
+import grpc.aio
 from anyio import create_task_group, get_cancelled_exc_class
 from jumpstarter_cli_common.config import opt_config
 from jumpstarter_cli_common.exceptions import find_exception_in_group, handle_exceptions_with_reauthentication

python/packages/jumpstarter-cli/jumpstarter_cli/shell_test.py

@@ -703,7 +703,7 @@ async def test_sleeps_30s_when_above_threshold(self, _mock_remaining, mock_sleep
 
         def check_cancelled():
             nonlocal call_count
-            call_count += 1
+            call_count += 1  # ty: ignore[unresolved-reference]
             return call_count > 1
 
         config = _make_config()

python/packages/jumpstarter-driver-androidemulator/jumpstarter_driver_androidemulator/client.py

@@ -28,7 +28,7 @@ def adb_device(self, timeout: int = 180):
         Yields:
             An ``adbutils.AdbDevice`` connected through the tunnel.
         """
-        import adbutils
+        import adbutils  # ty: ignore[unresolved-import]
 
         with self.adb.forward_adb(port=0) as (host, port):
             adb = adbutils.AdbClient(host=host, port=port)

python/packages/jumpstarter-driver-androidemulator/jumpstarter_driver_androidemulator/driver_test.py

@@ -1,8 +1,9 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from jumpstarter_driver_adb.driver import AdbServer
 
-from .driver import AndroidEmulator
+from .driver import AndroidEmulator, AndroidEmulatorPower
 from jumpstarter.common.exceptions import ConfigurationError
 
 
@@ -42,7 +43,7 @@ def test_init_invalid_port(mock_run, mock_adb_which, mock_emu_which):
 @patch("subprocess.run", return_value=_mock_adb_ok())
 def test_power_on_builds_cmdline(mock_run, mock_adb_which, mock_emu_which):
     emu = AndroidEmulator(avd_name="test_avd", console_port=5556)
-    power = emu.children["power"]
+    power: AndroidEmulatorPower = emu.children["power"]  # ty: ignore[invalid-assignment]
 
     with patch("subprocess.Popen") as mock_popen:
         mock_proc = MagicMock()
@@ -73,7 +74,7 @@ def test_power_on_builds_cmdline(mock_run, mock_adb_which, mock_emu_which):
 @patch("subprocess.run", return_value=_mock_adb_ok())
 def test_power_on_not_headless(mock_run, mock_adb_which, mock_emu_which):
     emu = AndroidEmulator(avd_name="test_avd", headless=False)
-    power = emu.children["power"]
+    power: AndroidEmulatorPower = emu.children["power"]  # ty: ignore[invalid-assignment]
 
     with patch("subprocess.Popen") as mock_popen:
         mock_proc = MagicMock()
@@ -93,7 +94,7 @@ def test_power_on_not_headless(mock_run, mock_adb_which, mock_emu_which):
 @patch("subprocess.run", return_value=_mock_adb_ok())
 def test_power_off_graceful(mock_run, mock_adb_which, mock_emu_which):
     emu = AndroidEmulator(avd_name="test_avd")
-    power = emu.children["power"]
+    power: AndroidEmulatorPower = emu.children["power"]  # ty: ignore[invalid-assignment]
 
     mock_proc = MagicMock()
     mock_proc.returncode = None
@@ -115,7 +116,7 @@ def test_power_off_force_kill(mock_run, mock_adb_which, mock_emu_which):
     from subprocess import TimeoutExpired
 
     emu = AndroidEmulator(avd_name="test_avd")
-    power = emu.children["power"]
+    power: AndroidEmulatorPower = emu.children["power"]  # ty: ignore[invalid-assignment]
 
     mock_proc = MagicMock()
     mock_proc.returncode = None
@@ -137,5 +138,5 @@ def test_custom_ports(mock_run, mock_adb_which, mock_emu_which):
     emu = AndroidEmulator(avd_name="test_avd", console_port=5556, adb_server_port=15038)
     assert emu.console_port == 5556
     assert emu.adb_server_port == 15038
-    adb_child = emu.children["adb"]
+    adb_child: AdbServer = emu.children["adb"]  # ty: ignore[invalid-assignment]
     assert adb_child.port == 15038

python/packages/jumpstarter-driver-ble/jumpstarter_driver_ble/driver.py

@@ -3,8 +3,9 @@
 from functools import partial
 
 import anyio
+import anyio.streams.memory
 from anyio.abc import ObjectStream
-from anyio.streams.memory import MemoryObjectSendStream
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from bleak import BleakClient, BleakGATTCharacteristic
 from bleak.exc import BleakError
 
@@ -38,7 +39,7 @@ def __init__(
 class AsyncBleWrapper(ObjectStream):
     client: BleakClient
     config: AsyncBleConfig
-    receive_stream: anyio.streams.memory.MemoryObjectReceiveStream
+    receive_stream: MemoryObjectReceiveStream
 
     async def send(self, data: bytes):
         await self.client.write_gatt_char(self.config.write_char_uuid, data)
@@ -109,7 +110,7 @@ async def connect(self):
         async with BleakClient(self.address) as client:
             try:
                 if client.is_connected:
-                    send_stream, receive_stream = anyio.create_memory_object_stream[bytearray](
+                    send_stream, receive_stream = anyio.create_memory_object_stream[bytearray](  # ty: ignore[call-non-callable]
                         max_buffer_size=1000)
                     self.logger.info(
                         "Connected to BLE device at Address: %s", self.address)

python/packages/jumpstarter-driver-ble/jumpstarter_driver_ble/driver_test.py

@@ -86,7 +86,7 @@ def test_ble_driver_connect_stream():
 
 def test_ble_notify_handler():
     """Test the notification handler puts data into the stream."""
-    send_stream, receive_stream = anyio.create_memory_object_stream[bytearray](max_buffer_size=10)
+    send_stream, receive_stream = anyio.create_memory_object_stream[bytearray](max_buffer_size=10)  # ty: ignore[call-non-callable]
     sender = MagicMock()
     test_data = bytearray(b"test_notification")
 
@@ -98,7 +98,7 @@ def test_ble_notify_handler():
 
 def test_ble_notify_handler_queue_full(capsys):
     """Test the notification handler handles a full buffer gracefully."""
-    send_stream, receive_stream = anyio.create_memory_object_stream[bytearray](max_buffer_size=1)
+    send_stream, receive_stream = anyio.create_memory_object_stream[bytearray](max_buffer_size=1)  # ty: ignore[call-non-callable]
     sender = MagicMock()
 
     # Fill the buffer

python/packages/jumpstarter-driver-doip/jumpstarter_driver_doip/driver_test.py

@@ -208,30 +208,30 @@ def test_doip_reconnect_failure(mock_doip_cls):
 
 def test_doip_missing_required_ecu_ip():
     with pytest.raises(ValidationError, match="ecu_ip"):
-        DoIP(ecu_logical_address=0x00E0)
+        DoIP(ecu_logical_address=0x00E0)  # ty: ignore[missing-argument]
 
 
 def test_doip_missing_required_ecu_logical_address():
     with pytest.raises(ValidationError, match="ecu_logical_address"):
-        DoIP(ecu_ip="192.168.1.100")
+        DoIP(ecu_ip="192.168.1.100")  # ty: ignore[missing-argument]
 
 
 def test_doip_invalid_ecu_ip_type():
     with pytest.raises(ValidationError):
-        DoIP(ecu_ip=12345, ecu_logical_address=0x00E0)
+        DoIP(ecu_ip=12345, ecu_logical_address=0x00E0)  # ty: ignore[invalid-argument-type]
 
 
 def test_doip_invalid_tcp_port_type():
     with pytest.raises(ValidationError):
-        DoIP(ecu_ip="192.168.1.100", ecu_logical_address=0x00E0, tcp_port="not_a_port")
+        DoIP(ecu_ip="192.168.1.100", ecu_logical_address=0x00E0, tcp_port="not_a_port")  # ty: ignore[invalid-argument-type]
 
 
 def test_doip_invalid_auto_reconnect_tcp_type():
     with pytest.raises(ValidationError):
         DoIP(
             ecu_ip="192.168.1.100",
             ecu_logical_address=0x00E0,
-            auto_reconnect_tcp="not_bool",
+            auto_reconnect_tcp="not_bool",  # ty: ignore[invalid-argument-type]
         )
 
 

python/packages/jumpstarter-driver-flashers/jumpstarter_driver_flashers/client.py

@@ -314,7 +314,7 @@ def _find_exception_in_chain(self, exception: Exception, target_type: type) -> E
         """
         # Check if this is an ExceptionGroup and look through its exceptions
         if hasattr(exception, "exceptions"):
-            for sub_exc in exception.exceptions:
+            for sub_exc in exception.exceptions:  # ty: ignore[not-iterable]
                 result = self._find_exception_in_chain(sub_exc, target_type)
                 if result is not None:
                     return result