EmulatedLifxServer¶

The EmulatedLifxServer class manages the UDP server and routes packets to emulated devices.

Overview¶

The server:

Listens on a UDP socket for LIFX protocol packets
Parses packet headers to determine routing
Forwards packets to appropriate devices
Sends response packets back to clients
Supports both targeted and broadcast packets

API Reference¶

EmulatedLifxServer ¶

EmulatedLifxServer(
    devices: list[EmulatedLifxDevice],
    device_manager: IDeviceManager,
    bind_address: str = "127.0.0.1",
    port: int = LIFX_UDP_PORT,
    track_activity: bool = True,
    storage=None,
    activity_observer: ActivityObserver | None = None,
    scenario_manager: HierarchicalScenarioManager | None = None,
    persist_scenarios: bool = False,
    scenario_storage: IScenarioStorageBackend | None = None,
    *,
    ipv6_bind_address: str = "::1",
    max_pending_packets: int = 1024,
    mdns_enabled: bool = False,
)

UDP server that simulates LIFX devices

METHOD	DESCRIPTION
`start`	Atomically bind endpoints, then apply the fleet-dependent mDNS policy.
`stop`	Drain admitted work, then close and clear every known endpoint.

Source code in packages/lifx-emulator-core/src/lifx_emulator/server.py

def __init__(
    self,
    devices: list[EmulatedLifxDevice],
    device_manager: IDeviceManager,
    bind_address: str = "127.0.0.1",
    port: int = LIFX_UDP_PORT,
    track_activity: bool = True,
    storage=None,
    activity_observer: ActivityObserver | None = None,
    scenario_manager: HierarchicalScenarioManager | None = None,
    persist_scenarios: bool = False,
    scenario_storage: IScenarioStorageBackend | None = None,
    *,
    ipv6_bind_address: str = "::1",
    max_pending_packets: int = 1024,
    mdns_enabled: bool = False,
):
    # Device manager (required dependency injection)
    self._mdns_enabled = mdns_enabled
    self._mdns: MdnsResponder | None = None
    self._mdns_status = MdnsStatus.STOPPED if mdns_enabled else MdnsStatus.DISABLED
    self._mdns_error: BaseException | None = None
    self._mdns_listener: DeviceLifecycleListener | None = None
    self._mdns_failure_task: asyncio.Task[None] | None = None
    self._device_manager = device_manager
    self.bind_address = bind_address
    self.ipv6_bind_address = ipv6_bind_address
    self.port = port
    self.transport: asyncio.DatagramTransport | None = None
    self._ipv4_transport: asyncio.DatagramTransport | None = None
    self._ipv6_transport: asyncio.DatagramTransport | None = None
    self._ipv4_protocol: EmulatedLifxServer.LifxProtocol | None = None
    self._ipv6_protocol: EmulatedLifxServer.LifxProtocol | None = None
    self._ipv4_endpoint: tuple[str, int] | None = None
    self._ipv6_endpoint: tuple[str, int] | None = None
    self._effective_port: int | None = None
    # Set while the IPv4 endpoint is bound to the wildcard on behalf of a
    # specific LAN address; peers outside its subnets are filtered.
    self.peer_filter: InterfaceSelection | None = None
    self.storage = storage
    self._max_pending_packets = max_pending_packets
    self._packet_generation = 0
    self._background_tasks = BackgroundTaskTracker(
        "server-generation-0", max_pending=max_pending_packets
    )
    self._retired_background_tasks: set[BackgroundTaskTracker] = set()
    self._lifecycle_lock = asyncio.Lock()
    self._endpoint_loss_tasks: set[asyncio.Task[None]] = set()

    # Scenario storage backend (optional - only needed for persistence)
    self.scenario_persistence: IScenarioStorageBackend | None = None
    if persist_scenarios:
        if scenario_storage is None:
            raise ValueError(
                "scenario_storage is required when persist_scenarios=True"
            )
        if scenario_manager is None:
            raise ValueError(
                "scenario_manager is required when persist_scenarios=True "
                "(must be pre-loaded from storage before server initialization)"
            )
        self.scenario_persistence = scenario_storage

    # Scenario manager (shared across all devices for runtime updates)
    self.scenario_manager = scenario_manager or HierarchicalScenarioManager()

    # Add initial devices to the device manager
    for device in devices:
        # Update device port to match server port
        device.state.port = self.port
        if not self._device_manager.add_device(device, self.scenario_manager):
            device.reject_admission()
            logger.warning(
                "Rejected initial device with duplicate serial: %s",
                device.state.serial,
            )

    # Activity observer - defaults to ActivityLogger if track_activity=True
    if activity_observer is not None:
        self.activity_observer = activity_observer
    elif track_activity:
        self.activity_observer = ActivityLogger(max_events=100)
    else:
        self.activity_observer = NullObserver()

    # Statistics tracking
    self.start_time = time.time()
    self._started_monotonic = time.monotonic()
    self.packets_received = 0
    self.packets_sent = 0
    self.packets_received_by_type: dict[int, int] = defaultdict(int)
    self.packets_sent_by_type: dict[int, int] = defaultdict(int)
    self.error_count = 0
    self.packets_dropped_overload = 0
    self.packets_filtered = 0
    self.websocket_events_dropped = 0

