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

def __iter__(self) -> NullLogStream:
    """Return self — iteration yields nothing."""
    return self

__next__()

Terminate iteration immediately.

Source code in src/terok_sandbox/runtime/null.py

def __next__(self) -> str:
    """Terminate iteration immediately."""
    raise StopIteration

__enter__()

Enter no-op context.

Source code in src/terok_sandbox/runtime/null.py

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

def __exit__(self, *exc: object) -> None:
    """Exit no-op context."""

close()

Nothing to close.

Source code in src/terok_sandbox/runtime/null.py

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

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

def __enter__(self) -> NullPortReservation:
    """Enter the context."""
    return self

__exit__(*exc)

Release the port.

Source code in src/terok_sandbox/runtime/null.py

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

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

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".

started_at property

Return the fixture start timestamp, or None when unset.

image property

Return the fixture image, or None when unset.

rw_size property

Return the fixture rw_size, or None when unset.

id property

Return the fixture container ID, or None when unset.

mounts property

Return the fixture mounts, or [] when unset.

env property

Fixture environment recorded for this container; {} when unset.

__repr__()

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

Source code in src/terok_sandbox/runtime/null.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

def __init__(self) -> None:
    self._container_states: dict[str, str] = {}
    self._container_started_at: dict[str, float] = {}
    self._container_images: dict[str, str] = {}
    self._container_rw_sizes: dict[str, int] = {}
    self._container_ids: dict[str, str] = {}
    self._container_mounts: dict[str, tuple[tuple[str, str], ...]] = {}
    self._container_env: dict[str, dict[str, str]] = {}
    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

def set_container_state(self, name: str, state: str) -> None:
    """Record *state* (``"running"``, ``"exited"``, ...) for container *name*."""
    self._container_states[name] = state

set_container_started_at(name, timestamp)

Record the start timestamp of container name.

Source code in src/terok_sandbox/runtime/null.py

def set_container_started_at(self, name: str, timestamp: float) -> None:
    """Record the start timestamp of container *name*."""
    self._container_started_at[name] = timestamp

set_container_image(name, image_ref)

Record the image ref behind container name.

Source code in src/terok_sandbox/runtime/null.py

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

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_container_id(name, container_id)

Record the full container ID Container.id returns for name.

Source code in src/terok_sandbox/runtime/null.py

def set_container_id(self, name: str, container_id: str) -> None:
    """Record the full container ID [`Container.id`][terok_sandbox.runtime.Container.id] returns for *name*."""
    self._container_ids[name] = container_id

set_container_env(name, env)

Fixture hook: record the environment name was created with.

Source code in src/terok_sandbox/runtime/null.py

def set_container_env(self, name: str, env: dict[str, str]) -> None:
    """Fixture hook: record the environment *name* was created with."""
    self._container_env[name] = dict(env)

set_container_mounts(name, mounts)

Record the (source, destination) mounts for container name.

Source code in src/terok_sandbox/runtime/null.py

def set_container_mounts(self, name: str, mounts: tuple[tuple[str, str], ...]) -> None:
    """Record the ``(source, destination)`` mounts for container *name*."""
    self._container_mounts[name] = mounts

set_exit_code(name, code)

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

Source code in src/terok_sandbox/runtime/null.py

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

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

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

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

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

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

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

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

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

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

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

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_started_at.pop(name, None)
        self._container_images.pop(name, None)
        self._container_rw_sizes.pop(name, None)
        self._container_ids.pop(name, None)
        self._container_mounts.pop(name, None)
        self._container_env.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

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)