Testing Module

testing

Public test helpers for functualize job testing.

Provides test doubles and builders for unit testing jobs:

from functualize.testing import TestRunContext, CapturingLog, MockInvoke, AutoPrompt, NoopPerf
from functualize.testing import FakeShell, FakeStdout

__all__ = ['AutoPrompt', 'CapturingLog', 'FakeShell', 'FakeShellCall', 'FakeStdout', 'MockInvoke', 'NoopPerf', 'TestRunContext'] module-attribute

TestRunContext

Builder for test RunContext instances with sensible defaults.

Example::

from functualize.testing import TestRunContext

rc = TestRunContext.create()
# Use rc in your job function under test

# Or with overrides:
rc = TestRunContext.create(log=my_custom_log, state=my_state)

rc.log(...) emits through the same Log the job would receive as an injected parameter, so messages logged either way are recorded by the default CapturingLog and can be asserted through captured_logs().

create(*, log=None, invoke=None, prompt=None, perf=None, state=None, job_context=None) staticmethod

Create a RunContext for testing with optional capability overrides.

Each omitted parameter uses a default test double: - log: CapturingLog (records all calls) - invoke: MockInvoke({}) (no configured results) - prompt: AutoPrompt([]) (no pre-configured answers) - perf: NoopPerf (silently accepts all calls) - state: empty State instance - job_context: JobContext(name="test", trace_id=None, metadata=empty)

Parameters:

Name	Type	Description	Default
log	Log | None	Override for the Log capability.	None
invoke	Invoke | None	Override for the Invoke capability.	None
prompt	Prompt | None	Override for the Prompt capability.	None
perf	Perf | None	Override for the Perf capability.	None
state	State | None	Override for the State capability.	None
job_context	JobContext | None	Override for the JobContext capability.	None

Returns:

Type	Description
RunContext	A fully-constructed RunContext backed by a DIRegistry containing
RunContext	the resolved test doubles.

Source code in src/functualize/testing/builder.py

@staticmethod
def create(
    *,
    log: Log | None = None,
    invoke: Invoke | None = None,
    prompt: Prompt | None = None,
    perf: Perf | None = None,
    state: State | None = None,
    job_context: JobContext | None = None,
) -> RunContext:
    """Create a RunContext for testing with optional capability overrides.

    Each omitted parameter uses a default test double:
    - log: CapturingLog (records all calls)
    - invoke: MockInvoke({}) (no configured results)
    - prompt: AutoPrompt([]) (no pre-configured answers)
    - perf: NoopPerf (silently accepts all calls)
    - state: empty State instance
    - job_context: JobContext(name="test", trace_id=None, metadata=empty)

    Args:
        log: Override for the Log capability.
        invoke: Override for the Invoke capability.
        prompt: Override for the Prompt capability.
        perf: Override for the Perf capability.
        state: Override for the State capability.
        job_context: Override for the JobContext capability.

    Returns:
        A fully-constructed RunContext backed by a DIRegistry containing
        the resolved test doubles.
    """
    # Apply defaults for any omitted capabilities
    effective_log = log if log is not None else CapturingLog()
    effective_invoke = invoke if invoke is not None else MockInvoke({})
    effective_prompt = prompt if prompt is not None else AutoPrompt([])
    effective_perf = perf if perf is not None else NoopPerf()
    effective_state = state if state is not None else _temp_state()
    effective_job_context = (
        job_context
        if job_context is not None
        else JobContext(
            name="test",
            trace_id=None,
        )
    )

    # Build a DIRegistry with test doubles
    registry = DIRegistry()
    registry.provide(Log, effective_log)
    registry.provide(Invoke, effective_invoke)
    registry.provide(Prompt, effective_prompt)
    registry.provide(Perf, effective_perf)
    registry.provide(State, effective_state)
    registry.provide(JobContext, effective_job_context)

    # Create a minimal config mock that satisfies the RunContext constructor
    mock_config: Any = MagicMock()
    mock_config.set_prefix = MagicMock()

    # Create a silent logger for tests
    test_logger = logging.getLogger("functualize.test")

    # Construct the RunContext with the DI registry
    rc = RunContext(
        name=effective_job_context.name,
        config=mock_config,
        logger=test_logger,
        _di_registry=registry,
        # The per-invocation capability map the engine would hand a real
        # RunContext. rc.log() reads its Log out of here, so the double
        # sees rc.log(...) exactly as the injected `log: Log` parameter
        # would in production.
        _caps={
            Log: effective_log,
            Invoke: effective_invoke,
            Prompt: effective_prompt,
            Perf: effective_perf,
            State: effective_state,
            JobContext: effective_job_context,
        },
    )

    return rc

