Fixtures & Utilities¶

Built-in pytest fixtures provided by the Zelos SDK plugins.

Core Fixtures¶

`check` - Assertion Framework¶

import zelos_sdk

motor = zelos_sdk.TraceSourceCache("motor")
motor.add_event("status", [
    zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
])


def test_example(check):
    """The check fixture is automatically provided."""
    motor.status.log(rpm=2000.0)
    check.that(motor.status.rpm, "==", 2000)
    check.that(motor.status.rpm, ">", 1900,
               temporal="within_duration", duration_s=1.0)

check is agent.check: requesting it pulls in the agent fixture below, so both spellings post to the same board and artifact.

Scope: function Auto-use: No (request explicitly)

`agent` - Agent Connection Fixture¶

def test_live_voltage(agent):
    """The agent fixture yields an `Agent` with Checker capture active."""
    voltage = agent.signal("battery/cell.voltage")
    assert agent.check.that(voltage, "<", 4.25, last=5.0).passed

def test_with_alias(agent, check):
    """`check` here is `agent.check` — sugar for shorter test bodies."""
    assert check.that(agent.signal("battery/cell.voltage"), "<", 4.25, last=5.0).passed

Wraps a session-scoped Agent in a per-test Checker context, so agent.check.that(...) results post to the test's board and artifact. The connection is lazy, with no I/O at fixture setup. Configure the target with --zelos-agent-url, ZELOS_AGENT_URL, or the http://localhost:2300 default. See Reference — Agent for the full fixture API.

Checks on literals and TraceSourceCache fields evaluate in-process, so they pass or fail without a running agent. A check on a signal, and any query, raises AgentUnavailable when no agent answers.

Scope: function (backed by a session-scoped connection) Auto-use: No (request explicitly)

`_agent_session` - Shared Agent Handle¶

The session-scoped Agent behind agent. It opens no Checker, so checks made through it do not post to a board. It also connects the SDK's trace publisher, unless --zelos-trace already did.

To stop the run early when no agent answers, call it from a session fixture in conftest.py:

# conftest.py
import pytest

@pytest.fixture(scope="session", autouse=True)
def require_agent(_agent_session):
    _agent_session.health()  # raises AgentUnavailable when no agent answers

Scope: session Auto-use: No (request explicitly)

`zelos_session` - SDK Initialization¶

Auto-enabled, so tests never request it directly. With --zelos-trace, it connects the SDK to the agent at --zelos-trace-url when the session starts. With --zelos-log alone, it enables SDK logging only. Configure it on the command line:

pytest --zelos-trace --zelos-trace-url=grpc://localhost:2300 --zelos-log --zelos-log-level=debug

Scope: session Auto-use: Yes

Trace Fixtures¶

Recording Fixtures¶

These fixtures automatically manage trace file recording at different scopes (active only when --zelos-trace-file is set and scope matches --zelos-trace-file-scope). They require --zelos-local-artifacts-dir to be configured.

One fixture per scope, selected by --zelos-trace-file-scope:

Fixture	Scope	Records
`trace_file_session`	session	The entire session
`trace_file_module`	module	One file per module
`trace_file_class`	class	One file per class
`trace_file_function`	function	One file per test (default)

pytest --zelos-trace-file --zelos-local-artifacts-dir=./artifacts --zelos-trace-file-scope=class

All four are auto-use. Each one records only when --zelos-trace-file is set and its scope matches.

File naming:

Default filename: {artifact_basename}-trace-{sanitized_nodeid}.trz
Override via pytest_zelos_trace_file_name(request) in conftest.py

See Trace Recording for the exact names.

`trace_logging` - Python Log Capture¶

import logging

# Run with: pytest --zelos-trace-logging
def test_with_logging():
    """Python logging records become trace events."""
    logging.info("Test started")
    logging.error("Something failed")

With --zelos-trace-logging, the fixture adds a TraceLoggingHandler to the root logger for the session. Each record at or above --zelos-trace-logging-level becomes a log event on the source named by --zelos-trace-logging-source-name (default logger). The fixture yields the handler, or None when the option is off.

