Checker Assertions¶

The Checker plugin provides a rich assertion framework with 30+ operators, time-based validations, and visual output for hardware and systems testing.

Agent-mediated checks (via Signals)¶

For data already flowing into an agent or recorded in a .trz trace, use agent.check / trace.check instead of the pytest fixture below. The predicate evaluates server-side against the live store or trace file — same operator vocabulary and checkerboard output as the fixture, no in-process cache plumbing required.

from zelos_sdk.agent import Agent

agent = Agent().connect()
voltage = agent.signals().by_path("device/pack.voltage")

result = agent.check.that(voltage, "<=", 4.2)
assert result.passed

Run a JSON suite (same lhs / op / rhs / temporal shape as the underlying CheckSpec) in one round-trip:

results = agent.check.suite("checks/live.json")
assert all(r.passed for r in results)

Trace mirrors live — open a .trz and call the same methods:

trace = agent.trace("/path/to/run.trz")
trace.check.suite("checks/trace.json", last=2.0)

CheckResult exposes .passed, .status (pass / fail / error), .fired_at_ns (wall-clock eval time), and .evidence (the satisfying / violating sample for signal-bearing checks). The same flow runs from the CLI via zelos live check / zelos trace check and renders the same checkerboard.

Cold-start references with `agent.signal()`¶

agent.signals().by_path(...) requires the agent's catalog to already know the path — it raises SignalNotFound if the producer hasn't emitted yet. For TDD-style tests where the producer is set up in the same test (or extension consumers that want to declare interest in a signal before it starts emitting), use agent.signal(path) instead. It constructs a typed Signal handle from a bare path string with no catalog roundtrip; the agent resolves the path server-side at the next check / latest / query call:

result = agent.check.that(
    agent.signal("device/pack.voltage"),
    "<=", 4.2,
    temporal="within_duration", duration_s=1.0,  # absorbs ingest latency
)

The local handle's data_type defaults to Float64 and is informational only — the wire encoding carries just the path string, and the agent's resolved signal carries the authoritative type.

Multi-segment resolution: `strict=`¶

When the same logical signal (source/message.signal) is recorded by multiple segments — most commonly because an extension restarted, a remote agent reconnected, or a test fixture spun up a fresh TraceSource — the resolver picks the segment with the most recent data by default. This matches the customer mental model: the freshest active producer wins, no manual disambiguation needed.

Pass strict=True to opt out — any multi-segment match then errors Reason::AmbiguousSignal, useful when a test wants to assert a unique-segment invariant:

agent.check.that(agent.signal("device/pack.voltage"), "<=", 4.2, strict=True)

Available on agent.check.that / .count / .suite (and the trace mirrors), and on the CLI as --strict. Cross-producer ambiguity (the same path emitted by different producers) is always an error regardless of strict — narrow with producers= to disambiguate.

Installation & Setup¶

# conftest.py
pytest_plugins = ["zelos_sdk.pytest.checker"]

Enable console output to see the checkerboard:

pytest --log-cli --log-cli-level=INFO

At a Glance¶

# Core API
check.that(
    lhs,                       # Signal, literal, or TraceSourceCacheLastField
    op="is_true",              # operator: canonical string or alias
    rhs=None,                  # comparison value (omit for unary ops)
    *,
    temporal=None,             # "latest" / "always" / "ever" / "never" /
                               #   "count" / "for_duration" / "within_duration"
    duration_s=None,            # required for temporal="for_duration" / "within_duration"
    interval_s=0.1,            # polling cadence for duration temporals
    nonblocking=False,         # background thread for duration checks
    rel_tol=None, abs_tol=None, nan_ok=False,   # tolerance ops
)

The checker fails the test immediately on failure by default (fail_fast=True).
To evaluate all checks and fail at the end, use @check_config(fail_fast=False).
With nonblocking=True, the duration check runs in a background thread; fixture teardown joins it before pytest reports — no manual while item.running(): ... needed.
temporal="for_duration" / temporal="within_duration" require duration_s; the two temporals are mutually exclusive.
Tolerance ops (is_close, is_approximately, ~=) take rel_tol / abs_tol / nan_ok as top-level kwargs.

Tip: Define TraceSourceCacheLast in fixtures (not in tests). Keep tests declarative and focused on checks.

Complete Operator Reference¶

Equality Operators¶