captured_logs(rc) staticmethod

Retrieve the ordered list of (level, message) tuples from the RunContext's log.

Accesses the CapturingLog instance registered in the RunContext's DI registry and returns its recorded calls. Messages emitted via rc.log(...) are included: the same double sits in the RunContext's per-invocation capability map, which is where RunContext.log() takes its sink from.

Parameters:

Name	Type	Description	Default
rc	RunContext	A RunContext created by TestRunContext.create().	required

Returns:

Type	Description
list[tuple[str, object]]	Ordered list of (level, message) tuples recorded by the CapturingLog.

Raises:

Type	Description
RuntimeError	If the RunContext has no DI registry attached.
TypeError	If the registered Log is not a CapturingLog instance.

Source code in src/functualize/testing/builder.py

@staticmethod
def captured_logs(rc: RunContext) -> list[tuple[str, object]]:
    """Retrieve the ordered list of (level, message) tuples from the RunContext's log.

    Accesses the CapturingLog instance registered in the RunContext's DI registry
    and returns its recorded calls. Messages emitted via ``rc.log(...)`` are
    included: the same double sits in the RunContext's per-invocation
    capability map, which is where ``RunContext.log()`` takes its sink from.

    Args:
        rc: A RunContext created by TestRunContext.create().

    Returns:
        Ordered list of (level, message) tuples recorded by the CapturingLog.

    Raises:
        RuntimeError: If the RunContext has no DI registry attached.
        TypeError: If the registered Log is not a CapturingLog instance.
    """
    log_instance = rc[Log]
    if not isinstance(log_instance, CapturingLog):
        raise TypeError(
            f"Expected CapturingLog in RunContext, got {type(log_instance).__name__}. "
            f"captured_logs() only works with the default CapturingLog test double."
        )
    return log_instance.calls

AutoPrompt(responses=None)

Bases: Prompt

Test double for the Prompt capability with FIFO responses.

Accepts a sequence of responses at construction and returns them one at a time in order. Raises IndexError when exhausted.

Example

prompt = AutoPrompt(["yes", True, "choice_a"]) prompt.text("Name?") # returns "yes" prompt.confirm("Sure?") # returns True prompt.text("Pick one?") # returns "choice_a" prompt.text("Another?") # raises IndexError

Source code in src/functualize/testing/doubles.py

def __init__(self, responses: list[Any] | None = None) -> None:
    super().__init__()
    self._responses: list[Any] = list(responses) if responses else []
    self._index: int = 0

ask(request)

Return the next pre-configured response.

Raises:

Type	Description
IndexError	When exhausted.

Source code in src/functualize/testing/doubles.py

def ask(self, request: Any) -> Any:
    """Return the next pre-configured response.

    Raises:
        IndexError: When exhausted.
    """
    return self._next_response()

confirm(question, *, default=None, **kwargs)

Return the next pre-configured response.

Raises:

Type	Description
IndexError	When exhausted.

Source code in src/functualize/testing/doubles.py

def confirm(
    self, question: str, *, default: bool | None = None, **kwargs: Any
) -> Any:
    """Return the next pre-configured response.

    Raises:
        IndexError: When exhausted.
    """
    return self._next_response()

choice(question, choices, **kwargs)

Return the next pre-configured response.

Raises:

Type	Description
IndexError	When exhausted.

Source code in src/functualize/testing/doubles.py

def choice(self, question: str, choices: list[Any], **kwargs: Any) -> Any:
    """Return the next pre-configured response.

    Raises:
        IndexError: When exhausted.
    """
    return self._next_response()

text(question, *, default=None, **kwargs)

Return the next pre-configured response.

Raises:

Type	Description
IndexError	When exhausted.

Source code in src/functualize/testing/doubles.py

def text(self, question: str, *, default: str | None = None, **kwargs: Any) -> Any:
    """Return the next pre-configured response.

    Raises:
        IndexError: When exhausted.
    """
    return self._next_response()

