Freeze returncode, Streams, and TimeoutExpired Before One Wrapper Extract

Extracting a command wrapper without a result snapshot breaks callers. Characterization tests must freeze returncode, streams, and timeout first. The smallest safe change starts after that freeze.

Callers depend on exit codes and stream types together. They also depend on TimeoutExpired instead of CalledProcessError. A rewrite that simplifies subprocess.run will break those callers.

Why success tests fail this refactor

A green happy-path test proves almost nothing useful here. It ignores stderr text, non-zero codes, and encoding. It also ignores timeout class names after the extract.

Consider a wrapper used by a local build script. The wrapper mixes check, text, and shell flags. One extract without pins will change exception types.

What the snapshot must record

Record these fields for every fixture command below.

1. argv or shell string actually passed through
2. returncode as an integer, including zero
3. stdout type, bytes or str, plus text
4. stderr type, bytes or str, plus text
5. exception class name, or a JSON null
6. timeout value in seconds, or null
7. text flag, encoding name, and check flag
8. env keys the wrapper actually reads

Skip cwd in this harness on purpose today. Path drift is a different failure class entirely. This article freezes process results and exception names only.

Artifact: a frozen result record

The artifact is a JSON record plus a pytest reader. The record is the source of truth here. The extract may proceed only after tests stay green.

Use this proposed layout in a local repo. Rename files to match the messy module. Treat the layout as an unexecuted example.

tests/characterization/subprocess_records/
  python_ok.json
  python_exit_2.json
  python_timeout.json
tests/characterization/test_run_job_record.py
tools/record_run_job.py

Label this as an unexecuted local example only. Adapt names to the messy module under change.

Step 1: isolate one wrapper

Do not start with a package-wide cleanup pass. Pick one function that still calls subprocess.run. Copy its signature into the recorder notes.

Leave production code untouched in this first step. The next block is a proposed messy target.

# jobs/run_job.py  (existing messy module; proposed example)
import os
import subprocess


def run_job(cmd, timeout=None, check=False):
    extra = os.environ.get("JOB_OPTS", "")
    if extra:
        cmd = f"{cmd} {extra}"
    return subprocess.run(
        cmd,
        shell=True,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=check,
    )

This wrapper concatenates env into a shell string. That behavior is a contract, not a style issue. Characterization must see JOB_OPTS behavior before any extract.

Step 2: record results, do not rewrite

Write a recorder that calls the live wrapper. Persist JSON files next to the characterization tests. Commit the JSON before any helper extract.

The next block is a proposed recorder script.

# tools/record_run_job.py  (unexecuted example)
from __future__ import annotations

import json
import os
from pathlib import Path

from jobs.run_job import run_job

OUT = Path("tests/characterization/subprocess_records")


def dump(name: str, payload: dict) -> None:
    OUT.mkdir(parents=True, exist_ok=True)
    path = OUT / f"{name}.json"
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")


def record_ok() -> None:
    completed = run_job("python -c 'print(\"ok\")'")
    dump(
        "python_ok",
        {
            "returncode": completed.returncode,
            "stdout": completed.stdout,
            "stderr": completed.stderr,
            "stdout_type": type(completed.stdout).__name__,
            "stderr_type": type(completed.stderr).__name__,
            "exception": None,
            "check": False,
            "text": True,
        },
    )


def record_exit_2() -> None:
    completed = run_job(
        "python -c 'import sys; sys.stderr.write(\"w\"); sys.exit(2)'"
    )
    dump(
        "python_exit_2",
        {
            "returncode": completed.returncode,
            "stdout": completed.stdout,
            "stderr": completed.stderr,
            "stdout_type": type(completed.stdout).__name__,
            "stderr_type": type(completed.stderr).__name__,
            "exception": None,
            "check": False,
        },
    )


def record_timeout() -> None:
    try:
        run_job("python -c 'import time; time.sleep(5)'", timeout=0.2)
        raise AssertionError("timeout did not fire")
    except Exception as exc:
        dump(
            "python_timeout",
            {
                "exception": type(exc).__name__,
                "timeout": 0.2,
                "returncode": None,
                "stdout": getattr(exc, "stdout", None),
                "stderr": getattr(exc, "stderr", None),
            },
        )


if __name__ == "__main__":
    os.environ.pop("JOB_OPTS", None)
    record_ok()
    record_exit_2()
    record_timeout()

Run the recorder once on a quiet machine.

python tools/record_run_job.py

Inspect the JSON before writing any pytest assertions. Confirm stdout is str, not bytes, here. Confirm a non-zero exit does not raise.

Confirm the timeout record stores TimeoutExpired by name. Confirm stderr text from the exit-two fixture. Do not start the extract when any field is wrong.

Step 3: turn records into failing-closed tests

A record without a test is only a note. Load JSON in pytest and compare live output. Fail on type drift, not only on text drift.

# tests/characterization/test_run_job_record.py  (unexecuted example)
from __future__ import annotations

import json
import subprocess
from pathlib import Path

import pytest

from jobs.run_job import run_job

REC = Path("tests/characterization/subprocess_records")


def load(name: str) -> dict:
    return json.loads((REC / f"{name}.json").read_text())


def test_python_ok_matches_record():
    rec = load("python_ok")
    completed = run_job("python -c 'print(\"ok\")'")
    assert completed.returncode == rec["returncode"]
    assert completed.stdout == rec["stdout"]
    assert type(completed.stdout).__name__ == rec["stdout_type"]
    assert completed.stderr == rec["stderr"]


