DevicePipeline — PLESTY Highlight ④

What is it?

Overview

DevicePipeline is a class in plesty.lib.test.device_pipeline. Instantiate it once in your test file with a VISA-free mock device class and the paths to your JSON schemas. Then call its 8 gate methods from plain pytest functions — one line each.

It handles all the heavy lifting: schema loading and validation, mock solver binding, lifecycle management, assertion logic, and error reporting. You provide the mock device; DevicePipeline does the rest.

DevicePipeline is also what Gate d1 of plesty check validates. When module_type = "device" is set in your pyproject.toml, the check command verifies all 8 gate functions exist in tests/ and pass without hardware.

Testing Philosophy

❌ Old way — every module on its own

Each device module reinvents its own mock test fixtures
Some test connect(); most skip it entirely
No standard for what "a passing device module" means
CI is inconsistent — modules pass on different criteria
Experiment authors can't trust the device API contract
When a device breaks, hard to tell if it's a regression

✓ DevicePipeline way — one contract

One import. 8 one-liner test stubs. Identical for every module.
Connect, lifecycle, identity, error checking — all verified
Experiment authors know exactly what guarantees they have
Gate d1 in CI enforces the contract on every push
Regressions caught immediately — the mock gates never pass silently
New contributors follow the same pattern without guessing

Architecture

_MockMyDevice

schema_param.json

schema_func.json

→

DevicePipeline(...)

→

test_schema_integrity()

test_lifecycle()

test_check_errors() …

→

pytest

The mock device and schemas flow into a single PIPELINE instance. All 8 gate methods are called from standalone pytest functions — no class, no fixture setup.

The 8 Gates in Depth

1 test_schema_integrity ▶

Loads both schema_param.json and schema_func.json. Validates that both are well-formed JSON and contain the required top-level structure.

Catches: malformed JSON (trailing commas, unquoted keys), wrong encoding, missing required top-level keys, empty schema files.

Fix: Run your schema through a JSON linter. Ensure both files are UTF-8 encoded and parseable with json.load().

2 test_param_key_resolution ▶

Calls get_config_list() on the mock device to get all declared parameter keys, then calls get_config(key) for each one. Asserts every key resolves without raising.

Catches: broken param solver binding in init(), missing SCPI command mappings, keys declared in the schema but not wired to the solver.

Fix: Check init() in your mock device class. Ensure SCPISolver is assigned to self._param_solver before any param access.

3 test_params_mock ▶

Round-trips all config parameters: reads each via get_config(key) and writes a valid value back via set_config(key, value) using a mock solver. Asserts no exceptions are raised.

Catches: params that are read-only when they shouldn't be, type mismatches between schema and solver, missing write command format strings.

Fix: Ensure your SCPISolver write format (wfmt) is set and matches the SCPI command expected by each parameter.

4 test_funcs_mock ▶

Calls every registered operation (from schema_func.json) against a mock solver. Asserts each returns a dict — not None, not raises, not a raw string.

Catches: operations wired with FuncMeta but whose mock response can't be cast to the expected type, custom op solvers that return the wrong shape.

Fix: Set _mock_tm.send_command.return_value to a string that can be cast to the schema's output_dtype (e.g. "1.234" for float).

5 test_lifecycle ▶

Uses the mock device as a context manager (with device: ...). Asserts is_operatable is True inside the block and False after the block exits.

Catches: connect() not setting self._tm, disconnect() not clearing solvers, __enter__/__exit__ not calling connect/disconnect.

Fix: In connect(), assign self._tm = self._mock_tm. In disconnect(), set self._tm = None and self._param_solver = None.

6 test_identity ▶

Calls identity() on a connected mock device. Asserts the return value is a non-empty string.

Catches: missing identity() implementation, returning None, returning empty string, returning a non-string type.

Fix: Implement identity() in your mock class to return a fixed vendor IDN string: return "VENDOR,MODEL,SN00001,1.0.0".

7 test_check_errors ▶

Calls check_errors() on a healthy mock device (with no errors injected). Asserts the return is an empty list [].

Catches: check_errors() returning None, raising an exception, returning a non-list type, or returning a non-empty list on a healthy device.

Fix: Implement check_errors() to return list[str]. Wire _mock_tm.send_command.return_value = '0,"No error"' so parsing returns an empty error list.

8 test_state_coverage ▶

Calls dev.state (the property that snapshots all param values) and asserts the returned dict's keys are a superset of get_config_list(). Every declared param must appear in the state snapshot.

Catches: params declared in the schema but excluded from _get_state(), incomplete state snapshots that would break experiment logging.