CapturingLog()

Bases: Log

Test double for the Log capability that records all log calls.

Each call is stored as a (level, message) tuple in insertion order. Supports both the call syntax and named level methods.

Example

log = CapturingLog() log("hello") log.warning("watch out") assert log.calls == [("info", "hello"), ("warning", "watch out")]

Source code in src/functualize/testing/doubles.py

def __init__(self) -> None:
    super().__init__()
    self.calls: list[tuple[str, object]] = []

calls = [] instance-attribute

__call__(message, level='info')

Record a log call as (level, message).

Validates the level exactly as the real Log does, so a bad level fails in tests instead of being recorded and only failing in production.

Raises:

Type	Description
ValueError	If level is not a valid log level.

Source code in src/functualize/testing/doubles.py

def __call__(self, message: object, level: str = "info") -> None:
    """Record a log call as (level, message).

    Validates the level exactly as the real Log does, so a bad level fails
    in tests instead of being recorded and only failing in production.

    Raises:
        ValueError: If level is not a valid log level.
    """
    validate_log_level(level)
    self.calls.append((level, message))

info(msg)

Record an info-level log call.

Source code in src/functualize/testing/doubles.py

def info(self, msg: object) -> None:
    """Record an info-level log call."""
    self(msg, level="info")

warning(msg)

Record a warning-level log call.

Source code in src/functualize/testing/doubles.py

def warning(self, msg: object) -> None:
    """Record a warning-level log call."""
    self(msg, level="warning")

error(msg)

Record an error-level log call.

Source code in src/functualize/testing/doubles.py

def error(self, msg: object) -> None:
    """Record an error-level log call."""
    self(msg, level="error")

debug(msg)

Record a debug-level log call.

Source code in src/functualize/testing/doubles.py

def debug(self, msg: object) -> None:
    """Record a debug-level log call."""
    self(msg, level="debug")

MockInvoke(results=None)

Bases: Invoke

Test double for the Invoke capability with pre-configured results.

Accepts a mapping of job names to result values at construction. Returns the mapped result when invoked; raises KeyError on unknown job names.

Example

invoke = MockInvoke({"deploy": result_obj}) result = invoke("deploy") # returns result_obj invoke("unknown") # raises KeyError

Source code in src/functualize/testing/doubles.py

def __init__(self, results: dict[str, Any] | None = None) -> None:
    self._results: dict[str, Any] = results or {}

__call__(job_or_fn, **kwargs)

Return the pre-configured result for the given job.

Every keyword the real Invoke accepts is absorbed and ignored, so a job that passes config=, timeout=, or a gate option behaves the same under test as it does in production.

Raises:

Type	Description
KeyError	If the job has no configured result.

Source code in src/functualize/testing/doubles.py

def __call__(self, job_or_fn: str | Callable[..., Any], **kwargs: Any) -> Any:
    """Return the pre-configured result for the given job.

    Every keyword the real Invoke accepts is absorbed and ignored, so a job
    that passes ``config=``, ``timeout=``, or a gate option behaves the same
    under test as it does in production.

    Raises:
        KeyError: If the job has no configured result.
    """
    job_name = self._name_of(job_or_fn)
    if job_name not in self._results:
        raise KeyError(
            f"MockInvoke has no configured result for job '{job_name}'. "
            f"Available: {sorted(self._results.keys())}"
        )
    return self._results[job_name]

parallel(jobs, **kwargs)

Return pre-configured results for each job in order.

Raises:

Type	Description
KeyError	If any job has no configured result.

Source code in src/functualize/testing/doubles.py

def parallel(
    self,
    jobs: Sequence[tuple[str | Callable[..., Any], dict[str, Any]]],
    **kwargs: Any,
) -> list[Any]:
    """Return pre-configured results for each job in order.

    Raises:
        KeyError: If any job has no configured result.
    """
    return [self(job_or_fn, **job_kwargs) for job_or_fn, job_kwargs in jobs]

schema(job_or_fn)

Return the pre-configured result for the given job as schema.

For testing purposes, this returns whatever is mapped for the job name.

Raises:

Type	Description
KeyError	If the job has no configured result.

Source code in src/functualize/testing/doubles.py