Functions¶

start `async` ¶

start()

Atomically bind endpoints, then apply the fleet-dependent mDNS policy.

Source code in packages/lifx-emulator-core/src/lifx_emulator/server.py

async def start(self):
    """Atomically bind endpoints, then apply the fleet-dependent mDNS policy."""
    async with self._lifecycle_lock:
        if self._mdns_enabled:
            for device in self.get_all_devices():
                resolve_address(device, self.bind_address, self.ipv6_bind_address)
            if not isinstance(self._device_manager, IDeviceLifecycleSource):
                raise TypeError(
                    "Enabled mDNS requires an IDeviceLifecycleSource manager"
                )
        already_running = self._has_complete_endpoint_pair()
        await self._start_locked()
        if not already_running or self._mdns_status != MdnsStatus.FAILED:
            try:
                await self._start_mdns_locked()
            except BaseException:
                if not already_running:
                    await self._stop_locked()
                raise

stop `async` ¶

stop()

Drain admitted work, then close and clear every known endpoint.

Source code in packages/lifx-emulator-core/src/lifx_emulator/server.py

async def stop(self):
    """Drain admitted work, then close and clear every known endpoint."""
    lifecycle_lock = getattr(self, "_lifecycle_lock", None)
    if lifecycle_lock is None:
        await self._stop_locked()
        return
    try:
        async with lifecycle_lock:
            await self._stop_locked()
    finally:
        failure = getattr(self, "_mdns_failure_task", None)
        if failure is not None and failure is not asyncio.current_task():
            await asyncio.shield(failure)

Usage¶

Basic Server¶

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

# Create devices
devices = [create_color_light("d073d5000001")]

# Set up repository and manager (required)
device_repository = DeviceRepository()
device_manager = DeviceManager(device_repository)

# Create server
server = EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700)

# Start server
await server.start()

# ... do work ...

# Stop server
await server.stop()

Context Manager¶

The recommended way to use the server:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

devices = [create_color_light("d073d5000001")]
device_manager = DeviceManager(DeviceRepository())

async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    # Server automatically starts
    # Your test code here
    pass
# Server automatically stops

Multiple Devices¶

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import (
    create_color_light,
    create_multizone_light,
    create_tile_device,
)
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

devices = [
    create_color_light("d073d5000001"),
    create_multizone_light("d073d5800001", zone_count=16),
    create_tile_device("d073d5900001", tile_count=5),
]

device_manager = DeviceManager(DeviceRepository())

async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    # All devices are discoverable and controllable
    pass

Parameters¶

`devices`¶

List of EmulatedLifxDevice instances to emulate.

Type: list[EmulatedLifxDevice]

`device_manager`¶

The device manager handling device lifecycle and packet routing.

Type: DeviceManager

Required: Yes

Notes: - The server delegates all device management operations to this manager - Must be created with a DeviceRepository instance - This is a required parameter in v2.0.0+

Example:

from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

device_repository = DeviceRepository()
device_manager = DeviceManager(device_repository)

`bind_address`¶

IPv4 address to bind the UDP server to.

Type: str

Examples: - "0.0.0.0" - Listen on all interfaces - "127.0.0.1" - Localhost only - "192.168.1.100" - Specific interface

Loopback and "0.0.0.0" bind directly. A specific interface address binds "0.0.0.0" on the port so that discovery broadcasts are received, and filters datagrams to peers on the subnets of the interface that owns the address. Filtered datagrams are counted in packets_filtered (see get_stats()), are not counted as received and do not reach the activity observer. start() raises ValueError if no local interface owns the address. Unicast from peers outside the interface's subnets is filtered; use "0.0.0.0" to accept every peer. The filter is based on the peer's source address and is an interface-selection convenience, not a security control.

The mDNS responder keeps advertising the configured bind_address.

`port`¶

UDP port to listen on.

Type: int

Default: 56700 (standard LIFX port)

`track_activity`¶

Enable packet activity tracking for the HTTP API dashboard.

Type: bool

Default: True

Notes: - When enabled, the server tracks recent packet activity for the API dashboard - Disable to reduce memory usage in production or CI environments - Activity tracking is independent of packet logging (controlled by --verbose CLI flag)

Example:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.devices import DeviceManager
from lifx_emulator.repositories import DeviceRepository