Fix: In _get_state(), iterate get_config_list() and include every key. Don't manually list param names — use the schema as the authoritative source.

Complete Worked Example

tests/test_mydevice.py

# tests/test_mydevice.py — complete worked example
from __future__ import annotations

import uuid
from typing import Any
from unittest.mock import MagicMock

from plesty.lib.solver.scpi import SCPISolver
from plesty.lib.test.device_pipeline import DevicePipeline
from plesty.mydevice.base_device import _MyDeviceOpSolver
from plesty.mydevice.device import Device


# ── Step 1: VISA-free mock device ─────────────────────────────────────────────

class _MockMyDevice(Device):
    """VISA-free mock. Each instance gets a unique address to avoid registry
    collisions when tests run in parallel."""

    def __init__(self) -> None:
        super().__init__(
            address=f"USB::MOCK::{uuid.uuid4().hex[:8].upper()}::INSTR"
        )
        self._mock_tm: MagicMock = MagicMock()
        self._mock_tm.is_open = True
        self._mock_tm.send_command.return_value = '0,"No error"'
        self._tm = self._mock_tm

    def init(self, main: Any = None) -> None:
        self._param_solver = SCPISolver(qfmt="{cmd}?", wfmt="{cmd} {value}")
        self.bind_op_solver(_MyDeviceOpSolver(self._mock_tm))

    def connect(self) -> bool:
        if self._param_solver is None:
            self.init()
        self._mock_tm.is_open = True
        self._tm = self._mock_tm
        return True

    def disconnect(self) -> None:
        self._tm = None
        self._param_solver = None

    def identity(self) -> str:
        return "VENDOR,MODEL,SN00001,1.0.0"


# ── Step 2: Instantiate the pipeline (shared across all gate tests) ───────────

PIPELINE = DevicePipeline(
    _MockMyDevice,
    param_schema="plesty/mydevice/schema_param.json",
    op_schema="plesty/mydevice/schema_func.json",
)


# ── Step 3: Write the 8 gate functions (required by plesty check Gate d1) ─────

def test_schema_integrity() -> None:
    PIPELINE.test_schema_integrity()

def test_param_key_resolution() -> None:
    PIPELINE.test_param_key_resolution()

def test_params_mock() -> None:
    PIPELINE.test_params_mock()

def test_funcs_mock() -> None:
    PIPELINE.test_funcs_mock()

def test_lifecycle() -> None:
    PIPELINE.test_lifecycle()

def test_identity() -> None:
    PIPELINE.test_identity()

def test_check_errors() -> None:
    PIPELINE.test_check_errors()

def test_state_coverage() -> None:
    PIPELINE.test_state_coverage()

What _MockMyDevice Replaces

Mock wiring guide

The mock device replaces the real VISA traffic manager (self._tm) with a MagicMock instance. Three attributes must be wired correctly:

# Required mock wiring in __init__
self._mock_tm.is_open = True               # → makes is_operatable return True
self._mock_tm.send_command.return_value = '0,"No error"'  # → check_errors() returns []
self._tm = self._mock_tm                   # → routes all solver calls through mock

# Methods you must implement in _MockMyDevice
# init()        — bind SCPISolver + op solver
# connect()     — set self._tm = self._mock_tm, return True
# disconnect()  — set self._tm = None, self._param_solver = None
# identity()    — return a fixed IDN string
# Everything else is inherited from plesty.lib.device.Device

Common Failure Modes

Gate Function	Cause	Fix
test_schema_integrity	schema_param.json has trailing comma	Fix JSON syntax — no trailing commas in JSON
test_param_key_resolution	`get_config("key")` raises KeyError	Check solver binding in `init()`
test_params_mock	Write command format string missing	Set `wfmt="{cmd} {value}"` in SCPISolver
test_funcs_mock	Mock response can't be cast to float	Set `send_command.return_value = "1.234"`
test_lifecycle	`connect()` doesn't assign `self._tm`	Add `self._tm = self._mock_tm` in `connect()`
test_identity	`identity()` not implemented	Return a non-empty string from `identity()`
test_check_errors	`check_errors()` returns `None`	Change return type to `list[str]`, return `[]`
test_state_coverage	`state` dict missing param keys	Iterate `get_config_list()` in `_get_state()`

Import Reference

Import from plesty.lib.test.device_pipeline import DevicePipeline

Dev dep plesty-lib >= 0.2.6

DevicePipeline — Standard Device Test Harness

❌ Old way — every module on its own

✓ DevicePipeline way — one contract