def schema(self, job_or_fn: str | Callable[..., Any]) -> Any:
    """Return the pre-configured result for the given job as schema.

    For testing purposes, this returns whatever is mapped for the job name.

    Raises:
        KeyError: If the job has no configured result.
    """
    job_name = self._name_of(job_or_fn)
    if job_name not in self._results:
        raise KeyError(
            f"MockInvoke.schema has no configured result for job '{job_name}'. "
            f"Available: {sorted(self._results.keys())}"
        )
    return self._results[job_name]

NoopPerf

Bases: Perf

Test double for the Perf capability that silently accepts all calls.

All mark, mark_start, and mark_end calls are accepted without recording or raising. phases() returns an empty list.

Example

perf = NoopPerf() perf.mark("init") # no-op perf.mark_start("phase_1") # no-op perf.mark_end("phase_1") # no-op perf.phases() # returns []

mark(name)

Accept a mark call silently.

Source code in src/functualize/testing/doubles.py

def mark(self, name: str) -> None:
    """Accept a mark call silently."""

mark_start(name)

Accept a mark_start call silently.

Source code in src/functualize/testing/doubles.py

def mark_start(self, name: str) -> None:
    """Accept a mark_start call silently."""

mark_end(name)

Accept a mark_end call silently.

Source code in src/functualize/testing/doubles.py

def mark_end(self, name: str) -> None:
    """Accept a mark_end call silently."""

phases(include=None, exclude=None)

Return an empty list of phases.

Source code in src/functualize/testing/doubles.py

def phases(
    self, include: str | None = None, exclude: str | None = None
) -> list[Any]:
    """Return an empty list of phases."""
    return []

FakeShell(mapping=None)

A scripted, recording stand-in for the Shell capability.

Parameters:

Name	Type	Description	Default
mapping	dict[Any, ShellResult] | None	Maps an exact command string or a compiled regex to the ShellResult to return. The command's display form is matched against string keys by equality and against regex keys by search.	None

Source code in src/functualize/testing/shell.py

def __init__(self, mapping: dict[Any, ShellResult] | None = None) -> None:
    self._mapping: dict[Any, ShellResult] = dict(mapping or {})
    self.calls: list[FakeShellCall] = []
    self._cd_stack: list[str] = []
    self._prefix_stack: list[list[str]] = []
    self._deferred: list[tuple[list[str] | str, dict[str, Any]]] = []

calls = [] instance-attribute

deferred property

The cleanup commands queued so far, in registration order.

__call__(command, *, check=True, **kwargs)

Resolve, record, and answer a command from the mapping.

Raises:

Type	Description
AssertionError	If no mapping entry matches (loud on unexpected).

Source code in src/functualize/testing/shell.py

def __call__(
    self,
    command: list[str] | str,
    *,
    check: bool = True,
    **kwargs: Any,
) -> ShellResult:
    """Resolve, record, and answer a command from the mapping.

    Raises:
        AssertionError: If no mapping entry matches (loud on unexpected).
    """
    argv, display = self._resolve(command)
    argv, display = self._apply_prefix(argv, display)
    # An active `cd` block shows up the way the real shell applies it: as
    # the call's effective working directory. Recorded rather than folded
    # into the display, so mapping keys stay the command a job wrote.
    cwd = self._effective_cwd(kwargs.get("cwd"))
    if cwd is not None:
        kwargs = {**kwargs, "cwd": cwd}
    self.calls.append(FakeShellCall(argv=argv, command=display, kwargs=kwargs))

    result = self._lookup(display)
    if result is None:
        raise AssertionError(
            f"FakeShell received an unexpected command: {display!r}. "
            f"Known commands: {[self._key_repr(k) for k in self._mapping]}"
        )
    if check and result.returncode != 0:
        raise ShellError(result)
    return result

sudo(command, **kwargs)

Record a sudo-prefixed call and answer it from the mapping.

Mirrors :meth:WiredShell.sudo for testability: the recorded command is sudo <command> (no -S/password machinery — a fake never spawns a real sudo). preserve_env/password/watchers are accepted and ignored. Match sudo <command> in the mapping like any other command.

Source code in src/functualize/testing/shell.py

