AsyncSubprocessPlugin

AsyncSubprocessPlugin

AsyncSubprocessPlugin(verifier)

Bases: StateMachinePlugin

Async subprocess interception plugin.

Replaces asyncio.create_subprocess_exec and asyncio.create_subprocess_shell with fake implementations at activate() time and restores the originals at deactivate() time. Uses reference counting so nested sandboxes work correctly.

States: created -> running -> terminated

Source code in src/bigfoot/plugins/async_subprocess_plugin.py

def __init__(self, verifier: "StrictVerifier") -> None:
    super().__init__(verifier)
    self._spawn_sentinel = _StepSentinel(_SOURCE_SPAWN)
    self._communicate_sentinel = _StepSentinel(_SOURCE_COMMUNICATE)
    self._wait_sentinel = _StepSentinel(_SOURCE_WAIT)

activate

activate()

Reference-counted class-level patch installation.

Source code in src/bigfoot/plugins/async_subprocess_plugin.py

def activate(self) -> None:
    """Reference-counted class-level patch installation."""
    global _bigfoot_create_subprocess_exec, _bigfoot_create_subprocess_shell

    with AsyncSubprocessPlugin._install_lock:
        if AsyncSubprocessPlugin._install_count == 0:
            self._check_conflicts()
            AsyncSubprocessPlugin._original_exec = asyncio.create_subprocess_exec
            AsyncSubprocessPlugin._original_shell = asyncio.create_subprocess_shell

            async def _fake_create_subprocess_exec(
                program: str,
                *args: Any,  # noqa: ANN401
                **kwargs: Any,  # noqa: ANN401
            ) -> _AsyncFakeProcess:
                plugin = _find_async_subprocess_plugin()
                proc = _AsyncFakeProcess()
                proc._plugin = plugin
                plugin._bind_connection(proc)
                command = [program, *[str(a) for a in args]]
                stdin = kwargs.get("stdin")
                plugin._execute_step(
                    plugin._lookup_session(proc), "spawn", (program, *args), kwargs,
                    _SOURCE_SPAWN,
                    details={
                        "command": command,
                        "stdin": stdin if isinstance(stdin, (bytes, type(None))) else None,
                    },
                )
                return proc

            async def _fake_create_subprocess_shell(
                cmd: str,
                **kwargs: Any,  # noqa: ANN401
            ) -> _AsyncFakeProcess:
                plugin = _find_async_subprocess_plugin()
                proc = _AsyncFakeProcess()
                proc._plugin = plugin
                plugin._bind_connection(proc)
                stdin = kwargs.get("stdin")
                plugin._execute_step(
                    plugin._lookup_session(proc), "spawn", (cmd,), kwargs,
                    _SOURCE_SPAWN,
                    details={
                        "command": cmd,
                        "stdin": stdin if isinstance(stdin, (bytes, type(None))) else None,
                    },
                )
                return proc

            _bigfoot_create_subprocess_exec = _fake_create_subprocess_exec
            _bigfoot_create_subprocess_shell = _fake_create_subprocess_shell

            asyncio.create_subprocess_exec = _fake_create_subprocess_exec  # type: ignore[assignment]
            asyncio.create_subprocess_shell = _fake_create_subprocess_shell  # type: ignore[assignment]

        AsyncSubprocessPlugin._install_count += 1

matches

matches(interaction, expected)

Field-by-field comparison with dirty-equals support.

Source code in src/bigfoot/plugins/async_subprocess_plugin.py

def matches(self, interaction: Interaction, expected: dict[str, Any]) -> bool:
    """Field-by-field comparison with dirty-equals support."""
    try:
        for key, expected_val in expected.items():
            actual_val = interaction.details.get(key)
            if expected_val != actual_val:
                return False
        return True
    except Exception:
        return False

assertable_fields

assertable_fields(interaction)

Return assertable fields for each step type.

Source code in src/bigfoot/plugins/async_subprocess_plugin.py

def assertable_fields(self, interaction: Interaction) -> frozenset[str]:
    """Return assertable fields for each step type."""
    if interaction.source_id == _SOURCE_WAIT:
        return frozenset()
    return frozenset(interaction.details.keys())