Operator	Aliases	Description	Example
`==`	`=`, `is`, `is equal to`	Exact equality	`check.that(value, "==", 5.0)`
`!=`	`is not`, `is not equal to`	Inequality	`check.that(state, "!=", "ERROR")`

def test_equality(check):
    # All these are equivalent
    check.that(sensor_value, "==", 5.0)
    check.that(sensor_value, "=", 5.0)
    check.that(sensor_value, "is", 5.0)
    check.that(sensor_value, "is equal to", 5.0)

    # Inequality
    check.that(status, "!=", "FAILED")
    check.that(status, "is not", "FAILED")

Comparison Operators¶

Operator	Aliases	Description	Example
`>`	`is greater than`	Greater than	`check.that(rpm, ">", 1000)`
`>=`	`is greater than or equal to`	Greater or equal	`check.that(voltage, ">=", 4.5)`
`<`	`is less than`	Less than	`check.that(temp, "<", 80)`
`<=`	`is less than or equal to`	Less or equal	`check.that(current, "<=", 10)`

def test_comparisons(check):
    check.that(motor_rpm, ">", 1000)
    check.that(motor_rpm, "is greater than", 1000)

    check.that(battery_voltage, ">=", 11.5)
    check.that(temperature, "<", 85.0)
    check.that(current_draw, "<=", 50.0)

Membership Operators¶

Operator	Aliases	Description	Example
`in`	`is in`	Element in collection	`check.that(2, "in", [1,2,3])`
`not in`	`is not in`	Element not in collection	`check.that(4, "not in", [1,2,3])`
`contains`	-	Collection contains element	`check.that([1,2,3], "contains", 2)`

def test_membership(check):
    valid_states = ["IDLE", "RUNNING", "PAUSED"]

    check.that(current_state, "in", valid_states)
    check.that(current_state, "is in", valid_states)

    check.that("ERROR", "not in", valid_states)
    check.that(valid_states, "contains", "IDLE")

    # Works with strings too
    check.that("H", "in", "Hello")
    check.that("Hello World", "contains", "World")

Approximation Operators¶

Operator	Aliases	Description	Example
`is_close`	`is_close_to`, `is_around`	Math.isclose comparison	`check.that(pi, "is_close_to", 3.14159)`
`~=`	`is_approximately`, `is_approximately_equal_to`	Pytest.approx comparison	`check.that(value, "~=", expected)`

def test_approximations(check):
    measured = 3.14159

    # Using math.isclose (with default tolerances)
    check.that(measured, "is_close_to", 3.14)
    check.that(measured, "is_around", 3.14)

    # With custom tolerances (top-level kwargs)
    check.that(measured, "is_close_to", 3.14,
               rel_tol=0.01, abs_tol=0.01)

    # Using pytest.approx
    check.that(measured, "~=", 3.14)
    check.that(measured, "is_approximately", 3.14)

    # With pytest.approx parameters
    check.that(measured, "~=", 3.14,
               rel_tol=0.01, abs_tol=0.01)

String Operators¶

Operator	Description	Example
`starts with`	String starts with substring	`check.that(msg, "starts with", "OK")`
`ends with`	String ends with substring	`check.that(file, "ends with", ".txt")`
`contains`	String contains substring	`check.that(log, "contains", "ERROR")`

def test_strings(check):
    message = "OK: System initialized"
    filename = "data_20240115.csv"

    check.that(message, "starts with", "OK")
    check.that(message, "contains", "System")
    check.that(filename, "ends with", ".csv")

    # Case sensitive
    check.that("Hello World", "starts with", "Hello")  # Pass
    check.that("Hello World", "starts with", "hello")  # Fail

Collection Operators¶

Operator	Description	Example
`is_empty`	String is empty (`len(x) == 0`)	`check.that("", "is_empty")`

def test_collections(check):
    message = ""

    check.that(message, "is_empty")

Numeric Operators¶

Operator	Description	Example
`is_positive`	Number > 0	`check.that(value, "is_positive")`
`is_not_positive`	Number <= 0	`check.that(value, "is_not_positive")`
`is_negative`	Number < 0	`check.that(value, "is_negative")`
`is_not_negative`	Number >= 0	`check.that(value, "is_not_negative")`
`is_divisible_by`	Modulo == 0 (integer operands)	`check.that(10, "is_divisible_by", 5)`