Scope: session Auto-use: Yes (records only when --zelos-trace-logging is set)

`trace_stdout` - Console Output¶

import time
import zelos_sdk

source = zelos_sdk.TraceSource("source")

# Run with: pytest -s --zelos-trace-stdout --zelos-log
def test_debug_output():
    """Events print to console for debugging."""
    source.log("event", {"value": 123})
    time.sleep(0.2)  # the sink prints in batches

The sink prints one line per trace message, not per field value:

INFO zelos_trace_py::stdout: [source] SEGMENT_START segment_id=01a10920-67e0-7902-bee5-cb7abf78704b time_ns=1791154481120732000
INFO zelos_trace_py::stdout: [event] SCHEMA segment_id=01a10920-67e0-7902-bee5-cb7abf78704b fields=[value:Int64]
INFO zelos_trace_py::stdout: [event] BATCH segment_id=01a10920-67e0-7902-bee5-cb7abf78704b rows=1

The lines go through the SDK log, so they print only with --zelos-log. --zelos-trace-stdout-level sets the level of each line; it must be at or above --zelos-log-level.

Scope: session Auto-use: Yes (prints only when --zelos-trace-stdout is set)

Utilities¶

Check Configuration¶

The check_config marker configures the agent and check fixtures for one test. If you run pytest with --strict-markers, list check_config under markers (see Configuration).

import pytest

@pytest.mark.check_config(fail_fast=False)
def test_multiple_assertions(check):
    """Continue even if checks fail."""
    check.that(1, "==", 1)
    check.that(2, "==", 3)  # Fails but continues
    check.that(3, "==", 3)  # Still runs

fail_fast=False runs every check and fails the test at the end. zelos_sdk.pytest.checker.check_config is the same marker, so @check_config(fail_fast=False) works after from zelos_sdk.pytest.checker import check_config.

Range / tolerance with the typed vocabulary¶

The typed Check API ships 21 built-in ops. Most range and tolerance assertions are best expressed with these directly:

def test_voltage_in_range(check):
    """Express a range as two bounded checks; both post to the board."""
    voltage = 3.85
    check.that(voltage, ">=", 3.0)
    check.that(voltage, "<=", 4.25)


def test_voltage_settled(check):
    """`is_close` uses math.isclose defaults; `is_approximately`
    uses pytest.approx defaults — pick whichever matches your
    domain's expected-error model."""
    check.that(3.851, "is_close", 3.85, rel_tol=0.01)
    check.that(3.851, "is_approximately", 3.85, rel_tol=0.01)

Custom local-eval operators¶

For domain-specific predicates that don't fit the typed vocabulary, the zelos_sdk.pytest.checker.ops registry lets you register a DescriptiveOp and use it like any other op string. Custom ops are local-eval only — the callable runs in-process and the result posts to the board, but the spec never crosses the wire to the agent (so they pair with literal / cache-field operands, not Signal handles).

from zelos_sdk.pytest.checker import ops

def within_range(value, bounds):
    lo, hi = bounds
    return lo <= value <= hi

ops.register_op(ops.DescriptiveOp(within_range, "is within range"))

def test_range(check):
    check.that(3.85, "is within range", (3.0, 4.25))

Or pass a callable directly without registration (the function's __name__ becomes the board label):

def test_inline_predicate(check):
    check.that(5.0, lambda v, b: b[0] <= v <= b[1], (0, 10))

Typed ops always win the dispatch race — registering "==" as a custom op can't shadow the typed fast path. Custom ops paired with a Signal raise immediately (use the typed Op vocabulary for agent-side temporal evaluation instead). register_op(..., overwrite=True) replaces an existing entry; unregister_op("description") removes one.

Logging Integration¶

from zelos_sdk.hooks.logging import TraceLoggingHandler
import logging

# Add to any test
def test_with_logs():
    handler = TraceLoggingHandler(source_name="test_logs")
    logging.getLogger().addHandler(handler)
    try:
        logging.info("Test started")
        # All logs now appear in trace
    finally:
        logging.getLogger().removeHandler(handler)