plesty.lib.test.client_field_test

The field test one layer up: the path an experiment actually uses.

FieldTestPipeline drives a device in-process, on the host. No experiment reaches an instrument that way. Every one of them goes:

DeviceTCPIPClient → ZMQ → DeviceTCPIPServer → async wrapper → device

and until now nothing verified that path. A module could hold a green 8/8 host-tier report and fail the first time an experiment talked to it, because the report covered the bottom of the chain and the experiment used all of it. Five of the defects the 2026-08-04 round surfaced live above the device layer, and every one was found by running an experiment rather than by a test.

Two modes, and the difference between them is not a detail

Attached — ClientFieldTest(address="tcp://192.168.1.40:5555"). The server runs on the instrument PC and this runs on another machine on the same network. This is the deployment PLESTY actually uses, and the only mode whose timings mean anything: finding 13 (ZMQ sockets losing messages across threads) and finding 14 (client resends collapsing a serialized server) both need real network latency to appear, and neither reproduces on loopback. The runner cannot start or restart a server on another machine, so gates that need a restart skip and say so.

Spawned — ClientFieldTest(DemoDevice). The runner starts a server in a local subprocess and talks to it over loopback. No instrument, no second machine, so it runs in CI — the first field-test tier that can. It exercises the protocol, the wrapper and the resource manager honestly, and its latencies describe loopback and nothing else.

The report records which mode produced it. A latency profile from loopback and one from a LAN are not comparable, and “client = host + transport” is only a number when the transport was real.

Usage on the instrument network:

# on the instrument PC: the module's own server, already running
# on this machine:
report = ClientFieldTest(
    address="tcp://192.168.1.40:5555",
    params=["wavelength"],
).run()

Usage in CI:

from plesty.lib.sim.demo_device import DemoDevice

report = ClientFieldTest(DemoDevice, params=["wavelength"]).run()
raise SystemExit(0 if report.ok else 1)

Attributes

`_SERVER_START_TIMEOUT_S`
`_CLIENT_TIMEOUT_MS`
`_ABANDON_TIMEOUT_S`
`ATTACHED`
`SPAWNED`

Classes

ClientFieldTest

Run the standard gates against a device through its server.

Functions

`_free_port`(→ int)	Return a port nothing is listening on.
`_serve`(→ None)	Run a device server until killed. The spawned subprocess entry point.

Module Contents

plesty.lib.test.client_field_test._SERVER_START_TIMEOUT_S = 15.0

plesty.lib.test.client_field_test._CLIENT_TIMEOUT_MS = 5000

plesty.lib.test.client_field_test._ABANDON_TIMEOUT_S = 0.001

plesty.lib.test.client_field_test.ATTACHED = 'attached'

plesty.lib.test.client_field_test.SPAWNED = 'spawned'

plesty.lib.test.client_field_test._free_port() → int

Return a port nothing is listening on.

Returns:: A port number a spawned server can bind.
Return type:: int

plesty.lib.test.client_field_test._serve(device_factory: Any, address: str, fixed_threading: bool) → None

Run a device server until killed. The spawned subprocess entry point.

Parameters:

device_factory (Any) – Called with no arguments to build the device.
address (str) – ZMQ bind address.
fixed_threading (bool) – Pin device calls to one dedicated thread.

Return type:

None

class plesty.lib.test.client_field_test.ClientFieldTest(device_factory: Any = None, address: str | None = None, params: collections.abc.Sequence[str] | None = None, safe_ops: collections.abc.Sequence[str] | None = None, resources: collections.abc.Sequence[str] | None = None, expected_types: dict[str, type] | None = None, op_kinds: dict[str, plesty.lib.device.device_utils.OpKind | dict[str, Any]] | None = None, fixed_threading: bool = True, label: str = '', instrument: plesty.lib.test.report_artifact.Instrument | None = None, baseline: str | pathlib.Path | None = None, project_root: str | pathlib.Path | None = None, reports_dir: str | pathlib.Path | None = None)

Run the standard gates against a device through its server.

Parameters:

device_factory (Any) – Called with no arguments to build a device for a locally spawned server. Omit when attaching to a server on the instrument PC.
address (str | None) – Server address. Required when attaching — the instrument PC’s address on the local network, e.g. "tcp://192.168.1.40:5555". Chosen automatically on loopback when spawning.
params (collections.abc.Sequence[str] | None) – Configuration keys to round-trip over the wire. The client presents query/write and proxies for the served functions but no get_config_list(), so the keys are named here.
safe_ops (collections.abc.Sequence[str] | None) – Served functions the run may call. Nothing not named here is called: the client tier inherits the host tier’s safety judgement rather than making a second one that could disagree with it.
op_kinds (dict[str, plesty.lib.device.device_utils.OpKind | dict[str, Any]] | None) – What each operation does, per name, as the host tier was told (OpKind or its fields as a mapping). The client cannot read the module’s declaration through describe’s flat method list, so the same kinds are handed over: a motion is driven one step out from what its position_key reports and back, a lifecycle once, a control once after the motions, a configure never.
resources (collections.abc.Sequence[str] | None) – Lockable resources to contend for. Two clients ask for these at once, and the second must be refused.
expected_types (dict[str, type] | None) – Parameter key to the type it is declared as. The timeout gate uses it to tell an orphaned reply from a fresh one — without it, a reply that belongs to another parameter is invisible when both are floats.
fixed_threading (bool) – Threading mode for a spawned server. Ignored when attaching — the mode is whatever the running server was started with.
label (str) – Short name for this run, for comparing several.
instrument (plesty.lib.test.report_artifact.Instrument | None) – What the published report says this run was pointed at.
baseline (str | pathlib.Path | None) – The host tier’s published field-test.json. Its coverage block says what that run actually reached on this instrument — already filtered by what the fitted hardware supports — so taking the surface from it makes the two tiers comparable without maintaining a second exclusion list. SAFE_OPS says what is permitted; the baseline says what worked.
project_root (str | pathlib.Path | None) – Directory holding the module’s pyproject.toml. The published report identifies the module from it, the same way the host tier does — one reader of a module’s identity, not two.
reports_dir (str | pathlib.Path | None) – Where to publish. Given all three of these, the run writes <reports_dir>/<slug>/field-test-client.json — beside the host tier’s report rather than over it, since the two answer different questions about the same instrument.

Store the configuration; nothing is started until run().

device_factory = None

mode = 'spawned'

address = 'tcp://127.0.0.1:Instance of builtins.int'

params = []

safe_ops = []

op_kinds: dict[str, plesty.lib.device.device_utils.OpKind]

resources = []

_expected_types

fixed_threading = True

label = ''

instrument = None

project_root

reports_dir = None

_server: Any = None

_timed_connect: float | None = None

_samples: dict[str, list[float]]

_misses: dict[str, int]

_called: set[str]

baseline

surface_source = ''

_target: dict[str, int]

_start_server() → None

Spawn a local server and wait until it answers.

Does nothing when attached: the server belongs to another machine.

Raises:: RuntimeError – If the spawned server does not answer in time.
Return type:: None

_answers() → bool

Return whether something is listening on the server’s port yet.

A plain TCP connect, deliberately, rather than a real handshake. A client built against a server that is not up leaves an unsent message in its ZMQ context, and ctx.term() then blocks forever waiting to deliver it — the readiness probe hung instead of reporting “not yet”.

Returns:: True when the port accepts a connection.
Return type:: bool

_stop_server() → None

Terminate the spawned server. Never touches an attached one.

Tolerates a process that was created but never fully started: that happens when the spawn itself failed, and raising here would replace the real cause with an AttributeError from the cleanup path.

Return type:: None

_client(timeout_ms: int = _CLIENT_TIMEOUT_MS, client_id: str = 'field-test', resources: collections.abc.Sequence[str] | None = None) → Any

Return a client pointed at the server under test.

Parameters:

timeout_ms (int) – Receive timeout in milliseconds.
client_id (str) – ZMQ ROUTER identity. Every concurrent client needs its own: the identity is what the server addresses replies to, so two sockets sharing one have their replies misrouted, and the symptom is a timeout that looks like a dead server.
resources (collections.abc.Sequence[str] | None) – Lockable resources to request during the handshake.

Returns:

A DeviceTCPIPClient.

Return type:

Any

test_connect(client: Any) → plesty.lib.test.field_test.GateResult

The handshake answers with this client’s identity.

Parameters:: client (Any) – A connected client.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

test_describe(client: Any) → plesty.lib.test.field_test.GateResult

The server advertises the functions the client binds proxies from.

Parameters:: client (Any) – A connected client.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

test_protocol_fidelity(client: Any) → plesty.lib.test.field_test.GateResult

Declared parameters cross JSON with their type intact.

Parameters:: client (Any) – A connected client.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

test_functions(client: Any) → plesty.lib.test.field_test.GateResult

Call every operation the host tier is allowed to call, over the wire.

The two tiers only compare if they exercise the same surface. The host tier calls safe_ops; a client tier that called nothing left “client = host + transport” undefined for operations, and published a coverage block claiming 21 operations declared and none reached.

The allow-list is the host tier’s, inherited rather than decided again: an operation unsafe on the bench is unsafe over the wire.

Parameters:: client (Any) – A connected client.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

_motion_over_the_wire(client: Any, name: str, kind: plesty.lib.device.device_utils.OpKind) → Any

Drive a motion one step out and back, as many times as the host tier did.

The host tier’s count for a motion is out-and-back pairs, so half of it is the number of pairs to make here; at least one.

Parameters:

client (Any) – A connected client.
name (str) – The motion operation.
kind (plesty.lib.device.device_utils.OpKind) – Its declaration — position_key, target and step must all be known.

Returns:

The answer to the last call.

Raises:

ValueError – If the declaration is incomplete.

Return type:

Any

test_session_survives_a_bad_result(client: Any) → plesty.lib.test.field_test.GateResult

A call whose result JSON cannot carry must not end the session.

Finding 3 of the 2026-08-04 round. Encoding happened in the server loop rather than in the request handler, so one client calling one method that returned a dataclass took the instrument away from every other client — and it looked like a timeout, because nothing answered.

Parameters:: client (Any) – A connected client.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

test_wrapper_modes() → plesty.lib.test.field_test.GateResult

Compare fixed against pool threading — needs to restart the server.

Returns:: The gate result, skipped when attached.
Return type:: plesty.lib.test.field_test.GateResult

_median_query_ms(key: str, repetitions: int = 21) → float

Return the median latency of a repeated query, in milliseconds.

Parameters:

key (str) – Parameter to query.
repetitions (int) – How many queries to time.

Returns:

The median round-trip time.

Return type:

float

test_timeout_recovery(client: Any) → plesty.lib.test.field_test.GateResult

A timed-out request must not leave its answer for the next one.

The host tier proved the VISA transport is not where orphaned replies come from and said to search the layer above. This is that layer: the client socket carries an identity and requests are not lock-stepped, so a reply that arrives after its caller gave up is available for whoever reads next — and the symptom is a parameter answering with another parameter’s value, which no type check catches when both are floats.

Parameters:: client (Any) – A connected client, used only for its address.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

test_resource_allocation(client: Any) → plesty.lib.test.field_test.GateResult

A resource held by one client is refused to the next, and released.

Parameters:: client (Any) – A connected client, unused; the gate needs its own.
Returns:: The gate result.
Return type:: plesty.lib.test.field_test.GateResult

_guarded(name: str, gate: Any, *args: Any) → plesty.lib.test.field_test.GateResult

Run one gate, turning a raised exception into a failed result.

A gate that raises took the whole run with it, so the gates that had already passed were lost and the report was never written — the first real-network run ended in a traceback with nothing to show for the four gates that had succeeded. A gate that cannot answer has failed; it has not made the other gates unanswerable.

Parameters:

name (str) – Gate name, for the result when it raises.
gate (Any) – The bound gate method.
*args (Any) – Arguments to pass it.