def sudo(self, command: list[str], **kwargs: Any) -> ShellResult:
    """Record a ``sudo``-prefixed call and answer it from the mapping.

    Mirrors :meth:`WiredShell.sudo` for testability: the recorded command is
    ``sudo <command>`` (no ``-S``/password machinery — a fake never spawns a
    real sudo). ``preserve_env``/``password``/``watchers`` are accepted and
    ignored. Match ``sudo <command>`` in the mapping like any other command.
    """
    if not isinstance(command, (list, tuple)):
        raise ValueError("FakeShell.sudo requires the list command form.")
    for sudo_only in ("preserve_env", "password", "watchers"):
        kwargs.pop(sudo_only, None)
    return self(["sudo", *command], **kwargs)

cd(path)

Record commands in the block as running in path (§B.3).

Nestable, resolving relative to the enclosing one, exactly as WiredShell.cd does. The effective directory lands on each recorded call's kwargs["cwd"], so a test can assert where a command ran without the fake spawning anything.

Source code in src/functualize/testing/shell.py

@contextmanager
def cd(self, path: str) -> Iterator[None]:
    """Record commands in the block as running in ``path`` (§B.3).

    Nestable, resolving relative to the enclosing one, exactly as
    ``WiredShell.cd`` does. The effective directory lands on each recorded
    call's ``kwargs["cwd"]``, so a test can assert *where* a command ran
    without the fake spawning anything.
    """
    self._cd_stack.append(str(path))
    try:
        yield
    finally:
        self._cd_stack.pop()

prefix(command)

Prepend command to every command in the block (§B.3).

The prefix is applied before matching, because the real shell runs the prefixed argv — so a mapping keyed on poetry run pytest is what matches inside with sh.prefix(["poetry", "run"]). Folding it in afterwards would let a test pass against a command that never ran.

Source code in src/functualize/testing/shell.py

@contextmanager
def prefix(self, command: list[str] | str) -> Iterator[None]:
    """Prepend ``command`` to every command in the block (§B.3).

    The prefix is applied **before** matching, because the real shell runs
    the prefixed argv — so a mapping keyed on ``poetry run pytest`` is what
    matches inside ``with sh.prefix(["poetry", "run"])``. Folding it in
    afterwards would let a test pass against a command that never ran.
    """
    tokens = (
        list(command)
        if isinstance(command, (list, tuple))
        else shlex.split(command)
    )
    self._prefix_stack.append([str(t) for t in tokens])
    try:
        yield
    finally:
        self._prefix_stack.pop()

defer(command, **kwargs)

Queue a cleanup command, as WiredShell.defer does (§B.5).

Nothing runs until :meth:run_deferred. In a real run the engine calls that on the job-exit unwind; in a test, call it yourself — or assert on :attr:deferred to check what a job registered without running it.

Source code in src/functualize/testing/shell.py

def defer(self, command: list[str] | str, **kwargs: Any) -> None:
    """Queue a cleanup command, as ``WiredShell.defer`` does (§B.5).

    Nothing runs until :meth:`run_deferred`. In a real run the engine calls
    that on the job-exit unwind; in a test, call it yourself — or assert on
    :attr:`deferred` to check *what* a job registered without running it.
    """
    self._deferred.append((command, dict(kwargs)))

run_deferred()

Run and clear the queued cleanups, LIFO.

check=False by default, like the real unwind: a cleanup that fails must not mask the job's own outcome. Unlike the real unwind, an unexpected command still raises — a fake that silently swallowed an unmapped cleanup would be a fake that cannot be asserted on.

Source code in src/functualize/testing/shell.py

def run_deferred(self) -> None:
    """Run and clear the queued cleanups, LIFO.

    ``check=False`` by default, like the real unwind: a cleanup that fails
    must not mask the job's own outcome. Unlike the real unwind, an
    unexpected command still raises — a fake that silently swallowed an
    unmapped cleanup would be a fake that cannot be asserted on.
    """
    while self._deferred:
        command, kwargs = self._deferred.pop()
        kwargs.setdefault("check", False)
        self(command, **kwargs)

FakeShellCall(argv, command, kwargs=dict()) dataclass

A recorded FakeShell invocation.

Attributes:

Name	Type	Description
argv	list[str]	The resolved argument vector (list form, or the split display).
command	str	The display string the command resolved to.
kwargs	dict[str, Any]	The keyword options passed to the call.

argv instance-attribute

command instance-attribute

kwargs = field(default_factory=dict) class-attribute instance-attribute

FakeStdout(output_format='auto')

