MockECN

Part of Offline testing in the Python API reference.

Related guidance: getting-started/mock-setup.

Runnable examples: preflight.py.

from picogrid_ecn_client.testing import MockECN

Run a deterministic MQTT v5 test broker.

class MockECN: ...

Source · testing/mock_ecn.py

Use MockECN with async with to manage its asynchronous lifecycle.

__init__

Create a stopped offline mock.

def __init__(
    *,
    host: str = '127.0.0.1',
    mqtt_port: int = 0,
    tokens: Mapping[str, Iterable[str]] | None = None,
    allow_external_bind: bool = False,
    allow_unauthenticated: bool = False,
    maximum_packet_size: int = 1024 * 1024,
) -> None

Parameters

Name	Type	Default	Description
host	str	‘127.0.0.1’	Hostname or IP literal on which to listen; non-loopback values require the development/CI-only allow_external_bind opt-in.
mqtt_port	int	0	Listener port, or 0 to select an available port.
tokens	Mapping[str, Iterable[str]] | None	None	Synthetic token-to-mock-ACL mappings, or built-in test mappings.
allow_external_bind	bool	False	Whether this development/CI-only mock may bind a non-loopback address; defaults to false.
allow_unauthenticated	bool	False	Whether this development/CI-only mock accepts credential-free MQTT CONNECT packets; defaults to false.
maximum_packet_size	int	1024 * 1024	MQTT packet-size bound in bytes.

Raises

Exception	Description
ValueError	If the endpoint, packet bound, token, or mock ACL label is invalid.

Source · testing/mock_ecn.py

events

Synchronization events for this mock instance.

events: MockEvents

Source · testing/mock_ecn.py

scenario

Mutable deterministic controls for this mock instance.

scenario: MockScenario

Source · testing/mock_ecn.py

host

Return the read-only validated loopback host the listener is bound to.

host: str

Source · testing/mock_ecn.py

mqtt_port

Return the bound MQTT listener port.

mqtt_port: int

Source · testing/mock_ecn.py

is_running

Return whether the MQTT listener is running.

is_running: bool

Source · testing/mock_ecn.py

active_connection_count

Return the number of currently accepted MQTT client connections.

active_connection_count: int

Source · testing/mock_ecn.py

active_task_count

Return the number of live asynchronous mock broker tasks.

active_task_count: int

Source · testing/mock_ecn.py

entity_state

Return a defensive snapshot of entities observed by the mock.

entity_state: Mapping[str, Mapping[str, Any]]

Source · testing/mock_ecn.py

location_state

Return a defensive snapshot of locations observed by the mock.

location_state: Mapping[str, Mapping[str, Any]]

Source · testing/mock_ecn.py

start

Start the MQTT listener and wait until it accepts connections.

async def start() -> None

Raises

Exception	Description
OSError	If the configured port is occupied or the operating system otherwise refuses the loopback or explicitly allowed external bind.

Source · testing/mock_ecn.py

close

Close the listener and all accepted clients idempotently.

async def close() -> None

Source · testing/mock_ecn.py

client_config

Build a development/CI-only client configuration for this mock.

def client_config(
    integration_name: str,
    token: str = FULL_ACCESS_TOKEN,
    *,
    container_network: str | None = None,
) -> ECNConfig

Parameters

Name	Type	Default	Description
integration_name	str	—	Integration identity for the test client.
token	str	FULL_ACCESS_TOKEN	Synthetic token understood only by a loopback mock.
container_network	str | None	None	Caller-supplied reviewed container-network name, required for a non-loopback unauthenticated mock.

Returns

Type	Description
ECNConfig	A client configuration pinned to the running mock listener.

Raises

Exception	Description
RuntimeError	If the mock has not been started.
ValueError	If an external listener does not allow unauthenticated clients or lacks an explicit container-network attestation.

Source · testing/mock_ecn.py

set_delay

Configure a deterministic publication-handling delay.

def set_delay(seconds: float, *, operation: str = 'all') -> None

Parameters

Name	Type	Default	Description
seconds	float	—	Finite non-negative delay in seconds; 0 removes it.
operation	str	‘all’	Mock operation to delay: all or mqtt.publish.

Raises

Exception	Description
ValueError	If the delay or operation is invalid.

Source · testing/mock_ecn.py

