Run pytest in parallel with pytest-xdist (#6825)

* Run pytest in parallel with pytest-xdist

CI executes the full pytest suite serially, which currently takes
around 4-5 minutes. Most of that time is spent in fixture setup
(D-Bus session, mock services, CoreSys construction) rather than the
test bodies, but each test pays this setup cost on its own worker.

Add pytest-xdist and run with -n auto --dist=loadfile so tests are
distributed across CPU cores while keeping all tests of a file on the
same worker (preserves locality and keeps progress output readable).
GitHub Actions standard runners ship 4 vCPUs, so -n auto picks 4
workers in CI; locally on 8-core machines it picks 8. This matches
the pattern Home Assistant Core has been running for a long time
(--numprocesses auto --dist=loadfile in core's pytest-full job),
so the configuration is already battle-tested in a sibling project.

On an 8-core machine this cuts the local run from ~280s to ~88s
(~3.2x); on the 4-vCPU CI runner expect roughly a ~2x reduction.

The dbus-daemon and other session-scoped fixtures are spawned per
worker, so there is no shared state. pytest-cov already handles xdist
worker coverage merging via the standard .coverage.* files.

* Poll for resolution state in datadisk signal tests

test_multiple_datadisk_add_remove_signals and
test_disabled_datadisk_add_remove_signals fire UDisks2
InterfacesAdded/InterfacesRemoved signals through the real session
dbus-daemon and then assert that supervisor's signal handler chain
has updated coresys.resolution.issues. Each assertion was preceded
by ``await udisks2_service.ping()`` (only confirms signal delivery)
and ``await asyncio.sleep(0.2)`` to let the chained async tasks
finish.

The 0.2 s margin was effectively only 0.1 s of slack:
DataDisk._udisks2_interface_added itself does
``await asyncio.sleep(0.1)`` internally to wait for
UDisks2._interfaces_added (a sibling subscriber on the same signal)
to finish updating the block-device cache before the check runs.
Under xdist parallelism on a 4-vCPU CI runner that 100 ms cushion
evaporates, and the assertion races the handler.

Bumping the test sleep would just kick the can. Instead, replace
the four ``sleep+assert`` sites with a small polling helper
(tests.common.wait_for) that re-checks the predicate every 10 ms
up to a 5 s deadline. The wait completes the moment the resolution
state matches, so the test stays fast on idle and is robust under
load — and the assertion's failure mode becomes a clear "predicate
did not become true within 5 s" instead of a value mismatch.

The product-code sleep inside _udisks2_interface_added is still a
real smell (handler ordering shouldn't depend on a fixed sleep)
but is left for a separate fix; this commit is scoped to the test.

.github/workflows/ci.yaml

@@ -13,8 +13,8 @@ env:
   MYPY_CACHE_VERSION: 1
 
 concurrency:
-  group: "${{ github.workflow }}-${{ github.ref }}"
-  cancel-in-progress: true
+  group: ${{ github.workflow }}-${{ github.ref }}-${{ github.event_name == 'pull_request' && 'pr' || github.run_id }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
 
 jobs:
   # Separate job to pre-populate the base dependency cache
@@ -331,7 +331,7 @@ jobs:
       - name: Run mypy
         run: |
           . venv/bin/activate
-          mypy --ignore-missing-imports supervisor
+          mypy supervisor
 
   pytest:
     runs-on: ubuntu-latest
@@ -380,6 +380,8 @@ jobs:
           . venv/bin/activate
           pytest \
             -qq \
+            -n auto \
+            --dist=loadfile \
             --durations=10 \
             --cov supervisor \
             -o console_output_style=count \

pyproject.toml

@@ -223,6 +223,13 @@ overgeneral-exceptions = ["builtins.BaseException", "builtins.Exception"]
 [tool.pylint.DESIGN]
 max-positional-arguments = 10
 
+[tool.mypy]
+warn_unreachable = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+enable_error_code = ["exhaustive-match", "possibly-undefined"]
+disable_error_code = ["import-not-found", "import-untyped"]
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 norecursedirs = [".git"]

requirements_tests.txt

@@ -7,8 +7,9 @@ pytest-aiohttp==1.1.0
 pytest-asyncio==1.3.0
 pytest-cov==7.1.0
 pytest-timeout==2.4.0
+pytest-xdist==3.8.0
 pytest==9.0.3
 ruff==0.15.12
 time-machine==3.2.0
-types-pyyaml==6.0.12.20260408
-urllib3==2.6.3
+types-pyyaml==6.0.12.20260508
+urllib3==2.7.0

supervisor/coresys.py

@@ -632,12 +632,10 @@ class CoreSys:
         self, coroutine: Coroutine, *, eager_start: bool | None = None
     ) -> asyncio.Task:
         """Create an async task."""
-        # eager_start kwarg works but wasn't added for mypy visibility until 3.14
-        # can remove the type ignore then
         return self.loop.create_task(
             coroutine,
             context=self._create_context(),
-            eager_start=eager_start,  # type: ignore
+            eager_start=eager_start,
         )
 
     def call_later(

supervisor/docker/interface.py

@@ -332,15 +332,17 @@ class DockerInterface(JobGroup, ABC):
                 )
                 await async_capture_exception(err)
 
+        # Get credentials for private registries to pass to aiodocker.
+        # Done before registering the listener so a failure here does not
+        # leak a stale event listener.
+        credentials, pull_image_name = self._get_credentials(image)
+
         listener = self.sys_bus.register_event(
             BusEvent.DOCKER_IMAGE_PULL_UPDATE, process_pull_event
         )
 
         _LOGGER.info("Downloading docker image %s with tag %s.", image, version)
         try:
-            # Get credentials for private registries to pass to aiodocker
-            credentials, pull_image_name = self._get_credentials(image)
-
             # Pull new image, passing credentials to aiodocker
             docker_image = await self.sys_docker.pull_image(
                 current_job.uuid,

supervisor/host/configuration.py

@@ -267,7 +267,8 @@ class Interface:
                 return InterfaceMethod.AUTO
             case NMInterfaceMethod.MANUAL:
                 return InterfaceMethod.STATIC
-        return InterfaceMethod.DISABLED
+            case _:
+                return InterfaceMethod.DISABLED
 
     @staticmethod
     def _map_nm_addr_gen_mode(addr_gen_mode: int | None) -> InterfaceAddrGenMode:
@@ -333,7 +334,8 @@ class Interface:
                 return InterfaceType.WIRELESS
             case DeviceType.VLAN.value:
                 return InterfaceType.VLAN
-        raise ValueError(f"Invalid device type: {device_type}")
+            case _:
+                raise ValueError(f"Invalid device type: {device_type}")
 
     @staticmethod
     def _map_nm_wifi(inet: NetworkInterface) -> WifiConfig | None:

supervisor/misc/scheduler.py

@@ -73,26 +73,22 @@ class Scheduler(CoreSysAttributes):
 
     def _schedule_task(self, task: _Task) -> None:
         """Schedule a task on loop."""
-        if isinstance(task.interval, (int, float)):
-            task.next = self.sys_call_later(task.interval, self._run_task, task)
-        elif isinstance(task.interval, time):
-            today = datetime.combine(date.today(), task.interval)
-            tomorrow = datetime.combine(date.today() + timedelta(days=1), task.interval)
+        match task.interval:
+            case int() | float():
+                task.next = self.sys_call_later(task.interval, self._run_task, task)
+            case time():
+                today = datetime.combine(date.today(), task.interval)
+                tomorrow = datetime.combine(
+                    date.today() + timedelta(days=1), task.interval
+                )
 
-            # Check if we run it today or next day
-            if today > datetime.today():
-                calc = today
-            else:
-                calc = tomorrow
+                # Check if we run it today or next day
+                if today > datetime.today():
+                    calc = today
+                else:
+                    calc = tomorrow
 
-            task.next = self.sys_call_at(calc, self._run_task, task)
-        else:
-            _LOGGER.critical(
-                "Unknown interval %s (type: %s) for scheduler %s",
-                task.interval,
-                type(task.interval),
-                task.id,
-            )
+                task.next = self.sys_call_at(calc, self._run_task, task)
 
     async def shutdown(self, timeout=10) -> None:
         """Shutdown all task inside the scheduler."""

supervisor/mounts/mount.py

@@ -151,7 +151,8 @@ class Mount(CoreSysAttributes, ABC):
                 return PurePath(PATH_MEDIA, self.name)
             case MountUsage.SHARE:
                 return PurePath(PATH_SHARE, self.name)
-        return None
+            case MountUsage.BACKUP | None:
+                return None
 
     @property
     def failed_issue(self) -> Issue:

supervisor/os/manager.py

@@ -4,7 +4,6 @@ from dataclasses import dataclass
 from datetime import datetime
 import logging
 from pathlib import Path, PurePath
-from typing import cast
 
 import aiohttp
 from awesomeversion import AwesomeVersion, AwesomeVersionException
@@ -58,8 +57,8 @@ class SlotStatus:
             device=PurePath(data["device"]),
             bundle_compatible=data.get("bundle.compatible"),
             sha256=data.get("sha256"),
-            size=cast(int | None, data.get("size")),
-            installed_count=cast(int | None, data.get("installed.count")),
+            size=data.get("size"),
+            installed_count=data.get("installed.count"),
             bundle_version=AwesomeVersion(data["bundle.version"])
             if "bundle.version" in data
             else None,
@@ -67,7 +66,7 @@ class SlotStatus:
             if "installed.timestamp" in data
             else None,
             status=data.get("status"),
-            activated_count=cast(int | None, data.get("activated.count")),
+            activated_count=data.get("activated.count"),
             activated_timestamp=datetime.fromisoformat(data["activated.timestamp"])
             if "activated.timestamp" in data
             else None,

supervisor/store/repository.py

@@ -186,8 +186,7 @@ class RepositoryGit(Repository, ABC):
                 repository_file = Path(self._git.path / f"repository{filetype}")
                 if repository_file.exists():
                     break
-
-            if not repository_file.exists():
+            else:
                 return False
 
             # If valid?

tests/common.py

@@ -1,7 +1,7 @@
 """Common test functions."""
 
 import asyncio
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from datetime import datetime
 from functools import partial
 from importlib import import_module
@@ -32,6 +32,29 @@ async def fire_bus_event(coresys: CoreSys, event: BusEvent, data: Any) -> None:
     await asyncio.gather(*coresys.bus.fire_event(event, data))
 
 
+async def wait_for(
+    predicate: Callable[[], bool],
+    *,
+    timeout: float = 5.0,
+    interval: float = 0.01,
+) -> None:
+    """Poll a synchronous predicate until truthy or the deadline elapses.
+
+    Useful when a test fires a D-Bus signal (or another out-of-band
+    trigger) and needs to observe state mutated by the resulting async
+    chain — e.g. a signal handler that schedules its own follow-up
+    tasks. Completes the moment the predicate is true, so the wait
+    costs only what's actually needed; this avoids the choice between a
+    fixed sleep that's fast on idle and racy under load and a fixed
+    sleep that's robust under load and wasteful on idle.
+    """
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not predicate():
+        if asyncio.get_running_loop().time() >= deadline:
+            raise AssertionError(f"Predicate did not become true within {timeout}s")
+        await asyncio.sleep(interval)
+
+
 def get_fixture_path(filename: str) -> Path:
     """Get path for fixture."""
     return Path(Path(__file__).parent.joinpath("fixtures"), filename)

tests/os/test_data_disk.py

@@ -1,6 +1,5 @@
 """Test OS API."""
 
-import asyncio
 from dataclasses import replace
 from pathlib import PosixPath
 from unittest.mock import patch
@@ -16,7 +15,7 @@ from supervisor.os.data_disk import Disk
 from supervisor.resolution.const import ContextType, IssueType
 from supervisor.resolution.data import Issue
 
-from tests.common import mock_dbus_services
+from tests.common import mock_dbus_services, wait_for
 from tests.dbus_service_mocks.agent_datadisk import DataDisk as DataDiskService
 from tests.dbus_service_mocks.agent_system import System as SystemService
 from tests.dbus_service_mocks.base import DBusServiceMock
@@ -357,11 +356,13 @@ async def test_multiple_datadisk_add_remove_signals(
         },
     )
     await udisks2_service.ping()
-    await asyncio.sleep(0.2)
-
-    assert (
-        Issue(IssueType.MULTIPLE_DATA_DISKS, ContextType.SYSTEM, reference="/dev/sdb1")
-        in coresys.resolution.issues
+    await wait_for(
+        lambda: (
+            Issue(
+                IssueType.MULTIPLE_DATA_DISKS, ContextType.SYSTEM, reference="/dev/sdb1"
+            )
+            in coresys.resolution.issues
+        )
     )
 
     udisks2_service.InterfacesRemoved(
@@ -369,9 +370,7 @@ async def test_multiple_datadisk_add_remove_signals(
         ["org.freedesktop.UDisks2.Block", "org.freedesktop.UDisks2.Filesystem"],
     )
     await udisks2_service.ping()
-    await asyncio.sleep(0.2)
-
-    assert coresys.resolution.issues == []
+    await wait_for(lambda: coresys.resolution.issues == [])
 
 
 async def test_disabled_datadisk_add_remove_signals(
@@ -409,11 +408,13 @@ async def test_disabled_datadisk_add_remove_signals(
         },
     )
     await udisks2_service.ping()
-    await asyncio.sleep(0.2)
-
-    assert (
-        Issue(IssueType.DISABLED_DATA_DISK, ContextType.SYSTEM, reference="/dev/sdb1")
-        in coresys.resolution.issues
+    await wait_for(
+        lambda: (
+            Issue(
+                IssueType.DISABLED_DATA_DISK, ContextType.SYSTEM, reference="/dev/sdb1"
+            )
+            in coresys.resolution.issues
+        )
     )
 
     udisks2_service.InterfacesRemoved(
@@ -421,6 +422,4 @@ async def test_disabled_datadisk_add_remove_signals(
         ["org.freedesktop.UDisks2.Block", "org.freedesktop.UDisks2.Filesystem"],
     )
     await udisks2_service.ping()
-    await asyncio.sleep(0.2)
-
-    assert coresys.resolution.issues == []
+    await wait_for(lambda: coresys.resolution.issues == [])