Add integration tests for host (#8912)

.coveragerc

@@ -1,2 +1,4 @@
 [run]
-omit = esphome/components/*
+omit = 
+    esphome/components/*
+    tests/integration/*

.github/workflows/ci.yml

@@ -214,12 +214,12 @@ jobs:
         if: matrix.os == 'windows-latest'
         run: |
           ./venv/Scripts/activate
-          pytest -vv --cov-report=xml --tb=native tests
+          pytest -vv --cov-report=xml --tb=native -n auto tests
       - name: Run pytest
         if: matrix.os == 'ubuntu-latest' || matrix.os == 'macOS-latest'
         run: |
           . venv/bin/activate
-          pytest -vv --cov-report=xml --tb=native tests
+          pytest -vv --cov-report=xml --tb=native -n auto tests
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5.4.3
         with:

requirements_test.txt

@@ -9,5 +9,6 @@ pytest==8.3.5
 pytest-cov==6.1.1
 pytest-mock==3.14.0
 pytest-asyncio==0.26.0
+pytest-xdist==3.6.1
 asyncmock==0.4.2
 hypothesis==6.92.1

script/integration_test

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+
+set -e
+
+script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+cd "${script_dir}/.."
+
+set -x
+
+pytest -vvs --no-cov --tb=native -n 0 tests/integration/

tests/integration/README.md

@@ -0,0 +1,80 @@
+# ESPHome Integration Tests
+
+This directory contains end-to-end integration tests for ESPHome, focusing on testing the complete flow from YAML configuration to running devices with API connections.
+
+## Structure
+
+- `conftest.py` - Common fixtures and utilities
+- `const.py` - Constants used throughout the integration tests
+- `types.py` - Type definitions for fixtures and functions
+- `fixtures/` - YAML configuration files for tests
+- `test_*.py` - Individual test files
+
+## How it works
+
+### Automatic YAML Loading
+
+The `yaml_config` fixture automatically loads YAML configurations based on the test name:
+- It looks for a file named after the test function (e.g., `test_host_mode_basic` → `fixtures/host_mode_basic.yaml`)
+- The fixture file must exist or the test will fail with a clear error message
+- The fixture automatically injects a dynamic port number into the API configuration
+
+### Key Fixtures
+
+- `run_compiled` - Combines write, compile, and run operations into a single context manager
+- `api_client_connected` - Creates an API client that automatically connects using ReconnectLogic
+- `reserved_tcp_port` - Reserves a TCP port by holding the socket open until ESPHome needs it
+- `unused_tcp_port` - Provides the reserved port number for each test
+
+### Writing Tests
+
+The simplest way to write a test is to use the `run_compiled` and `api_client_connected` fixtures:
+
+```python
+@pytest.mark.asyncio
+async def test_my_feature(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    # Write, compile and run the ESPHome device, then connect to API
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        # Test your feature using the connected client
+        device_info = await client.device_info()
+        assert device_info is not None
+```
+
+### Creating YAML Fixtures
+
+Create a YAML file in the `fixtures/` directory with the same name as your test function (without the `test_` prefix):
+
+```yaml
+# fixtures/my_feature.yaml
+esphome:
+  name: my-test-device
+host:
+api:  # Port will be automatically injected
+logger:
+# Add your components here
+```
+
+## Running Tests
+
+```bash
+# Run all integration tests
+script/integration_test
+
+# Run a specific test
+pytest -vv tests/integration/test_host_mode_basic.py
+
+# Debug compilation errors or see ESPHome output
+pytest -s tests/integration/test_host_mode_basic.py
+```
+
+## Implementation Details
+
+- Tests automatically wait for the API port to be available before connecting
+- Process cleanup is handled automatically, with graceful shutdown using SIGINT
+- Each test gets its own temporary directory and unique port
+- Port allocation minimizes race conditions by holding the socket until just before ESPHome starts
+- Output from ESPHome processes is displayed for debugging

tests/integration/__init__.py

@@ -0,0 +1,3 @@
+"""ESPHome integration tests."""
+
+from __future__ import annotations

tests/integration/conftest.py

@@ -0,0 +1,402 @@
+"""Common fixtures for integration tests."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncGenerator, Generator
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from pathlib import Path
+import platform
+import signal
+import socket
+import tempfile
+
+from aioesphomeapi import APIClient, APIConnectionError, ReconnectLogic
+import pytest
+import pytest_asyncio
+
+# Skip all integration tests on Windows
+if platform.system() == "Windows":
+    pytest.skip(
+        "Integration tests are not supported on Windows", allow_module_level=True
+    )
+
+from .const import (
+    API_CONNECTION_TIMEOUT,
+    DEFAULT_API_PORT,
+    LOCALHOST,
+    PORT_POLL_INTERVAL,
+    PORT_WAIT_TIMEOUT,
+    SIGINT_TIMEOUT,
+    SIGTERM_TIMEOUT,
+)
+from .types import (
+    APIClientConnectedFactory,
+    APIClientFactory,
+    CompileFunction,
+    ConfigWriter,
+    RunCompiledFunction,
+    RunFunction,
+)
+
+
+@pytest.fixture
+def integration_test_dir() -> Generator[Path]:
+    """Create a temporary directory for integration tests."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield Path(tmpdir)
+
+
+@pytest.fixture
+def reserved_tcp_port() -> Generator[tuple[int, socket.socket]]:
+    """Reserve an unused TCP port by holding the socket open."""
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.bind(("", 0))
+    port = s.getsockname()[1]
+    try:
+        yield port, s
+    finally:
+        s.close()
+
+
+@pytest.fixture
+def unused_tcp_port(reserved_tcp_port: tuple[int, socket.socket]) -> int:
+    """Get the reserved TCP port number."""
+    return reserved_tcp_port[0]
+
+
+@pytest_asyncio.fixture
+async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> str:
+    """Load YAML configuration based on test name."""
+    # Get the test function name
+    test_name: str = request.node.name
+    # Extract the base test name (remove test_ prefix and any parametrization)
+    base_name = test_name.replace("test_", "").partition("[")[0]
+
+    # Load the fixture file
+    fixture_path = Path(__file__).parent / "fixtures" / f"{base_name}.yaml"
+    if not fixture_path.exists():
+        raise FileNotFoundError(f"Fixture file not found: {fixture_path}")
+
+    loop = asyncio.get_running_loop()
+    content = await loop.run_in_executor(None, fixture_path.read_text)
+
+    # Replace the port in the config if it contains api section
+    if "api:" in content:
+        # Add port configuration after api:
+        content = content.replace("api:", f"api:\n  port: {unused_tcp_port}")
+
+    return content
+
+
+@pytest_asyncio.fixture
+async def write_yaml_config(
+    integration_test_dir: Path, request: pytest.FixtureRequest
+) -> AsyncGenerator[ConfigWriter]:
+    """Write YAML configuration to a file."""
+    # Get the test name for default filename
+    test_name = request.node.name
+    base_name = test_name.replace("test_", "").split("[")[0]
+
+    async def _write_config(content: str, filename: str | None = None) -> Path:
+        if filename is None:
+            filename = f"{base_name}.yaml"
+        config_path = integration_test_dir / filename
+        loop = asyncio.get_running_loop()
+        await loop.run_in_executor(None, config_path.write_text, content)
+        return config_path
+
+    yield _write_config
+
+
+async def _run_esphome_command(
+    command: str,
+    config_path: Path,
+    cwd: Path,
+) -> asyncio.subprocess.Process:
+    """Run an ESPHome command with the given arguments."""
+    return await asyncio.create_subprocess_exec(
+        "esphome",
+        command,
+        str(config_path),
+        cwd=cwd,
+        stdout=None,  # Inherit stdout
+        stderr=None,  # Inherit stderr
+        stdin=asyncio.subprocess.DEVNULL,
+        # Start in a new process group to isolate signal handling
+        start_new_session=True,
+    )
+
+
+@pytest_asyncio.fixture
+async def compile_esphome(
+    integration_test_dir: Path,
+) -> AsyncGenerator[CompileFunction]:
+    """Compile an ESPHome configuration."""
+
+    async def _compile(config_path: Path) -> None:
+        proc = await _run_esphome_command("compile", config_path, integration_test_dir)
+        await proc.wait()
+        if proc.returncode != 0:
+            raise RuntimeError(
+                f"Failed to compile {config_path}, return code: {proc.returncode}. "
+                f"Run with 'pytest -s' to see compilation output."
+            )
+
+    yield _compile
+
+
+@pytest_asyncio.fixture
+async def run_esphome_process(
+    integration_test_dir: Path,
+) -> AsyncGenerator[RunFunction]:
+    """Run an ESPHome process and manage its lifecycle."""
+    processes: list[asyncio.subprocess.Process] = []
+
+    async def _run(config_path: Path) -> asyncio.subprocess.Process:
+        process = await _run_esphome_command("run", config_path, integration_test_dir)
+        processes.append(process)
+        return process
+
+    yield _run
+
+    # Cleanup: terminate all "run" processes gracefully
+    for process in processes:
+        if process.returncode is None:
+            # Send SIGINT (Ctrl+C) for graceful shutdown of the running ESPHome instance
+            process.send_signal(signal.SIGINT)
+            try:
+                await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
+            except asyncio.TimeoutError:
+                # If SIGINT didn't work, try SIGTERM
+                process.terminate()
+                try:
+                    await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
+                except asyncio.TimeoutError:
+                    # Last resort: SIGKILL
+                    process.kill()
+                    await process.wait()
+
+
+@asynccontextmanager
+async def create_api_client(
+    address: str = LOCALHOST,
+    port: int = DEFAULT_API_PORT,
+    password: str = "",
+    noise_psk: str | None = None,
+    client_info: str = "integration-test",
+) -> AsyncGenerator[APIClient]:
+    """Create an API client context manager."""
+    client = APIClient(
+        address=address,
+        port=port,
+        password=password,
+        noise_psk=noise_psk,
+        client_info=client_info,
+    )
+    try:
+        yield client
+    finally:
+        await client.disconnect()
+
+
+@pytest_asyncio.fixture
+async def api_client_factory(
+    unused_tcp_port: int,
+) -> AsyncGenerator[APIClientFactory]:
+    """Factory for creating API client context managers."""
+
+    def _create_client(
+        address: str = LOCALHOST,
+        port: int | None = None,
+        password: str = "",
+        noise_psk: str | None = None,
+        client_info: str = "integration-test",
+    ) -> AbstractAsyncContextManager[APIClient]:
+        return create_api_client(
+            address=address,
+            port=port if port is not None else unused_tcp_port,
+            password=password,
+            noise_psk=noise_psk,
+            client_info=client_info,
+        )
+
+    yield _create_client
+
+
+@asynccontextmanager
+async def wait_and_connect_api_client(
+    address: str = LOCALHOST,
+    port: int = DEFAULT_API_PORT,
+    password: str = "",
+    noise_psk: str | None = None,
+    client_info: str = "integration-test",
+    timeout: float = API_CONNECTION_TIMEOUT,
+) -> AsyncGenerator[APIClient]:
+    """Wait for API to be available and connect."""
+    client = APIClient(
+        address=address,
+        port=port,
+        password=password,
+        noise_psk=noise_psk,
+        client_info=client_info,
+    )
+
+    # Create a future to signal when connected
+    loop = asyncio.get_running_loop()
+    connected_future: asyncio.Future[None] = loop.create_future()
+
+    async def on_connect() -> None:
+        """Called when successfully connected."""
+        if not connected_future.done():
+            connected_future.set_result(None)
+
+    async def on_disconnect(expected_disconnect: bool) -> None:
+        """Called when disconnected."""
+        if not connected_future.done() and not expected_disconnect:
+            connected_future.set_exception(
+                APIConnectionError("Disconnected before fully connected")
+            )
+
+    async def on_connect_error(err: Exception) -> None:
+        """Called when connection fails."""
+        if not connected_future.done():
+            connected_future.set_exception(err)
+
+    # Create and start the reconnect logic
+    reconnect_logic = ReconnectLogic(
+        client=client,
+        on_connect=on_connect,
+        on_disconnect=on_disconnect,
+        zeroconf_instance=None,  # Not using zeroconf for integration tests
+        name=f"{address}:{port}",
+        on_connect_error=on_connect_error,
+    )
+
+    try:
+        # Start the connection
+        await reconnect_logic.start()
+
+        # Wait for connection with timeout
+        try:
+            await asyncio.wait_for(connected_future, timeout=timeout)
+        except asyncio.TimeoutError:
+            raise TimeoutError(f"Failed to connect to API after {timeout} seconds")
+
+        yield client
+    finally:
+        # Stop reconnect logic and disconnect
+        await reconnect_logic.stop()
+        await client.disconnect()
+
+
+@pytest_asyncio.fixture
+async def api_client_connected(
+    unused_tcp_port: int,
+) -> AsyncGenerator[APIClientConnectedFactory]:
+    """Factory for creating connected API client context managers."""
+
+    def _connect_client(
+        address: str = LOCALHOST,
+        port: int | None = None,
+        password: str = "",
+        noise_psk: str | None = None,
+        client_info: str = "integration-test",
+        timeout: float = API_CONNECTION_TIMEOUT,
+    ) -> AbstractAsyncContextManager[APIClient]:
+        return wait_and_connect_api_client(
+            address=address,
+            port=port if port is not None else unused_tcp_port,
+            password=password,
+            noise_psk=noise_psk,
+            client_info=client_info,
+            timeout=timeout,
+        )
+
+    yield _connect_client
+
+
+async def wait_for_port_open(
+    host: str, port: int, timeout: float = PORT_WAIT_TIMEOUT
+) -> None:
+    """Wait for a TCP port to be open and accepting connections."""
+    loop = asyncio.get_running_loop()
+    start_time = loop.time()
+
+    # Small yield to ensure the process has a chance to start
+    await asyncio.sleep(0)
+
+    while loop.time() - start_time < timeout:
+        try:
+            # Try to connect to the port
+            _, writer = await asyncio.open_connection(host, port)
+            writer.close()
+            await writer.wait_closed()
+            return  # Port is open
+        except (ConnectionRefusedError, OSError):
+            # Port not open yet, wait a bit and try again
+            await asyncio.sleep(PORT_POLL_INTERVAL)
+
+    raise TimeoutError(f"Port {port} on {host} did not open within {timeout} seconds")
+
+
+@asynccontextmanager
+async def run_compiled_context(
+    yaml_content: str,
+    filename: str | None,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    run_esphome_process: RunFunction,
+    port: int,
+    port_socket: socket.socket | None = None,
+) -> AsyncGenerator[asyncio.subprocess.Process]:
+    """Context manager to write, compile and run an ESPHome configuration."""
+    # Write the YAML config
+    config_path = await write_yaml_config(yaml_content, filename)
+
+    # Compile the configuration
+    await compile_esphome(config_path)
+
+    # Close the port socket right before running to release the port
+    if port_socket is not None:
+        port_socket.close()
+
+    # Run the ESPHome device
+    process = await run_esphome_process(config_path)
+    assert process.returncode is None, "Process died immediately"
+
+    # Wait for the API server to start listening
+    await wait_for_port_open(LOCALHOST, port, timeout=PORT_WAIT_TIMEOUT)
+
+    try:
+        yield process
+    finally:
+        # Process cleanup is handled by run_esphome_process fixture
+        pass
+
+
+@pytest_asyncio.fixture
+async def run_compiled(
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    run_esphome_process: RunFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> AsyncGenerator[RunCompiledFunction]:
+    """Write, compile and run an ESPHome configuration."""
+    port, port_socket = reserved_tcp_port
+
+    def _run_compiled(
+        yaml_content: str, filename: str | None = None
+    ) -> AbstractAsyncContextManager[asyncio.subprocess.Process]:
+        return run_compiled_context(
+            yaml_content,
+            filename,
+            write_yaml_config,
+            compile_esphome,
+            run_esphome_process,
+            port,
+            port_socket,
+        )
+
+    yield _run_compiled

tests/integration/const.py

@@ -0,0 +1,14 @@
+"""Constants for integration tests."""
+
+# Network constants
+DEFAULT_API_PORT = 6053
+LOCALHOST = "localhost"
+
+# Timeout constants
+API_CONNECTION_TIMEOUT = 30.0  # seconds
+PORT_WAIT_TIMEOUT = 30.0  # seconds
+PORT_POLL_INTERVAL = 0.1  # seconds
+
+# Process shutdown timeouts
+SIGINT_TIMEOUT = 5.0  # seconds
+SIGTERM_TIMEOUT = 2.0  # seconds

tests/integration/fixtures/host_mode_basic.yaml

@@ -0,0 +1,5 @@
+esphome:
+  name: host-test
+host:
+api:
+logger:

tests/integration/fixtures/host_mode_noise_encryption.yaml

@@ -0,0 +1,7 @@
+esphome:
+  name: host-noise-test
+host:
+api:
+  encryption:
+    key: N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU=
+logger:

tests/integration/fixtures/host_mode_noise_encryption_wrong_key.yaml

@@ -0,0 +1,7 @@
+esphome:
+  name: host-noise-test
+host:
+api:
+  encryption:
+    key: N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU=
+logger:

tests/integration/fixtures/host_mode_reconnect.yaml

@@ -0,0 +1,5 @@
+esphome:
+  name: host-reconnect-test
+host:
+api:
+logger:

tests/integration/fixtures/host_mode_with_sensor.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: host-sensor-test
+host:
+api:
+logger:
+sensor:
+  - platform: template
+    name: Test Sensor
+    id: test_sensor
+    unit_of_measurement: °C
+    lambda: return 42.0;
+    update_interval: 0.1s

tests/integration/test_host_mode_basic.py

@@ -0,0 +1,22 @@
+"""Basic integration test for Host mode."""
+
+from __future__ import annotations
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_host_mode_basic(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test basic Host mode functionality with API connection."""
+    # Write, compile and run the ESPHome device, then connect to API
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        # Verify we can get device info
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "host-test"

tests/integration/test_host_mode_noise_encryption.py

@@ -0,0 +1,53 @@
+"""Integration test for Host mode with noise encryption."""
+
+from __future__ import annotations
+
+from aioesphomeapi import InvalidEncryptionKeyAPIError
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+# The API key for noise encryption
+NOISE_KEY = "N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU="
+
+
+@pytest.mark.asyncio
+async def test_host_mode_noise_encryption(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test Host mode with noise encryption enabled."""
+    # Write, compile and run the ESPHome device, then connect to API
+    # The API client should handle noise encryption automatically
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected(noise_psk=NOISE_KEY) as client,
+    ):
+        # If we can get device info, the encryption is working
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "host-noise-test"
+
+        # List entities to ensure the encrypted connection is fully functional
+        entities = await client.list_entities_services()
+        assert entities is not None
+
+
+@pytest.mark.asyncio
+async def test_host_mode_noise_encryption_wrong_key(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that connection fails with wrong encryption key."""
+    # Write, compile and run the ESPHome device
+    async with run_compiled(yaml_config):
+        # Try to connect with wrong key - should fail with InvalidEncryptionKeyAPIError
+        with pytest.raises(InvalidEncryptionKeyAPIError):
+            async with api_client_connected(
+                noise_psk="wrong_key_that_should_not_work",
+                timeout=5,  # Shorter timeout for expected failure
+            ) as client:
+                # This should not be reached
+                await client.device_info()

tests/integration/test_host_mode_reconnect.py

@@ -0,0 +1,28 @@
+"""Integration test for Host mode reconnection."""
+
+from __future__ import annotations
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_host_mode_reconnect(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test reconnecting to a Host mode device."""
+    # Write, compile and run the ESPHome device
+    async with run_compiled(yaml_config):
+        # First connection
+        async with api_client_connected() as client:
+            device_info = await client.device_info()
+            assert device_info is not None
+
+        # Reconnect with a new client
+        async with api_client_connected() as client2:
+            device_info2 = await client2.device_info()
+            assert device_info2 is not None
+            assert device_info2.name == device_info.name

tests/integration/test_host_mode_sensor.py

@@ -0,0 +1,49 @@
+"""Integration test for Host mode with sensor."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import EntityState
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_host_mode_with_sensor(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test Host mode with a sensor component."""
+    # Write, compile and run the ESPHome device, then connect to API
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        # Subscribe to state changes
+        states: dict[int, EntityState] = {}
+        sensor_future: asyncio.Future[EntityState] = asyncio.Future()
+
+        def on_state(state: EntityState) -> None:
+            states[state.key] = state
+            # If this is our sensor with value 42.0, resolve the future
+            if (
+                hasattr(state, "state")
+                and state.state == 42.0
+                and not sensor_future.done()
+            ):
+                sensor_future.set_result(state)
+
+        client.subscribe_states(on_state)
+
+        # Wait for sensor with specific value (42.0) with timeout
+        try:
+            test_sensor_state = await asyncio.wait_for(sensor_future, timeout=5.0)
+        except asyncio.TimeoutError:
+            pytest.fail(
+                f"Sensor with value 42.0 not received within 5 seconds. "
+                f"Received states: {list(states.values())}"
+            )
+
+        # Verify the sensor state
+        assert test_sensor_state.state == 42.0
+        assert len(states) > 0, "No states received"

tests/integration/types.py

@@ -0,0 +1,46 @@
+"""Type definitions for integration tests."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from contextlib import AbstractAsyncContextManager
+from pathlib import Path
+from typing import Protocol
+
+from aioesphomeapi import APIClient
+
+ConfigWriter = Callable[[str, str | None], Awaitable[Path]]
+CompileFunction = Callable[[Path], Awaitable[None]]
+RunFunction = Callable[[Path], Awaitable[asyncio.subprocess.Process]]
+RunCompiledFunction = Callable[
+    [str, str | None], AbstractAsyncContextManager[asyncio.subprocess.Process]
+]
+WaitFunction = Callable[[APIClient, float], Awaitable[bool]]
+
+
+class APIClientFactory(Protocol):
+    """Protocol for API client factory."""
+
+    def __call__(  # noqa: E704
+        self,
+        address: str = "localhost",
+        port: int | None = None,
+        password: str = "",
+        noise_psk: str | None = None,
+        client_info: str = "integration-test",
+    ) -> AbstractAsyncContextManager[APIClient]: ...
+
+
+class APIClientConnectedFactory(Protocol):
+    """Protocol for connected API client factory."""
+
+    def __call__(  # noqa: E704
+        self,
+        address: str = "localhost",
+        port: int | None = None,
+        password: str = "",
+        noise_psk: str | None = None,
+        client_info: str = "integration-test",
+        timeout: float = 30,
+    ) -> AbstractAsyncContextManager[APIClient]: ...