def test_exit_2_does_not_raise():
    rec = load("python_exit_2")
    completed = run_job(
        "python -c 'import sys; sys.stderr.write(\"w\"); sys.exit(2)'"
    )
    assert completed.returncode == rec["returncode"] == 2
    assert completed.stderr == rec["stderr"]
    assert type(completed.stderr).__name__ == rec["stderr_type"]


def test_timeout_class_is_stable():
    rec = load("python_timeout")
    exc_type = getattr(subprocess, rec["exception"])
    with pytest.raises(exc_type) as info:
        run_job("python -c 'import time; time.sleep(5)'", timeout=0.2)
    assert type(info.value).__name__ == rec["exception"]
    assert info.value.timeout == rec["timeout"]

Run only this characterization file at first.

pytest tests/characterization/test_run_job_record.py -q

Do not collect the whole suite yet. A broad collection can hide import side effects. Keep the pin local, cheap, and readable.

Step 4: add a JOB_OPTS matrix

The wrapper reads JOB_OPTS and mutates the command. That mutation is part of the observed contract. Record two more fixtures, unset and set.

1. JOB_OPTS unset keeps the original command string intact.
2. JOB_OPTS set to --flag may change stdout or stderr.
3. JOB_OPTS with spaces keeps today's shell splitting behavior.

Do not fix the shell concatenation in this pass. A fix is a later, separately pinned change. Characterization first means today's bugs stay visible.

print('ok') yields ok\n under text=True. A cleanup pass often strips that newline. The record must keep the newline byte-for-byte.

Decision table for the extract

Use this table before touching production code.

Observed pin	Allowed extract	Forbidden extract
text=True, stdout is str	move run() into a helper	flip to bytes mode
check=False, exit 2 returns	keep check default False	add check=True
TimeoutExpired on timeout	pass timeout through	swallow and return None
JOB_OPTS concatenated	keep concat in the wrapper	drop env mutation
shell=True	keep shell=True this pass	switch to an argv list
CalledProcessError when check=True	keep the stdlib class	invent JobError now

The table is the review artifact for this extract. Paste it into the pull request body. Reviewers should reject diffs that violate a row.

If one caller already passes check=True, pin that path too. CalledProcessError still carries returncode, stdout, and stderr. Do not replace that class in the same patch.

Step 5: make the smallest safe change

One change means one behavior-preserving move this pass. Extract an internal _run_shell helper and stop. Keep run_job as the public facade function.

Do not rename flags in the same patch. Do not drop JOB_OPTS in the same patch. Do not flip check defaults in the same patch.

# jobs/run_job.py  (proposed extract only)
import os
import subprocess


def _run_shell(cmd, timeout=None, check=False):
    return subprocess.run(
        cmd,
        shell=True,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=check,
    )


def run_job(cmd, timeout=None, check=False):
    extra = os.environ.get("JOB_OPTS", "")
    if extra:
        cmd = f"{cmd} {extra}"
    return _run_shell(cmd, timeout=timeout, check=check)

Re-run the characterization file after the extract. If JSON mismatches, revert the extract at once. Do not edit JSON to match a new idea.

The record is older than the extract on purpose. Age is what makes the pin a regression gate.

pytest tests/characterization/test_run_job_record.py -q
git add tests/characterization/subprocess_records
git add tests/characterization/test_run_job_record.py jobs/run_job.py
git commit -m "test: pin run_job results before extracting _run_shell"

Step 6: PR checklist

Put records, tests, and extract in one PR. Keep unrelated refactors out of that PR. Quote the decision table in the description.

1. Records and tests are committed before the extract diff.
2. pytest runs only the characterization file in CI logs.
3. Decision table rows still match the production flags.
4. No extra rename, format, or import cleanup lands.

Where a free model can draft tests

Models invent cleaner argv APIs under time pressure. Cleaner is not the same as caller compatible. Generate extra tests against frozen JSON only.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

MonkeyCode offers free model access and a free server option. Those help after the JSON records already exist. A model can draft extra pytest cases from the table. It should not invent new subprocess flags here. Run every draft on the local wrapper first.

Discard any stub that changes exception names. Keep secrets out of JOB_OPTS sample values. Characterization fixtures store command output, not credentials.

Limitations

This harness does not pin scheduling jitter at all. TimeoutExpired at 0.2s can flake on loaded CI. Raise the timeout record if CI is slower.

Do not treat flakes as a reason to swallow errors. Flakes mean the pin needs a wider timeout window. They do not mean the exception class is optional.

The harness does not pin shell metacharacters today. A later argv-list extract needs a new record set. Bytes versus text is pinned; locale is not.

PYTHONIOENCODING can still shift captured stdout text today. JSON will not store raw NUL bytes in stdout. Binary tools need a different snapshot format instead.

Use base64 fields for those binary tools. This article does not cover that format.

Who should not use this approach

Skip this if the wrapper is already a thin argv helper. Skip this if no caller reads stderr or returncode. Skip this if the command is not a process.

Skip this if policy forbids recording command output. Do not use these tests as a threat review. shell=True remains a risk after a clean extract.

This workflow preserves behavior, including unsafe shell behavior. It does not make shell=True safe for untrusted input.

Step 7: retire a record on purpose

Retire a JSON record only with an explicit test delete. Do not leave orphan fixtures after a contract change. A deleted pin must appear in the PR text.

Freeze returncode, stream types, and timeout class first. Extract one private runner second, then stop. Leave JOB_OPTS and check defaults untouched this pass.

The characterization file is the merge gate here. No extract should land without a green pin. Keep the JSON records in the same PR.