In-memory Stdout double.

Parameters:

Name	Type	Description	Default
output_format	str	The format emit renders with — "auto" (default, dispatch by value type), "json", "ndjson", "raw", or "none". Mirrors the --emit-format flag so a test can pin the wire shape a caller would get.	'auto'

Source code in src/functualize/testing/stdout.py

def __init__(self, output_format: str = "auto") -> None:
    self._format = output_format or "auto"
    self._buffer = StringIO()
    #: Objects passed to :meth:`emit`, in order.
    self.emitted: list[Any] = []
    #: Raw payloads passed to :meth:`write`, in order.
    self.writes: list[str | bytes] = []

emitted = [] instance-attribute

writes = [] instance-attribute

text property

Everything written so far, as a pipe consumer would see it.

emit(value)

Record value and render it per the configured format.

Source code in src/functualize/testing/stdout.py

def emit(self, value: Any) -> None:
    """Record ``value`` and render it per the configured format."""
    self.emitted.append(value)
    if self._format == "none" or value is None:
        return
    StdoutEmitter(format=self._format, stream=self._buffer).emit(value)

write(data)

Record and buffer a raw passthrough write.

Source code in src/functualize/testing/stdout.py

def write(self, data: str | bytes) -> None:
    """Record and buffer a raw passthrough write."""
    self.writes.append(data)
    if self._format == "none":
        return
    text = data if isinstance(data, str) else bytes(data).decode("utf-8", "replace")
    self._buffer.write(text)

---

Overview

functualize.testing provides test doubles and builders for unit testing jobs without spawning processes, hitting the filesystem, or standing up a live app.

Module location: src/functualize/testing/

Public API

from functualize.testing import (
    TestRunContext,
    CapturingLog,
    MockInvoke,
    AutoPrompt,
    NoopPerf,
    FakeShell,
    FakeShellCall,
    FakeStdout,
)

TestRunContext, CapturingLog, MockInvoke, AutoPrompt, NoopPerf, and FakeShell are covered in the Composing Capabilities and Shell Capability guides. FakeShellCall and FakeStdout are documented here.

---

FakeShellCall

A single recorded FakeShell invocation. FakeShell (the Shell capability's test double) appends one of these to fake.calls for every command it resolves, whether or not the command was actually mapped to a result.

@dataclass(frozen=True)
class FakeShellCall:
    argv: list[str]
    command: str
    kwargs: dict[str, Any] = field(default_factory=dict)

Attribute	Type	Description
argv	list[str]	The resolved argument vector — the list form, or the display string split back apart.
command	str	The display string the command resolved to (what a mapping key matches against).
kwargs	dict[str, Any]	The keyword options passed to the call — includes an effective cwd when the call ran inside a sh.cd(...) block.

from functualize.testing import FakeShell

fake = FakeShell({"git status": ShellResult(0, "clean\n", "", "git status", 12.0, None)})
deploy(sh=fake)

assert fake.calls[0].argv == ["git", "status"]
assert fake.calls[0].command == "git status"

---

FakeStdout

An in-memory double for the Stdout capability (out: Stdout). Records what a job emits so pipeline behavior can be asserted without capturing a real process's stdout.

class FakeStdout:
    def __init__(self, output_format: str = "auto") -> None: ...

    emitted: list[Any]      # objects passed to emit(), in order
    writes: list[str | bytes]  # raw payloads passed to write(), in order

    @property
    def text(self) -> str: ...  # everything written so far, as a pipe consumer would see it

    def emit(self, value: Any) -> None: ...
    def write(self, data: str | bytes) -> None: ...

Member	Type	Description
output_format (constructor arg)	str	The format emit renders with — "auto" (default, dispatch by value type), "json", "ndjson", "raw", or "none". Mirrors the --emit-format flag so a test can pin the wire shape a caller would get.
emitted	list[Any]	The objects handed to emit(), in order — assert on data, not on formatting.
writes	list[str \| bytes]	Raw write() payloads, in order.
text (property)	str	The rendered stream exactly as a pipe would see it — assert on the wire format.

from functualize.testing import FakeStdout

fake = FakeStdout()
run_job(export, out=fake)

assert fake.emitted == [{"id": 1}, {"id": 2}]
assert fake.text.splitlines() == ['{"id":1}', '{"id":2}']