def test_numeric(check):
    profit = 1500.50
    loss = -250.75
    zero = 0.0

    check.that(profit, "is_positive")
    check.that(loss, "is_negative")
    check.that(zero, "is_not_positive")
    check.that(zero, "is_not_negative")

    # Divisibility
    check.that(100, "is_divisible_by", 10)
    check.that(15, "is_divisible_by", 3)

Boolean Operators¶

Operator	Aliases	Description	Example
`is_true`	`is_True`	Boolean operand is `True`	`check.that(flag, "is_true")`
`is_false`	`is_False`	Boolean operand is `False`	`check.that(flag, "is_false")`

def test_boolean(check):
    enabled = True
    disabled = False

    check.that(enabled, "is_true")
    check.that(disabled, "is_false")

Time-Based Checks¶

Within Duration¶

Ensure a condition becomes true within a specified time:

def test_within_duration(check):
    source = zelos_sdk.TraceSourceCacheLast("test")

    # Start operation
    system.start_async_operation()

    # Verify completion within 5 seconds
    check.that(
        source.status.complete,
        "is_true",
        temporal="within_duration",
        duration_s=5.0,
        interval_s=0.1,  # Poll every 100ms
    )

    # Non-blocking version — runs in a background thread; fixture
    # teardown joins it before pytest reports.
    check.that(
        source.signal.value, ">", 100,
        temporal="within_duration",
        duration_s=3.0,
        nonblocking=True,
    )

    # Do other work while the background check polls
    perform_other_tests()

For Duration¶

Ensure a condition remains true for a specified duration:

def test_for_duration(check):
    source = zelos_sdk.TraceSourceCacheLast("test")

    # Verify stability
    check.that(
        source.voltage.value,
        "is_close_to", 5.0,
        temporal="for_duration",
        duration_s=10.0,    # Must stay close for 10 seconds
        interval_s=0.5,     # Poll every 500ms
        abs_tol=0.1,        # Within ±0.1V
    )

    # Non-blocking version
    check.that(
        source.vibration.amplitude, "<", 0.5,
        temporal="for_duration",
        duration_s=5.0,
        nonblocking=True,
    )

    # Continue with other operations — fixture teardown waits for
    # the background loop before emitting the board / artifact.
    run_test_sequence()

Non‑Blocking + Fail‑Fast¶

With nonblocking=True, the duration check runs in a background thread. The pytest fixture tracks the thread and joins it during teardown, so failures still flip the test result before pytest reports — no manual join required.
To aggregate failures across all checks (instead of raising on the first), use @check_config(fail_fast=False).
If you use nonblocking=True with fail_fast=True, a failure may occur in the background while the test body continues. Fixture teardown surfaces the failure cleanly via pytest's delayed-fail hook.

Using with TraceSourceCacheLast (fixtures pattern)¶

Recommended: define sources and events in fixtures, then consume fields in tests.

# conftest.py
import pytest
import zelos_sdk

@pytest.fixture(scope="module")
def motor():
    source = zelos_sdk.TraceSourceCacheLast("motor")
    source.add_event("status", [
        zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
        zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),
        zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "°C"),
    ])

    yield source

# test_motor.py
def test_trace_integration(motor, check):
    # Direct field checks (checker calls .get() on fields)
    check.that(motor.status.rpm, "==", 2500)
    check.that(motor.status.torque, ">", 30)
    check.that(motor.status.temperature, "<", 80)

    # The checker automatically:
    # 1) Calls .get() on TraceSourceCacheLastField objects
    # 2) Extracts the field name for display
    # 3) Shows both name and value in output

Tolerance Parameters¶

is_close_to / is_close (math.isclose): rel_tol (default 1e-9), abs_tol (default 0).
~= / is_approximately (pytest.approx): rel_tol (default 1e-6), abs_tol (default 1e-12), nan_ok (default False).

Pass any subset as top-level kwargs on check.that; the rest stay at the stdlib defaults.

Configuration¶

Test-Level Configuration¶

from zelos_sdk.pytest.checker import check_config

@check_config(fail_fast=False)
def test_multiple_checks(check):
    """All checks run even if some fail."""
    check.that(1, "==", 1)  # Pass
    check.that(2, "==", 3)  # Fail - but test continues
    check.that(3, "==", 3)  # Pass - still executes
    # Test fails at the end if any check failed

Visual Output¶

