Skip to content

null

null

Null backend for .protocol.ContainerRuntime.

Runs no subprocesses — useful for tests and dry-run modes. Every handle operation returns a safe default (None, False, [], empty strings). NullRuntime.exec returns an empty success.

State fixtures can be injected via NullRuntime.set_container_state and NullRuntime.add_image when tests need a specific shape.

ExecStdioStep = tuple[str, bytes] module-attribute

NullLogStream

Empty log stream — yields no lines and closes cleanly.

__iter__()

Return self — iteration yields nothing.

Source code in src/terok_sandbox/runtime/null.py 
40
41
42
def __iter__(self) -> NullLogStream:
    """Return self — iteration yields nothing."""
    return self

__next__()

Terminate iteration immediately.

Source code in src/terok_sandbox/runtime/null.py 
44
45
46
def __next__(self) -> str:
    """Terminate iteration immediately."""
    raise StopIteration

__enter__()

Enter no-op context.

Source code in src/terok_sandbox/runtime/null.py 
48
49
50
def __enter__(self) -> NullLogStream:
    """Enter no-op context."""
    return self

__exit__(*exc)

Exit no-op context.

Source code in src/terok_sandbox/runtime/null.py 
52
53
def __exit__(self, *exc: object) -> None:
    """Exit no-op context."""

close()

Nothing to close.

Source code in src/terok_sandbox/runtime/null.py 
55
56
57
def close(self) -> None:
    """Nothing to close."""
    return None

NullPortReservation(host='127.0.0.1')

Real port reservation (binds a socket) — useful even in null mode because callers usually need an actually-free port to hand off.

Source code in src/terok_sandbox/runtime/null.py 
64
65
66
67
68
69
70
71
72
def __init__(self, host: str = "127.0.0.1") -> None:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.bind((host, 0))
    except BaseException:
        s.close()
        raise
    self._socket: socket.socket | None = s
    self.port = s.getsockname()[1]

port = s.getsockname()[1] instance-attribute

__enter__()

Enter the context.

Source code in src/terok_sandbox/runtime/null.py 
74
75
76
def __enter__(self) -> NullPortReservation:
    """Enter the context."""
    return self

__exit__(*exc)

Release the port.

Source code in src/terok_sandbox/runtime/null.py 
78
79
80
def __exit__(self, *exc: object) -> None:
    """Release the port."""
    self.close()

close()

Release the port (idempotent).

Source code in src/terok_sandbox/runtime/null.py 
82
83
84
85
86
def close(self) -> None:
    """Release the port (idempotent)."""
    if self._socket is not None:
        self._socket.close()
        self._socket = None

NullContainer(name, *, runtime)

Container handle that reads from a NullRuntime's fixtures.

Source code in src/terok_sandbox/runtime/null.py 
92
93
94
def __init__(self, name: str, *, runtime: NullRuntime) -> None:
    self.name = name
    self._runtime = runtime

name = name instance-attribute

state property

Return the fixture state, or None when unset.

running property

Return True when the fixture state is "running".

image property

Return the fixture image, or None when unset.

rw_size property

Return the fixture rw_size, or None when unset.

__repr__()

Render as NullContainer(name='...').

Source code in src/terok_sandbox/runtime/null.py 
96
97
98
def __repr__(self) -> str:
    """Render as ``NullContainer(name='...')``."""
    return f"NullContainer(name={self.name!r})"

__eq__(other)

Equality by name.

Source code in src/terok_sandbox/runtime/null.py 
100
101
102
def __eq__(self, other: object) -> bool:
    """Equality by name."""
    return isinstance(other, NullContainer) and self.name == other.name

__hash__()

Hash by name.

Source code in src/terok_sandbox/runtime/null.py 
104
105
106
def __hash__(self) -> int:
    """Hash by name."""
    return hash(("NullContainer", self.name))

start()

Flip the fixture state to "running".

Source code in src/terok_sandbox/runtime/null.py 
129
130
131
def start(self) -> None:
    """Flip the fixture state to ``"running"``."""
    self._runtime._container_states[self.name] = "running"

stop(*, timeout=10)

Flip the fixture state to "exited".

Source code in src/terok_sandbox/runtime/null.py 
133
134
135
def stop(self, *, timeout: int = 10) -> None:
    """Flip the fixture state to ``"exited"``."""
    self._runtime._container_states[self.name] = "exited"

wait(timeout=None)

Return the fixture exit code (default 0).

Source code in src/terok_sandbox/runtime/null.py 
137
138
139
def wait(self, timeout: float | None = None) -> int:
    """Return the fixture exit code (default 0)."""
    return self._runtime._container_exit_codes.get(self.name, 0)

copy_in(src, dest)