drop_next_messages

Configure upcoming publications to be suppressed.

def drop_next_messages(count: int = 1) -> None

Parameters

Name	Type	Default	Description
count	int	1	Non-negative number of upcoming publications to suppress.

Raises

Exception	Description
ValueError	If count is negative.

Source · testing/mock_ecn.py

malform_next_messages

Configure upcoming publications to carry malformed payloads.

def malform_next_messages(count: int = 1) -> None

Parameters

Name	Type	Default	Description
count	int	1	Non-negative number of upcoming publications to malform.

Raises

Exception	Description
ValueError	If count is negative.

Source · testing/mock_ecn.py

set_authorization_failure

Toggle a synthetic mock-only ACL denial.

def set_authorization_failure(operation: str, *, enabled: bool = True) -> None

Parameters

Name	Type	Default	Description
operation	str	—	Mock-only ACL operation label: entity.read, entity.write, location.read, location.write, task.receive, or task.send.
enabled	bool	True	Whether the operation is forced to deny.

Raises

Exception	Description
ValueError	If operation is not a supported mock label.

Source · testing/mock_ecn.py

drop_next_connections

Close upcoming valid MQTT CONNECT attempts without sending CONNACK.

def drop_next_connections(count: int = 1) -> None

Parameters

Name	Type	Default	Description
count	int	1	Number of upcoming connection attempts to drop.

Raises

Exception	Description
ValueError	If count is outside the supported fault bound.

Source · testing/mock_ecn.py

reject_next_authentications

Return a negative CONNACK for upcoming otherwise-valid credentials.

def reject_next_authentications(count: int = 1) -> None

Parameters

Name	Type	Default	Description
count	int	1	Number of upcoming authentication attempts to reject.

Raises

Exception	Description
ValueError	If count is outside the supported fault bound.

Source · testing/mock_ecn.py

deny_next_acknowledgements

Return operation-specific authorization failures for upcoming ACKs.

def deny_next_acknowledgements(acknowledgement: str, count: int = 1) -> None

Parameters

Name	Type	Default	Description
acknowledgement	str	—	MQTT acknowledgement family to deny: suback, puback, or unsuback.
count	int	1	Number of upcoming acknowledgements to deny.

Raises

Exception	Description
ValueError	If the acknowledgement family or count is unsupported.

Source · testing/mock_ecn.py

return_next_acknowledgement_reason

Return one raw failure-reason byte in upcoming MQTT acknowledgements.

def return_next_acknowledgement_reason(
    acknowledgement: str,
    reason_code: int,
    count: int = 1,
) -> None

Parameters

Name	Type	Default	Description
acknowledgement	str	—	MQTT acknowledgement family to alter: suback, puback, or unsuback.
reason_code	int	—	MQTT v5 failure-reason byte to return.
count	int	1	Number of upcoming acknowledgements to alter.

Raises

Exception	Description
ValueError	If the family, reason code, or count is unsupported.

Source · testing/mock_ecn.py

malform_next_protocol_responses

Send an invalid MQTT v5 property length in upcoming responses.

def malform_next_protocol_responses(response: str, count: int = 1) -> None

Parameters

Name	Type	Default	Description
response	str	—	MQTT response family to malform: connack, suback, puback, or unsuback.
count	int	1	Number of upcoming responses to malform.

Raises

Exception	Description
ValueError	If the response family or count is unsupported.

Source · testing/mock_ecn.py

disconnect_next_publications

Disconnect upcoming publications at one explicit QoS boundary.

def disconnect_next_publications(phase: str, count: int = 1) -> None

Parameters

Name	Type	Default	Description
phase	str	—	Publication-delivery phase at which to disconnect: qos0_before_completion, before_puback, or after_puback.
count	int	1	Number of upcoming publications to disconnect.

Raises

Exception	Description
ValueError	If the phase or count is unsupported.

Source · testing/mock_ecn.py

disconnect_before_publish

Disconnect live clients before the caller begins a publication.

async def disconnect_before_publish() -> None

Source · testing/mock_ecn.py

reset_scenario

Reset fault controls and synchronization events in place.

def reset_scenario() -> None

Source · testing/mock_ecn.py

disconnect_clients

Disconnect all live clients while leaving the listener available.

async def disconnect_clients() -> None

Source · testing/mock_ecn.py