plesty.bench.client

The bench client: one ZMQ DEALER socket, three request types.

The wire is the plesty-lib device-server wire — JSON frames, status: ok or status: error replies — so a plesty.lib.service.build_client() or a describe probe understands the agent too. Requests:

{"type": "describe"}

→ {"status": "ok", "result": {"kind": "plesty-server-agent", "version": …, "home": …, "commands": [...]}}

{"type": "exec", "argv": [...], "token": "…"}

→ {"status": "ok", "result": {"code": 0, "stdout": "…", "stderr": "…"}}

{"type": "fetch", "path": "relative/to/home", "token": "…"}

→ {"status": "ok", "result": {"path": …, "size": n, "data": "<base64>"}}

Everything long-running is already a job on the bench: ask for it with --no-wait, then BenchClient.follow() the job log.

Attributes

DEFAULT_PORT

ADDRESS_ENV

TOKEN_ENV

Exceptions

BenchClientError

The bench did not answer, refused the request, or the command failed to run.

Classes

ExecResult

What one remote plesty-server invocation produced.

BenchClient

A connection to one bench agent.

Functions

parse_address(→ str)

Turn host, host:port or tcp://host:port into a ZMQ endpoint.

Module Contents

plesty.bench.client.DEFAULT_PORT = 5550
plesty.bench.client.ADDRESS_ENV = 'PLESTY_BENCH'
plesty.bench.client.TOKEN_ENV = 'PLESTY_BENCH_TOKEN'
exception plesty.bench.client.BenchClientError

Bases: RuntimeError

The bench did not answer, refused the request, or the command failed to run.

Initialize self. See help(type(self)) for accurate signature.

class plesty.bench.client.ExecResult

What one remote plesty-server invocation produced.

Variables:

  • argv – The arguments that were run.

  • code – Exit code.

  • stdout – Captured standard output.

  • stderr – Captured standard error (progress lines).

argv: tuple[str, Ellipsis]
code: int
stdout: str
stderr: str
property ok: bool

Whether the command exited 0.

Return type:

bool

json() → Any

Parse stdout as JSON (for commands run with --json).

Return type:

Any

plesty.bench.client.parse_address(value: str | None) → str

Turn host, host:port or tcp://host:port into a ZMQ endpoint.

Parameters:

value (str | None) – The address; None reads PLESTY_BENCH, else localhost.

Return type:

str

class plesty.bench.client.BenchClient(address: str | None = None, token: str | None = None, timeout_ms: int = 10000, context: zmq.Context[Any] | None = None)

A connection to one bench agent.

Connect lazily to address.

Parameters:

  • address (str | None) – host, host:port or tcp://host:port; None reads PLESTY_BENCH.

  • token (str | None) – Shared secret the agent was started with; None reads PLESTY_BENCH_TOKEN.

  • timeout_ms (int) – Default reply timeout for request().

  • context (zmq.Context[Any] | None) – ZMQ context to use; the global instance by default.

address
token
timeout_ms = 10000
_context = None
_socket: zmq.Socket[Any] | None = None
__enter__() → BenchClient

Open the socket.

Return type:

BenchClient

__exit__(*_exc: object) → None

Close the socket.

Parameters:

_exc (object)

Return type:

None

close() → None

Close the socket (a later request reconnects).

Return type:

None

_connect() → zmq.Socket[Any]
Return type:

zmq.Socket[Any]

request(message: dict[str, Any], timeout_ms: int | None = None) → dict[str, Any]

Send one request and return the result of an ok reply.

Raises:

BenchClientError – On no reply in time, a malformed reply, or an error reply (its message is the exception text).

Parameters:

  • message (dict[str, Any])

  • timeout_ms (int | None)

Return type:

dict[str, Any]

describe(timeout_ms: int | None = None) → dict[str, Any]

The agent’s self-description (kind, version, home, commands).

Parameters:

timeout_ms (int | None)

Return type:

dict[str, Any]

ping(timeout_ms: int = 2000) → bool

Whether a bench agent answers at the address.

Parameters:

timeout_ms (int)

Return type:

bool

exec(argv: collections.abc.Sequence[str], timeout_ms: int | None = None) → ExecResult

Run plesty-server <argv> on the bench and return what it produced.

The result carries the exit code; it does not raise on a non-zero exit, only when the agent could not run the command at all.

Parameters:

  • argv (collections.abc.Sequence[str])

  • timeout_ms (int | None)

Return type:

ExecResult

fetch(path: str, timeout_ms: int | None = None) → bytes

Download a file from the bench home (path relative to it).

Parameters:

  • path (str)

  • timeout_ms (int | None)

Return type:

bytes

follow(job_id: str, sink: collections.abc.Callable[[str], None], poll_s: float = 0.5, timeout_s: float | None = None) → int

Stream the log of job_id to sink until the job ends; returns its exit code.

Uses jobs log <id> --offset N --json on the bench, so nothing is held open between polls and a lost connection just resumes.

Raises:

BenchClientError – When the job is unknown or the bench stops answering.

Parameters:

  • job_id (str)

  • sink (collections.abc.Callable[[str], None])

  • poll_s (float)

  • timeout_s (float | None)

Return type:

int