Record the copy_in call for test inspection.

Source code in src/terok_sandbox/runtime/null.py 
141
142
143
def copy_in(self, src: Path, dest: str) -> None:
    """Record the copy_in call for test inspection."""
    self._runtime._copy_in_calls.append((self.name, src, dest))

login_command(*, command=())

Return a placeholder login argv.

Source code in src/terok_sandbox/runtime/null.py 
145
146
147
def login_command(self, *, command: tuple[str, ...] = ()) -> list[str]:
    """Return a placeholder login argv."""
    return ["null-exec", self.name, *command]

logs(*, follow=False, tail=None)

Return an empty NullLogStream.

Source code in src/terok_sandbox/runtime/null.py 
149
150
151
def logs(self, *, follow: bool = False, tail: int | None = None) -> LogStream:
    """Return an empty [`NullLogStream`][terok_sandbox.runtime.null.NullLogStream]."""
    return NullLogStream()

stream_initial_logs(ready_check, timeout_sec)

Return the fixture readiness result (default True).

Source code in src/terok_sandbox/runtime/null.py 
153
154
155
156
157
158
159
def stream_initial_logs(
    self,
    ready_check: Callable[[str], bool],
    timeout_sec: float | None,
) -> bool:
    """Return the fixture readiness result (default ``True``)."""
    return self._runtime._ready_results.get(self.name, True)

NullImage(ref, *, runtime)

Image handle backed by NullRuntime fixtures.

Source code in src/terok_sandbox/runtime/null.py 
165
166
167
def __init__(self, ref: str, *, runtime: NullRuntime) -> None:
    self.ref = ref
    self._runtime = runtime

ref = ref instance-attribute

id property

Return the ref itself when the image exists, else None.

repository property

Return the fixture repository, or "".

tag property

Return the fixture tag, or "".

size property

Return the fixture size, or "".

created property

Return the fixture created timestamp, or "".

__repr__()

Render as NullImage(ref='...').

Source code in src/terok_sandbox/runtime/null.py 
169
170
171
def __repr__(self) -> str:
    """Render as ``NullImage(ref='...')``."""
    return f"NullImage(ref={self.ref!r})"

__eq__(other)

Equality by ref.

Source code in src/terok_sandbox/runtime/null.py 
173
174
175
def __eq__(self, other: object) -> bool:
    """Equality by ref."""
    return isinstance(other, NullImage) and self.ref == other.ref

__hash__()

Hash by ref.

Source code in src/terok_sandbox/runtime/null.py 
177
178
179
def __hash__(self) -> int:
    """Hash by ref."""
    return hash(("NullImage", self.ref))

exists()

Return True when the fixture lists this ref.

Source code in src/terok_sandbox/runtime/null.py 
206
207
208
def exists(self) -> bool:
    """Return ``True`` when the fixture lists this ref."""
    return self.ref in self._runtime._image_records

labels()

Return the fixture labels, or {}.

Source code in src/terok_sandbox/runtime/null.py 
210
211
212
def labels(self) -> dict[str, str]:
    """Return the fixture labels, or ``{}``."""
    return dict(self._runtime._image_labels.get(self.ref, {}))

history()

Return the fixture history, or [].

Source code in src/terok_sandbox/runtime/null.py 
214
215
216
def history(self) -> list[str]:
    """Return the fixture history, or ``[]``."""
    return list(self._runtime._image_history.get(self.ref, ()))

remove()

Remove from the fixture; return True when actually removed.

Source code in src/terok_sandbox/runtime/null.py 
218
219
220
221
222
223
224
def remove(self) -> bool:
    """Remove from the fixture; return ``True`` when actually removed."""
    had = self.ref in self._runtime._image_records
    self._runtime._image_records.pop(self.ref, None)
    self._runtime._image_labels.pop(self.ref, None)
    self._runtime._image_history.pop(self.ref, None)
    return had

NullRuntime()

Stub ContainerRuntime for tests and dry-run modes.

All state lives in dictionaries on the runtime instance. Tests pre-populate fixtures via the set_container_state, add_image, etc. helpers.

Source code in src/terok_sandbox/runtime/null.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def __init__(self) -> None:
    self._container_states: dict[str, str] = {}
    self._container_images: dict[str, str] = {}
    self._container_rw_sizes: dict[str, int] = {}
    self._container_exit_codes: dict[str, int] = {}
    self._ready_results: dict[str, bool] = {}
    self._image_records: dict[str, dict[str, str]] = {}
    self._image_labels: dict[str, dict[str, str]] = {}
    self._image_history: dict[str, tuple[str, ...]] = {}
    self._exec_results: dict[tuple[str, tuple[str, ...]], ExecResult] = {}
    self._exec_stdio_scripts: dict[
        tuple[str, tuple[str, ...]], tuple[tuple[ExecStdioStep, ...], int]
    ] = {}
    self._exec_stdio_calls: list[tuple[str, tuple[str, ...], dict[str, str]]] = []
    self._copy_in_calls: list[tuple[str, Path, str]] = []
    self._force_remove_calls: list[list[str]] = []

