esphome/tests/integration

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:

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

# fixtures/my_feature.yaml
esphome:
  name: my-test-device
host:
api:  # Port will be automatically injected
logger:
# Add your components here

Running Tests

# 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

Integration Test Writing Guide

Test Patterns and Best Practices

1. Test File Naming Convention

• Use descriptive names: test_{category}_{feature}.py
• Common categories: host_mode, api, scheduler, light, areas_and_devices
• Examples:
  • test_host_mode_basic.py - Basic host mode functionality
  • test_api_message_batching.py - API message batching
  • test_scheduler_stress.py - Scheduler stress testing

2. Essential Imports

from __future__ import annotations

import asyncio
from typing import Any

import pytest
from aioesphomeapi import EntityState, SensorState

from .types import APIClientConnectedFactory, RunCompiledFunction

3. Common Test Patterns

Basic Entity Test

@pytest.mark.asyncio
async def test_my_sensor(
    yaml_config: str,
    run_compiled: RunCompiledFunction,
    api_client_connected: APIClientConnectedFactory,
) -> None:
    """Test sensor functionality."""
    async with run_compiled(yaml_config), api_client_connected() as client:
        # Get entity list
        entities, services = await client.list_entities_services()

        # Find specific entity
        sensor = next((e for e in entities if e.object_id == "my_sensor"), None)
        assert sensor is not None

State Subscription Pattern

# Track state changes with futures
loop = asyncio.get_running_loop()
states: dict[int, EntityState] = {}
state_future: asyncio.Future[EntityState] = loop.create_future()

def on_state(state: EntityState) -> None:
    states[state.key] = state
    # Check for specific condition using isinstance
    if isinstance(state, SensorState) and state.state == expected_value:
        if not state_future.done():
            state_future.set_result(state)

client.subscribe_states(on_state)

# Wait for state with timeout
try:
    result = await asyncio.wait_for(state_future, timeout=5.0)
except asyncio.TimeoutError:
    pytest.fail(f"Expected state not received. Got: {list(states.values())}")

Service Execution Pattern

# Find and execute service
entities, services = await client.list_entities_services()
my_service = next((s for s in services if s.name == "my_service"), None)
assert my_service is not None

# Execute with parameters
client.execute_service(my_service, {"param1": "value1", "param2": 42})

Multiple Entity Tracking

# For tests with many entities
loop = asyncio.get_running_loop()
entity_count = 50
received_states: set[int] = set()
all_states_future: asyncio.Future[bool] = loop.create_future()

def on_state(state: EntityState) -> None:
    received_states.add(state.key)
    if len(received_states) >= entity_count and not all_states_future.done():
        all_states_future.set_result(True)

client.subscribe_states(on_state)
await asyncio.wait_for(all_states_future, timeout=10.0)

4. YAML Fixture Guidelines

Naming Convention

• Match test function name: test_my_feature → fixtures/my_feature.yaml
• Note: Remove test_ prefix for fixture filename

Basic Structure

esphome:
  name: test-name  # Use kebab-case
  # Optional: areas, devices, platformio_options

host:  # Always use host platform for integration tests
api:   # Port injected automatically
logger:
  level: DEBUG  # Optional: Set log level

# Component configurations
sensor:
  - platform: template
    name: "My Sensor"
    id: my_sensor
    lambda: return 42.0;
    update_interval: 0.1s  # Fast updates for testing

Advanced Features

# External components for custom test code
external_components:
  - source:
      type: local
      path: EXTERNAL_COMPONENT_PATH  # Replaced by test framework
    components: [my_test_component]

# Areas and devices
esphome:
  name: test-device
  areas:
    - id: living_room
      name: "Living Room"
    - id: kitchen
      name: "Kitchen"
      parent_id: living_room
  devices:
    - id: my_device
      name: "Test Device"
      area_id: living_room

# API services
api:
  services:
    - service: test_service
      variables:
        my_param: string
      then:
        - logger.log:
            format: "Service called with: %s"
            args: [my_param.c_str()]

5. Testing Complex Scenarios

External Components

Create C++ components in fixtures/external_components/ for:

• Stress testing
• Custom entity behaviors
• Scheduler testing
• Memory management tests

Log Line Monitoring

log_lines: list[str] = []

def on_log_line(line: str) -> None:
    log_lines.append(line)
    if "expected message" in line:
        # Handle specific log messages

async with run_compiled(yaml_config, line_callback=on_log_line):
    # Test implementation

Example using futures for specific log patterns:

import re

loop = asyncio.get_running_loop()
connected_future = loop.create_future()
service_future = loop.create_future()

# Patterns to match
connected_pattern = re.compile(r"Client .* connected from")
service_pattern = re.compile(r"Service called")

def check_output(line: str) -> None:
    """Check log output for expected messages."""
    if not connected_future.done() and connected_pattern.search(line):
        connected_future.set_result(True)
    elif not service_future.done() and service_pattern.search(line):
        service_future.set_result(True)

async with run_compiled(yaml_config, line_callback=check_output):
    async with api_client_connected() as client:
        # Wait for specific log message
        await asyncio.wait_for(connected_future, timeout=5.0)

        # Do test actions...

        # Wait for service log
        await asyncio.wait_for(service_future, timeout=5.0)

Note: Tests that monitor log messages typically have fewer race conditions compared to state-based testing, making them more reliable. However, be aware that the host platform currently does not have a thread-safe logger, so logging from threads will not work correctly.

Timeout Handling

# Always use timeouts for async operations
try:
    result = await asyncio.wait_for(some_future, timeout=5.0)
except asyncio.TimeoutError:
    pytest.fail("Operation timed out - check test expectations")

6. Common Assertions

# Device info
assert device_info.name == "expected-name"
assert device_info.compilation_time is not None

# Entity properties
assert sensor.accuracy_decimals == 2
assert sensor.state_class == 1  # measurement
assert sensor.force_update is True

# Service availability
assert len(services) > 0
assert any(s.name == "expected_service" for s in services)

# State values
assert state.state == expected_value
assert state.missing_state is False

7. Debugging Tips

• Use pytest -s to see ESPHome output during tests
• Add descriptive failure messages to assertions
• Use pytest.fail() with detailed error info for timeouts
• Check log_lines for compilation or runtime errors
• Enable debug logging in YAML fixtures when needed

8. Performance Considerations

• Use short update intervals (0.1s) for faster tests
• Set reasonable timeouts (5-10s for most operations)
• Batch multiple assertions when possible
• Clean up resources properly using context managers

9. Test Categories

• Basic Tests: Minimal functionality verification
• Entity Tests: Sensor, switch, light behavior
• API Tests: Message batching, services, events
• Scheduler Tests: Timing, defer operations, stress
• Memory Tests: Conditional compilation, optimization
• Integration Tests: Areas, devices, complex interactions