# Disable activity tracking
device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(
    devices,
    DeviceManager(DeviceRepository()),
    "127.0.0.1",
    56700,
    track_activity=False,
)

`storage`¶

Optional persistent storage backend for device state.

Type: DevicePersistenceAsyncFile | None

Default: None

Notes: - Devices save their own state through the storage passed to their factory; pass the same instance here - The server uses it to delete persisted state when devices are removed (remove_device(), remove_all_devices(delete_storage=True)) - Allows device state to persist across emulator restarts - See Persistent Storage Guide for details

Example:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.devices import DeviceManager, DevicePersistenceAsyncFile
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository

storage = DevicePersistenceAsyncFile()
device = create_color_light("d073d5000001", storage=storage)

device_manager = DeviceManager(DeviceRepository())

# Create server with storage support
server = EmulatedLifxServer(
    [device],
    device_manager,
    "127.0.0.1",
    56700,
    storage=storage
)

`activity_observer`¶

Optional observer for packet activity events.

Type: ActivityObserver | None

Default: None

Notes: - Implement ActivityObserver protocol to receive packet events - Useful for custom activity tracking or metrics collection - Receives events for all packets sent and received

`scenario_manager`¶

Optional scenario manager for test scenario configuration.

Type: HierarchicalScenarioManager | None

Default: None

Notes: - Shared by all devices; if omitted, the server creates an empty one (available as server.scenario_manager) - When provided, enables runtime scenario management via REST API - Supports device-specific, type-specific, location-based, group-based, and global scenarios - Scenarios control packet dropping, delays, malformed responses, etc. - See Testing Scenarios Guide for detailed examples

Example:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.scenarios import HierarchicalScenarioManager
from lifx_emulator.devices import DeviceManager
from lifx_emulator.repositories import DeviceRepository

manager = HierarchicalScenarioManager()
device_manager = DeviceManager(DeviceRepository())

server = EmulatedLifxServer(
    devices,
    device_manager,
    "127.0.0.1",
    56700,
    scenario_manager=manager
)

# Now scenario management API is available

`persist_scenarios`¶

Enable persistent storage of scenario configurations.

Type: bool

Default: False

Notes: - When enabled, scenario changes made through the API are saved via scenario_storage (ScenarioPersistenceAsyncFile writes ~/.lifx-emulator/scenarios.json) - Requires both scenario_storage and scenario_manager; the constructor raises ValueError if either is missing - Load scenario_manager from scenario_storage before creating the server so saved scenarios are restored

Example:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.devices import DeviceManager
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.scenarios import ScenarioPersistenceAsyncFile

# Enable both state and scenario persistence
scenario_storage = ScenarioPersistenceAsyncFile()
manager = await scenario_storage.load()

device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(
    devices,
    device_manager,
    "127.0.0.1",
    56700,
    storage=storage,
    scenario_manager=manager,
    persist_scenarios=True,
    scenario_storage=scenario_storage,
)

Methods¶

Lifecycle Methods¶

`async start()`¶

Start the UDP server and begin accepting connections.

Notes: - Call this method before sending any packets to the emulator - Binds to the configured address and port - Logs server startup information - Required if not using context manager

Example:

device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700)
await server.start()
try:
    # Use server
    pass
finally:
    await server.stop()

`async stop()`¶

Stop the UDP server and clean up resources.

Notes: - Gracefully closes the UDP endpoint - Cleans up internal state - Safe to call multiple times - Automatically called by context manager

Context Manager Protocol¶

The server implements the async context manager protocol for clean resource management:

`async aenter()`¶

Enter async context manager - automatically calls start().

`async aexit()`¶

Exit async context manager - automatically calls stop().

Recommended Usage:

device_manager = DeviceManager(DeviceRepository())
async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    # Server is automatically started
    # Perform your tests
    pass
# Server is automatically stopped

Utility Methods¶

`get_stats()`¶

Get server statistics.

Returns: dict[str, Any] - Includes uptime_seconds, start_time, device_count, packets_received, packets_sent, packets_received_by_type, packets_sent_by_type, error_count, packets_dropped_overload, websocket_events_dropped and activity_enabled

Notes: - uptime_seconds is measured with a monotonic clock from server construction - Useful for performance testing and benchmarking

Example:

import asyncio

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.devices import DeviceManager
from lifx_emulator.repositories import DeviceRepository

device_manager = DeviceManager(DeviceRepository())
async with EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700) as server:
    await asyncio.sleep(1)
    stats = server.get_stats()
    print(f"Server uptime: {stats['uptime_seconds']:.2f}s")
    print(f"Packets received: {stats['packets_received']}")

`invalidate_all_scenario_caches()`¶

Clear every device's cached resolved scenario.