set_container_state(name, state)

Record state ("running", "exited", ...) for container name.

Source code in src/terok_sandbox/runtime/null.py 
254
255
256
def set_container_state(self, name: str, state: str) -> None:
    """Record *state* (``"running"``, ``"exited"``, ...) for container *name*."""
    self._container_states[name] = state

set_container_image(name, image_ref)

Record the image ref behind container name.

Source code in src/terok_sandbox/runtime/null.py 
258
259
260
def set_container_image(self, name: str, image_ref: str) -> None:
    """Record the image ref behind container *name*."""
    self._container_images[name] = image_ref

set_container_rw_size(name, bytes_)

Record the writable-layer size of container name.

Source code in src/terok_sandbox/runtime/null.py 
262
263
264
def set_container_rw_size(self, name: str, bytes_: int) -> None:
    """Record the writable-layer size of container *name*."""
    self._container_rw_sizes[name] = bytes_

set_exit_code(name, code)

Record the exit code Container.wait will return for name.

Source code in src/terok_sandbox/runtime/null.py 
266
267
268
def set_exit_code(self, name: str, code: int) -> None:
    """Record the exit code [`Container.wait`][terok_sandbox.runtime.Container.wait] will return for *name*."""
    self._container_exit_codes[name] = code

set_ready_result(name, ready)

Record the outcome Container.stream_initial_logs returns.

Source code in src/terok_sandbox/runtime/null.py 
270
271
272
def set_ready_result(self, name: str, ready: bool) -> None:
    """Record the outcome [`Container.stream_initial_logs`][terok_sandbox.runtime.Container.stream_initial_logs] returns."""
    self._ready_results[name] = ready

add_image(ref, *, repository='', tag='', size='', created='', labels=None, history=())

Register an image fixture.

Source code in src/terok_sandbox/runtime/null.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def add_image(
    self,
    ref: str,
    *,
    repository: str = "",
    tag: str = "",
    size: str = "",
    created: str = "",
    labels: dict[str, str] | None = None,
    history: tuple[str, ...] = (),
) -> None:
    """Register an image fixture."""
    self._image_records[ref] = {
        "repository": repository,
        "tag": tag,
        "size": size,
        "created": created,
    }
    if labels:
        self._image_labels[ref] = dict(labels)
    if history:
        self._image_history[ref] = tuple(history)

set_exec_result(container_name, cmd, result)

Pre-register the result exec returns for exact cmd.

Source code in src/terok_sandbox/runtime/null.py 
297
298
299
300
301
302
303
304
def set_exec_result(
    self,
    container_name: str,
    cmd: tuple[str, ...],
    result: ExecResult,
) -> None:
    """Pre-register the result [`exec`][exec] returns for exact *cmd*."""
    self._exec_results[(container_name, cmd)] = result

set_exec_stdio_script(container_name, cmd, script, *, exit_code=0)

Pre-register a stdio interaction for exec_stdio.

script is a sequence of ("read", bytes) / ("write", bytes) steps replayed in order: read consumes the matching prefix from the caller-supplied stdin; write emits the bytes to stdout. Use this to drive deterministic ACP-handshake tests without spinning up a real container.

Source code in src/terok_sandbox/runtime/null.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def set_exec_stdio_script(
    self,
    container_name: str,
    cmd: tuple[str, ...],
    script: tuple[ExecStdioStep, ...],
    *,
    exit_code: int = 0,
) -> None:
    """Pre-register a stdio interaction for [`exec_stdio`][terok_sandbox.runtime.null.NullRuntime.exec_stdio].

    *script* is a sequence of ``("read", bytes)`` / ``("write", bytes)``
    steps replayed in order: ``read`` consumes the matching prefix from
    the caller-supplied *stdin*; ``write`` emits the bytes to *stdout*.
    Use this to drive deterministic ACP-handshake tests without spinning
    up a real container.
    """
    self._exec_stdio_scripts[(container_name, cmd)] = (tuple(script), exit_code)

container(name)

Return a NullContainer handle.

Source code in src/terok_sandbox/runtime/null.py 
326
327
328
def container(self, name: str) -> Container:
    """Return a [`NullContainer`][terok_sandbox.runtime.null.NullContainer] handle."""
    return NullContainer(name, runtime=self)

containers_with_prefix(prefix)