Returns:

The gate’s result, or a failure carrying what it raised.

Return type:

plesty.lib.test.field_test.GateResult

run() → plesty.lib.test.field_test.FieldTestReport

Start a server if this runner owns one, run every gate, and report.

Returns:: The report, with tier="client".
Return type:: plesty.lib.test.field_test.FieldTestReport

_time(report: plesty.lib.test.field_test.FieldTestReport, command: str, seconds: float, ok: bool) → None

Record one lifecycle call’s latency and outcome.

The published document builds its lifecycle block from these, so a tier that timed nothing publishes no lifecycle at all — which reads as a run that never connected rather than one that did not measure.

Parameters:

report (plesty.lib.test.field_test.FieldTestReport) – The report to record into.
command (str) – Lifecycle command name.
seconds (float) – How long the call took.
ok (bool) – Whether it succeeded.

Return type:

None

_timed_call(stat_key: str, call: Any, *args: Any) → Any

Make one call over the wire, recording what it cost.

The key is the prefixed form the coverage block looks statistics up under — query:<key>, write:<key>, op:<name>. Recording under the bare name published a coverage block whose calls were all empty, which reads as a surface verified without ever being touched.

Parameters:

stat_key (str) – Prefixed statistics key for this call.
call (Any) – The client method to invoke.
*args (Any) – Its arguments.

Returns:

Whatever the call returned.

Raises:

Exception – Whatever the call raised, after recording the miss.

Return type:

Any

adopt_baseline() → str

Take the exercised surface from the host tier’s report.

The host tier resolves what the fitted sensor supports and skips the rest — calling an unsupported operation costs a VISA timeout, a device clear, a retry and a session reopen, roughly ten seconds of blocked device thread each, during which no other client is served. Its coverage block records the outcome: called for what it reached, declared for what it did not. Adopting that is how the two tiers exercise the same surface without a second list to keep in step.

Returns:: One line saying which surface this run used and why, for the report.
Return type:: str

_adopt_attempts(coverage: dict[str, Any]) → None

Match the host tier’s call counts, command by command.

The host tier repeats each command to build a distribution; calling once here would compare a single sample against twenty, and the difference in shape would read as transport cost rather than as a difference in sample size.

Parameters:: coverage (dict[str, Any]) – The baseline’s coverage block.
Return type:: None

_repeat(stat_key: str) → int

Return how many times to make one call.

Parameters:: stat_key (str) – The prefixed statistics key.
Returns:: The host tier’s count for it, or one when there is no baseline.
Return type:: int

_baseline_is_stale(document: dict[str, Any]) → str

Return a warning when the baseline predates the module under test.

Used anyway: a stale baseline still describes a real instrument, and refusing would leave the tiers incomparable over a version bump that may have changed nothing.

Parameters:: document (dict[str, Any]) – The baseline document.
Returns:: The warning, or an empty string.
Return type:: str

_record_coverage(report: plesty.lib.test.field_test.FieldTestReport, served: collections.abc.Sequence[str]) → None

Say what this tier reached, against what the server offers.

An empty coverage block reads as a run that exercised nothing, and the published page renders it that way. What the client tier can honestly claim is narrower than the host tier’s: it has no get_config_list() to enumerate parameters, so the declared set is what this run was told to round-trip, and the operations are what describe advertises.

Parameters:

report (plesty.lib.test.field_test.FieldTestReport) – The report to record into.
served (collections.abc.Sequence[str]) – Operation names the server advertised.

Return type:

None

publish(report: plesty.lib.test.field_test.FieldTestReport) → pathlib.Path | None

Write the sanitised report, when this run was told where to.

The projection is the host tier’s — the same FieldTestReport.to_public_document(), which drops the host name, the address and every driver message. A second projection would be a second chance to publish something that should not leave the lab.

Parameters:: report (plesty.lib.test.field_test.FieldTestReport) – The finished report.
Returns:: The path written, or None when the run has no instrument, module or reports directory to publish under.
Return type:: pathlib.Path | None