Notes: - Normally called automatically after scenario updates via API - Only needed if modifying scenario manager state outside of API - Safe to call - no side effects - Non-blocking operation

Example:

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.devices import DeviceManager
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.scenarios import HierarchicalScenarioManager, ScenarioConfig

manager = HierarchicalScenarioManager()
device_manager = DeviceManager(DeviceRepository())
server = EmulatedLifxServer(
    devices, device_manager, "127.0.0.1", 56700, scenario_manager=manager
)

async with server:
    # Update scenarios externally
    manager.set_global_scenario(ScenarioConfig(drop_packets={101: 0.5}))
    # Invalidate cache to apply changes immediately
    server.invalidate_all_scenario_caches()

Packet Routing¶

Broadcast Packets¶

Packets with tagged=True or target=000000000000 are forwarded to all devices:

# GetService broadcasts are answered by all devices
# Client discovers all emulated devices

Targeted Packets¶

Packets with a specific target serial are routed to that device:

# Light.SetColor for d073d5000001 goes only to that device
# Other devices don't see the packet

Unknown Targets¶

Packets for unknown serial addresses are silently dropped:

# Packet for d073d5999999 (not in server) is ignored
# No error or response generated

Response Handling¶

The server handles responses automatically:

Device processes packet and returns response(s)
Server packs response packets to bytes
Server sends responses back to source address
Multiple responses (e.g., multizone StateMultiZone) are sent sequentially

Concurrency¶

The server uses asyncio for concurrent operation:

# Multiple clients can send packets concurrently
# Each device processes packets independently
# Responses are sent asynchronously

Error Handling¶

The server handles errors gracefully:

Invalid packets are logged and dropped
Device exceptions are caught and logged
Server continues running despite errors
Malformed headers don't crash the server

Lifecycle¶

Startup¶

Create UDP endpoint
Bind to address and port
Start receiving packets
Log server start

Runtime¶

Receive packet bytes
Parse header
Route to device(s)
Get responses
Send responses

Shutdown¶

Stop accepting packets
Close UDP endpoint
Clean up resources
Log server stop

Testing Integration¶

pytest-asyncio¶

import pytest
from lifx_emulator import EmulatedLifxServer
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository
from lifx_emulator.devices import DeviceManager

@pytest.fixture
def device_manager():
    return DeviceManager(DeviceRepository())

@pytest.fixture
async def lifx_server(device_manager):
    device = create_color_light("d073d5000001")
    server = EmulatedLifxServer([device], device_manager, "127.0.0.1", 56700)

    async with server:
        yield server

@pytest.mark.asyncio
async def test_discovery(lifx_server):
    # Test code using the server
    pass

Module-Scoped Fixture¶

For faster tests, use module scope:

import pytest

from lifx_emulator import EmulatedLifxServer
from lifx_emulator.devices import DeviceManager
from lifx_emulator.factories import create_color_light
from lifx_emulator.repositories import DeviceRepository

@pytest.fixture(scope="module")
def device_manager():
    return DeviceManager(DeviceRepository())

@pytest.fixture(scope="module")
async def lifx_server(device_manager):
    devices = [create_color_light(f"d073d500000{i}") for i in range(5)]
    server = EmulatedLifxServer(devices, device_manager, "127.0.0.1", 56700)

    async with server:
        yield server

Next Steps¶

Device API - EmulatedLifxDevice documentation
Factory Functions - Device creation
Integration Testing Tutorial - Integration test examples

EmulatedLifxServer¶

Overview¶

API Reference¶

EmulatedLifxServer ¶

Functions¶

start async ¶

stop async ¶

Usage¶

Basic Server¶

Context Manager¶

Multiple Devices¶

Parameters¶

devices¶

device_manager¶

bind_address¶

port¶

track_activity¶

storage¶

activity_observer¶

scenario_manager¶

persist_scenarios¶

Methods¶

Lifecycle Methods¶

async start()¶

async stop()¶

Context Manager Protocol¶

async __aenter__()¶

async __aexit__()¶

Utility Methods¶

get_stats()¶

invalidate_all_scenario_caches()¶

Packet Routing¶

Broadcast Packets¶

Targeted Packets¶

Unknown Targets¶

Response Handling¶

Concurrency¶

Error Handling¶

Lifecycle¶

Startup¶

Runtime¶

Shutdown¶

Testing Integration¶

pytest-asyncio¶

Module-Scoped Fixture¶

Next Steps¶

start `async` ¶

stop `async` ¶

`devices`¶

`device_manager`¶

`bind_address`¶

`port`¶

`track_activity`¶

`storage`¶

`activity_observer`¶

`scenario_manager`¶

`persist_scenarios`¶

`async start()`¶

`async stop()`¶

`async aenter()`¶

`async aexit()`¶

`get_stats()`¶

`invalidate_all_scenario_caches()`¶