Understanding the Checkerboard¶

                     Zelos Checkerboard
                    test_comprehensive_example
┏━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━┳━━━━━━━━┓
┃ timestamp            ┃ lhs              ┃ operator     ┃ rhs            ┃ parameters     ┃ result ┃
┡━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━╇━━━━━━━━┩
│ 2024-01-15T14:30:22  │ voltage (5.01)   │ ~=           │ 5.0            │ rel=0.01       │ PASSED │
│ 2024-01-15T14:30:22  │ state (2)        │ in           │ [0, 1, 2, 3]   │                │ PASSED │
│ 2024-01-15T14:30:23  │ rpm (2500)       │ >            │ 2000           │                │ PASSED │
│ 2024-01-15T14:30:25  │ temp (75.5)      │ <            │ 80             │ abs_tol=0.05   │ PASSED │
│ 2024-01-15T14:30:25  │ message          │ starts with  │ OK             │                │ PASSED │
└──────────────────────┴──────────────────┴──────────────┴────────────────┴────────────────┴────────┘

Column descriptions:

timestamp: When the check was evaluated
lhs: Left-hand side (shows field name and value for trace fields)
operator: The operator used
rhs: Right-hand side (comparison value)
parameters: Additional parameters (duration, kwargs, etc.)
result: PASSED (green) or FAILED (red)

Styling¶

Alternating rows are dimmed for readability
Pass/fail results are color-coded
Field names from TraceSourceCacheLast are shown with values

API Reference¶

check.that(lhs, op="is_true", rhs=None, *, temporal=None, duration_s=None, interval_s=0.1, nonblocking=False, rel_tol=None, abs_tol=None, nan_ok=False, ...) -> CheckResult | dict[str, CheckResult] | None
- Returns a typed :class:CheckResult (or {address: CheckResult} for multi-producer fan-out). With nonblocking=True the call returns None and the background result posts to the board on completion.
- temporal="for_duration" / temporal="within_duration" require duration_s and are mutually exclusive.
Trace integration
- When lhs/rhs are TraceSourceCacheLastField, the SDK calls .get() at spec-build time and sends the captured value on the wire as a literal — the agent stores the check with the source path preserved.

Common Patterns¶

Hardware Validation Pattern (with fixtures)¶

def test_hardware_sequence(hw, check):
    """Complete hardware test pattern using the hw fixture."""
    # Power on sequence (performed by the fixture or here)
    hw.power_on()

    # Verify power good signal
    check.that(hw.status.power_good, "is_true",
               temporal="within_duration", duration_s=0.5)

    # Verify voltage stabilizes
    check.that(hw.status.voltage, "~=", 5.0,
               temporal="within_duration", duration_s=1.0,
               abs_tol=0.1)

    # Verify stays stable
    check.that(hw.status.voltage, "is_close_to", 5.0,
               temporal="for_duration", duration_s=2.0,
               abs_tol=0.05)

    # Verify ready signal
    check.that(hw.status.ready, "is_true",
               temporal="within_duration", duration_s=3.0)

State Machine Pattern (with fixtures)¶

def test_state_transitions(fsm, check):
    """Verify state machine transitions using the fsm fixture."""
    # Initial state
    check.that(fsm.state.current, "==", 0)

    # Trigger transition
    fsm.initialize()

    # Should transition to IDLE
    check.that(fsm.state.current, "==", 1,
               temporal="within_duration", duration_s=2.0)
    check.that(fsm.state.previous, "==", 0)

    # Start operation
    fsm.start()

    # Should transition to ACTIVE
    check.that(fsm.state.current, "==", 2,
               temporal="within_duration", duration_s=1.0)
    check.that(fsm.state.previous, "==", 1)

Troubleshooting¶

Issue: Checker Output Not Visible¶

# Solution: Enable logging
pytest --log-cli --log-cli-level=INFO

# Or in pytest.ini
[pytest]
log_cli = true
log_cli_level = INFO

Issue: Check Fails Immediately¶

# Problem: Default fail_fast=True stops on first failure
def test_problem(check):
    check.that(1, "==", 2)  # Fails and stops here
    check.that(2, "==", 2)  # Never executed

# Solution: Use fail_fast=False
@check_config(fail_fast=False)
def test_solution(check):
    check.that(1, "==", 2)  # Fails but continues
    check.that(2, "==", 2)  # Still executes