Checker Assertions¶

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

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,                     # value or TraceSourceCacheLastField
    op="is true",            # operator: string or ops.DescriptiveOp
    rhs=None,                # comparison value (omit for unary ops)
    *,
    within_duration_s=None,  # eventually becomes true within window
    for_duration_s=None,     # remains true for the full window
    interval_s=0.1,          # polling interval for time-based checks
    blocking=True,           # background when False (use .running())
    args=(), kwargs={},      # extra operator params
)

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).
When using non-blocking checks (blocking=False), wait with while item.running(): ....
Do not set both within_duration_s and for_duration_s at the same time (invalid).
Operator kwargs differ by operator: is close to uses rel_tol/abs_tol; ~= uses rel/abs/nan_ok.

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 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
    check.that(measured, "is close to", 3.14,
               kwargs={"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,
               kwargs={"rel": 0.01, "abs": 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`	Collection/string is empty	`check.that([], "is empty")`
`has length`	Collection/string has specific length	`check.that([1,2,3], "has length", 3)`

def test_collections(check):
    empty_list = []
    data = [1, 2, 3, 4, 5]
    message = "Hello"

    check.that(empty_list, "is empty")
    check.that("", "is empty")
    check.that({}, "is empty")

    check.that(data, "has length", 5)
    check.that(message, "has length", 5)
    check.that({"a": 1, "b": 2}, "has length", 2)

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	`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`	Strictly True (not truthy)	`check.that(flag, "is true")`
`is false`	`is False`	Strictly False (not falsy)	`check.that(flag, "is false")`

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

    # Strict boolean checks (not truthy/falsy)
    check.that(enabled, "is true")
    check.that(enabled, "is True")
    check.that(disabled, "is false")
    check.that(disabled, "is False")

    # These will FAIL (strict boolean check)
    # check.that(1, "is true")  # Fail: 1 is truthy but not True
    # check.that(0, "is false")  # Fail: 0 is falsy but not False

Type Operators¶

Operator	Description	Example
`is instance of`	Type checking	`check.that(obj, "is instance of", str)`
`has attribute`	Attribute exists	`check.that(obj, "has attribute", "name")`

def test_types(check):
    value = 42
    text = "Hello"

    class Device:
        def __init__(self):
            self.serial = "ABC123"
            self.status = "OK"

    device = Device()

    check.that(value, "is instance of", int)
    check.that(text, "is instance of", str)
    check.that(device, "is instance of", Device)

    check.that(device, "has attribute", "serial")
    check.that(device, "has attribute", "status")

Time-Based Checks¶

Within Duration¶

Ensure a condition becomes true within a specified time:

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

    # Start operation
    system.start_async_operation()

    # Verify completion within 5 seconds
    check.that(
        source.status.complete,
        "is true",
        within_duration_s=5.0,
        interval_s=0.1  # Check every 100ms
    )

    # Non-blocking version
    check_item = check.that(
        source.signal.value, ">", 100,
        within_duration_s=3.0,
        blocking=False  # Continue immediately
    )

    # Do other work
    perform_other_tests()

    # Optionally wait until done
    while check_item.running():
        time.sleep(0.1)

For Duration¶

Ensure a condition remains true for a specified duration:

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

    # Verify stability
    check.that(
        source.voltage.value,
        "is close to", 5.0,
        for_duration_s=10.0,  # Must stay close for 10 seconds
        interval_s=0.5,       # Check every 500ms
        kwargs={"abs_tol": 0.1}  # Within ±0.1V
    )

    # Non-blocking version
    stability_check = check.that(
        source.vibration.amplitude, "<", 0.5,
        for_duration_s=5.0,
        blocking=False
    )

    # Continue with other operations
    run_test_sequence()

    # Optionally wait until done
    while stability_check.running():
        time.sleep(0.1)

Non‑Blocking + Fail‑Fast¶

With blocking=False, checks run in a background thread. To surface failures deterministically on the main thread, either:
- Use @check_config(fail_fast=False) so failures are aggregated and raised at the end of the test, or
- Join each item with while item.running(): ... before the test exits.
- If you use blocking=False with fail_fast=True, a failure may occur in a background thread while the test continues; joining or disabling fail‑fast for the test ensures a clean failure in the main thread.

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

Custom Operators¶

Creating Custom Operators¶

from zelos_sdk.pytest.checker import ops

# Simple custom operator
def within_tolerance(measured, expected, tolerance_percent):
    """Check if value is within percentage tolerance."""
    tolerance = expected * (tolerance_percent / 100.0)
    return abs(measured - expected) <= tolerance

# Register it
ops.register_op(ops.DescriptiveOp(
    within_tolerance,
    "is within % tolerance of"
))

# Use in tests
def test_voltage_regulation(check):
    measured = 5.2
    check.that(
        measured,
        "is within % tolerance of", 5.0,
        args=(5,)  # 5% tolerance
    )

Complex Custom Operators¶

import numpy as np
from zelos_sdk.pytest.checker import ops

def signal_stable(signal_values, max_deviation):
    """Check if signal is stable (low standard deviation)."""
    if len(signal_values) < 2:
        return False
    return np.std(signal_values) <= max_deviation

ops.register_op(ops.DescriptiveOp(
    signal_stable,
    "has stable signal with max deviation"
))

def test_signal_stability(check):
    readings = [5.0, 5.1, 4.9, 5.0, 5.05, 4.95]
    check.that(
        readings,
        "has stable signal with max deviation", 0.1
    )

Unary Operators¶

from zelos_sdk.pytest.checker import ops
import math

def is_valid_temperature(temp):
    """Check if temperature is in valid range."""
    return -40 <= temp <= 125  # Industrial temp range

ops.register_op(ops.DescriptiveOp(
    is_valid_temperature,
    "is valid temperature"
))

def test_temperature_sensor(check):
    reading = 25.5
    check.that(reading, "is valid temperature")  # No RHS needed

Operator Parameters¶

is close to (math.isclose): pass tolerances via kwargs={"rel_tol": ..., "abs_tol": ...}.
~= (pytest.approx): pass tolerances via kwargs={"rel": ..., "abs": ..., "nan_ok": ...}.
For custom operators, pass extra operands as args=(...) in addition to lhs and rhs.

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, *, within_duration_s=None, for_duration_s=None, interval_s=0.1, blocking=True, args=(), kwargs={}) -> CheckerItem
- Returns an item. For non‑blocking checks, use item.running() to know when it completes; item.result is set once finished.
- Do not set both within_duration_s and for_duration_s.
Operators (ops)
- ops.register_op(ops.DescriptiveOp(callable, "description"))
- ops.get_op("description")
- ops.list_ops() → list of available operator strings
- Unary ops (single parameter) accept only lhs.
Trace integration
- When lhs/rhs are TraceSourceCacheLastField, the checker calls .get() and displays the field path and current value automatically.

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", within_duration_s=0.5)

    # Verify voltage stabilizes
    check.that(hw.status.voltage, "~=", 5.0,
               within_duration_s=1.0,
               kwargs={"abs": 0.1})

    # Verify stays stable
    check.that(hw.status.voltage, "is close to", 5.0,
               for_duration_s=2.0,
               kwargs={"abs_tol": 0.05})

    # Verify ready signal
    check.that(hw.status.ready, "is true", within_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, within_duration_s=2.0)
    check.that(fsm.state.previous, "==", 0)

    # Start operation
    fsm.start()

    # Should transition to ACTIVE
    check.that(fsm.state.current, "==", 2, within_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