Return fixtures whose name starts with prefix-.

Source code in src/terok_sandbox/runtime/null.py 
330
331
332
333
334
335
336
def containers_with_prefix(self, prefix: str) -> list[Container]:
    """Return fixtures whose name starts with ``prefix-``."""
    return [
        NullContainer(name, runtime=self)
        for name in self._container_states
        if name.startswith(f"{prefix}-")
    ]

image(ref)

Return a NullImage handle.

Source code in src/terok_sandbox/runtime/null.py 
338
339
340
def image(self, ref: str) -> Image:
    """Return a [`NullImage`][terok_sandbox.runtime.null.NullImage] handle."""
    return NullImage(ref, runtime=self)

images(*, dangling_only=False)

Return fixture images; dangling_only filters by tag == "<none>".

Source code in src/terok_sandbox/runtime/null.py 
342
343
344
345
346
347
348
349
def images(self, *, dangling_only: bool = False) -> list[Image]:
    """Return fixture images; *dangling_only* filters by ``tag == "<none>"``."""
    images: list[Image] = []
    for ref, rec in self._image_records.items():
        if dangling_only and rec.get("tag") != "<none>":
            continue
        images.append(NullImage(ref, runtime=self))
    return images

exec(container, cmd, *, timeout=None)

Return a pre-registered result, or a default empty success.

Source code in src/terok_sandbox/runtime/null.py 
351
352
353
354
355
356
357
358
359
360
def exec(
    self,
    container: Container,
    cmd: list[str],
    *,
    timeout: float | None = None,
) -> ExecResult:
    """Return a pre-registered result, or a default empty success."""
    key = (container.name, tuple(cmd))
    return self._exec_results.get(key, ExecResult(exit_code=0, stdout="", stderr=""))

exec_stdio(container, cmd, *, stdin, stdout, stderr=None, env=None, timeout=None)

Replay a pre-registered stdio script, or no-op with exit code 0.

Records every call (with env) for test inspection. When a script is registered for (container, cmd), replays it in order: read consumes from stdin and asserts a match; write pushes bytes to stdout. Without a script, returns immediately with exit code 0 — matches the empty-success default of exec.

Source code in src/terok_sandbox/runtime/null.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def exec_stdio(
    self,
    container: Container,
    cmd: list[str],
    *,
    stdin: BinaryIO,
    stdout: BinaryIO,
    stderr: BinaryIO | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
) -> int:
    """Replay a pre-registered stdio script, or no-op with exit code 0.

    Records every call (with env) for test inspection.  When a script is
    registered for ``(container, cmd)``, replays it in order: ``read``
    consumes from *stdin* and asserts a match; ``write`` pushes bytes to
    *stdout*.  Without a script, returns immediately with exit code 0
    — matches the empty-success default of [`exec`][terok_sandbox.runtime.null.NullRuntime.exec].
    """
    key = (container.name, tuple(cmd))
    self._exec_stdio_calls.append((container.name, tuple(cmd), dict(env or {})))
    script_entry = self._exec_stdio_scripts.get(key)
    if script_entry is None:
        return 0
    script, exit_code = script_entry
    for direction, payload in script:
        if direction == "read":
            got = stdin.read(len(payload))
            if got != payload:
                raise AssertionError(
                    f"NullRuntime.exec_stdio script mismatch for {cmd!r}: "
                    f"expected {payload!r}, got {got!r}"
                )
        elif direction == "write":
            stdout.write(payload)
            stdout.flush()
        else:
            raise ValueError(f"unknown exec_stdio script direction: {direction!r}")
    return exit_code

force_remove(containers)

Record the call and clear every fixture for each container.

Source code in src/terok_sandbox/runtime/null.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def force_remove(self, containers: list[Container]) -> list[ContainerRemoveResult]:
    """Record the call and clear every fixture for each container."""
    names = [c.name for c in containers]
    self._force_remove_calls.append(names)
    for name in names:
        self._container_states.pop(name, None)
        self._container_images.pop(name, None)
        self._container_rw_sizes.pop(name, None)
        self._container_exit_codes.pop(name, None)
        self._ready_results.pop(name, None)
        # Drop any pre-registered exec results keyed by this container name
        self._exec_results = {
            key: result for key, result in self._exec_results.items() if key[0] != name
        }
    return [ContainerRemoveResult(name=n, removed=True) for n in names]

reserve_port(host='127.0.0.1')

Reserve a real host port (even null backend callers want a live port).

Source code in src/terok_sandbox/runtime/null.py 
418
419
420
def reserve_port(self, host: str = "127.0.0.1") -> PortReservation:
    """Reserve a real host port (even null backend callers want a live port)."""
    return NullPortReservation(host)