goodwe-addon/goodwe_controller/test_bytecode.py

"""Runnable check for TEST-15 - the mutation harness's own honesty.
`python3 test_bytecode.py`

Every other file in this directory grades the add-on. This one grades the thing
that grades the add-on, because in this repository a mutation run IS the
evidence: "I broke it, the suite went red, I put it back, the suite went green"
is how nearly every safety claim here was established.

THE DEFECT, found by FCST-03's developer and reproduced below. CPython decides a
cached `.pyc` is still good from two numbers: the source's SIZE, and its
MTIME TRUNCATED TO WHOLE SECONDS. Reverting `self.provisional` to
`here.provisional` moves neither - same length, same second - so the interpreter
loaded the bytecode it already had. Confirmed then by disassembly: the source
read `here`, the running code loaded `self`. The mutation was reported as a
survivor when in truth it had never run.

WHICH DIRECTION IT LIES IN. A mutation that never took effect makes the suite
stay GREEN, so it is reported as a survivor: stale bytecode MANUFACTURES
SURVIVORS. The opposite - a revert that never took, leaving a red on correct
source - is possible too but announces itself, because a developer chases an
unexplained failure. That asymmetry is why the coverage-audit table in
test_control.py is a lower bound rather than nonsense, and it is what the survey
in MUTATION.md is scoped by.

WHAT THIS FILE PINS, in the order the card's acceptance criteria name it:

  AC1  The defect reproduces (part 1), the two mitigations that DO NOT fix it
       are shown not to (part 2), the one that does is shown to (part 3), and
       the shipped harness is shown to close it end to end on a tree that would
       otherwise mis-grade (part 4). Executed, not reasoned: every part builds a
       real package, imports it in a real subprocess, and reads back which
       value the running code returned.
  AC2  The guarantee is where a developer writing a mutation reads it: part 5
       requires EVERY test_*.py in this directory to point at MUTATION.md. The
       list is derived from the tree, so a new suite that joins without the
       pointer is red - CI-01's rule, for CI-01's reason.
  AC4  Part 6 pins the two properties that make CI immune, on the workflow file
       itself.

⚠️ NOTHING HERE SLEEPS. One-second mtime resolution is the mechanism; a harness
that waits out the clock would be slower on every run and would still only cover
the edit somebody remembered to slow down. Part 1 REPRODUCES the collision
deliberately (it retries until two writes land in the same second, which takes
milliseconds) rather than avoiding it.

⚠️ This file writes only under a temporary directory. It must stay that way: CI
mounts the checkout READ-ONLY, and a check that needs to write into the tree
would fail there for a reason that has nothing to do with what it grades.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or use
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ. See MUTATION.md.
"""

import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
fails = []


def check(name, cond):
    if cond:
        print(f"  ok   {name}")
    else:
        print(f"  FAIL {name}")
        fails.append(name)


# ---------------------------------------------------------------------------
# The fixture. `who()` returns one of two names that are THE SAME LENGTH, so
# swapping them leaves the file byte count identical - the first half of
# CPython's staleness test. The second half is the whole-second mtime, which
# part 1 arranges by writing both versions inside one second.
#
# It is deliberately not a copy of a shipped module: the property under test is
# CPython's, and using real code here would make this file go red whenever that
# module changed for unrelated reasons.
# ---------------------------------------------------------------------------
V_OLD = 'def who():\n    self = "SELF"\n    here = "HERE"\n    return self\n'
V_NEW = 'def who():\n    self = "SELF"\n    here = "HERE"\n    return here\n'
assert len(V_OLD) == len(V_NEW), "the fixture must be size-preserving or it grades nothing"

PROBE = "from app.m import who\nprint(who())\n"
# The harness runs `test_*.py` files and grades them by exit status, so the
# probe wears that shape in part 4: it FAILS when it sees the pre-mutation value.
TEST_PROBE = (
    "from app.m import who\n"
    "v = who()\n"
    "print('RUNS:', v)\n"
    "raise SystemExit(0 if v == 'HERE' else 1)\n"
)


def run(tree, args, env_extra=None, script="probe.py"):
    env = dict(os.environ)
    env.pop("PYTHONDONTWRITEBYTECODE", None)
    if env_extra:
        env.update(env_extra)
    out = subprocess.run(
        [sys.executable] + args + [script],
        cwd=tree, env=env, capture_output=True, text=True,
    )
    return out.stdout.strip()


def caches(tree):
    return sorted(str(p.relative_to(tree)) for p in Path(tree).rglob("__pycache__"))


def seed(tree):
    """Leave `tree` holding V_NEW on disk and V_OLD in its bytecode cache.

    Returns True when the collision was actually arranged. Both writes must land
    in the same whole second; at a few milliseconds per cycle that is the normal
    case, and the caller retries rather than asserting on one attempt, because a
    machine that happens to cross a second boundary mid-cycle would otherwise
    report a defect in CPython instead of a slow disk.
    """
    app = Path(tree) / "app"
    shutil.rmtree(app, ignore_errors=True)
    app.mkdir(parents=True)
    (app / "__init__.py").write_text("")
    (Path(tree) / "probe.py").write_text(PROBE)
    (Path(tree) / "test_probe.py").write_text(TEST_PROBE)

    m = app / "m.py"
    m.write_text(V_OLD)
    t0 = m.stat().st_mtime
    run(tree, [])                      # compiles V_OLD, writes app/__pycache__
    m.write_text(V_NEW)                # the "revert": same size, same second
    t1 = m.stat().st_mtime
    return int(t0) == int(t1) and caches(tree) != []


def seeded_tree(stack):
    tree = tempfile.mkdtemp(prefix="test_bytecode.")
    stack.append(tree)
    for _ in range(40):
        if seed(tree):
            return tree
    return None


tmpdirs = []
print("stale bytecode (TEST-15)")

# ---------------------------------------------------------------------------
print("\n1. the defect: a same-size edit inside one second is not seen at all")
# ---------------------------------------------------------------------------
tree = seeded_tree(tmpdirs)
check("the collision can be arranged at all (same size, same second)", tree is not None)
if tree is None:
    print("  cannot continue - the fixture never landed two writes in one second")
    raise SystemExit(1)

src = (Path(tree) / "app" / "m.py").read_text()
check("the source on disk says HERE", "return here" in src and "return self" not in src)
check("...and a plain interpreter runs SELF anyway", run(tree, []) == "SELF")
# Stated as its own check because this is the whole ticket in one line: what the
# file says and what the process executes have come apart, with no error, no
# warning, and a zero exit status.

# ---------------------------------------------------------------------------
print("\n2. the two mitigations that do NOT close it")
# ---------------------------------------------------------------------------
# ⚠️ READ THIS BEFORE 'FIXING' A HARNESS WITH -B ALONE. Both of these stop
# bytecode being WRITTEN. Neither stops a stale one being READ, and a mutation
# run always has one already: the previous run wrote it. The advice in
# test_control.py's coverage audit said `-B` first and CLEAR the cache second;
# the order is not a style choice, the clear is the half that works.
check("`python3 -B` over an existing cache still runs SELF", run(tree, ["-B"]) == "SELF")
check(
    "PYTHONDONTWRITEBYTECODE=1 over an existing cache still runs SELF",
    run(tree, [], {"PYTHONDONTWRITEBYTECODE": "1"}) == "SELF",
)

# ---------------------------------------------------------------------------
print("\n3. the purge closes it, and -B keeps the next run clean")
# ---------------------------------------------------------------------------
for p in Path(tree).rglob("__pycache__"):
    shutil.rmtree(p, ignore_errors=True)
check("after the purge, the plain interpreter runs HERE", run(tree, []) == "HERE")
check("...and it has written a fresh cache back", caches(tree) != [])

tree2 = seeded_tree(tmpdirs)
check("second fixture seeded", tree2 is not None)
if tree2:
    for p in Path(tree2).rglob("__pycache__"):
        shutil.rmtree(p, ignore_errors=True)
    check("purge + -B runs HERE", run(tree2, ["-B"]) == "HERE")
    check("...and leaves NO cache for the next mutation to inherit", caches(tree2) == [])
    # The second half is why the harness passes -B at all. Purging before a run
    # protects THIS run; not writing one protects the NEXT edit, which is the
    # one nobody is thinking about when they are three mutations into a batch.

# ---------------------------------------------------------------------------
print("\n4. the shipped harness, executed against a tree that would mis-grade")
# ---------------------------------------------------------------------------
harness = HERE / "tools" / "mutation_run.sh"
check("tools/mutation_run.sh exists", harness.is_file())
check("...and is executable", os.access(harness, os.X_OK))

tree3 = seeded_tree(tmpdirs)
check("third fixture seeded", tree3 is not None)
if tree3 and harness.is_file():
    # NON-VACUITY FIRST. If the naked run below were already green, the harness
    # run after it would prove nothing - it would be passing on a tree that
    # needed no help. This is the same trap test_peakguard.py part 7 documents:
    # a mutation harness that patches nothing passes beautifully.
    naked = subprocess.run(
        [sys.executable, "test_probe.py"], cwd=tree3, capture_output=True, text=True
    )
    check("without the harness the tree grades the STALE source (rc!=0)", naked.returncode != 0)
    check("...and says so: it saw SELF", "SELF" in naked.stdout)

    got = subprocess.run(
        ["sh", str(harness), tree3], cwd=tree3, capture_output=True, text=True
    )
    check("the harness grades the source on disk (rc==0)", got.returncode == 0)
    check("...having reported the purge", "no bytecode cache" in got.stdout)
    check("...and it leaves no cache behind", caches(tree3) == [])
    if got.returncode != 0:
        print(got.stdout[-2000:])
        print(got.stderr[-2000:])

    # The harness must also refuse to report on nothing - CI-01's zero-file rule.
    empty = tempfile.mkdtemp(prefix="test_bytecode.empty.")
    tmpdirs.append(empty)
    none = subprocess.run(
        ["sh", str(harness), empty], cwd=empty, capture_output=True, text=True
    )
    check("the harness refuses a tree with no test_*.py", none.returncode != 0)

# ---------------------------------------------------------------------------
print("\n5. the guarantee is stated where a mutation gets written")
# ---------------------------------------------------------------------------
# ⚠️ THE LIST IS DERIVED FROM THE TREE. A hand-written one is how a new suite
# joins this directory carrying no warning at all, which is the exact shape of
# failure CI-01's file glob exists to prevent. If you add a test_*.py, put the
# pointer in its docstring; that is the whole cost.
doc = HERE / "MUTATION.md"
check("MUTATION.md exists", doc.is_file())
if doc.is_file():
    text = doc.read_text()
    check("...and it names the purge as the load-bearing half", "__pycache__" in text and "purge" in text.lower())
    check("...and warns that -B alone does not read fresh", "-B" in text)
    check("...and points at the harness", "mutation_run.sh" in text)

suites = sorted(p.name for p in HERE.glob("test_*.py"))
check("there are test suites to survey at all", len(suites) >= 15)
missing = [n for n in suites if "MUTATION.md" not in (HERE / n).read_text()]
check(f"every one of the {len(suites)} suites points at MUTATION.md", missing == [])
if missing:
    print("       missing the pointer:", ", ".join(missing))

# ---------------------------------------------------------------------------
print("\n6. CI cannot carry a stale artefact between jobs")
# ---------------------------------------------------------------------------
# ⚠️ THE FIRST TWO PROPERTIES ARE THE ONES THAT ACTUALLY DO THE WORK, AND THEY
# PREDATE THIS CARD. Measured for TEST-15, not assumed: with the checkout bind
# mounted READ-ONLY the interpreter writes no __pycache__ into the tree at all,
# and it does so SILENTLY - no error, no warning, exit status 0. The workflow
# also `rm -rf`s its workspace and clones afresh, so nothing survives from a
# previous job even on the same runner. PYTHONDONTWRITEBYTECODE=1 is belt and
# braces on top: it is what keeps this true if somebody ever needs to drop the
# `:ro`, and dropping the `:ro` is a one-character edit that nothing else here
# would notice.
wf = HERE.parent / ".gitea" / "workflows" / "checks.yml"
check("the workflow file is where this expects it", wf.is_file())
if wf.is_file():
    y = wf.read_text()
    # ⚠️ GRADED ON THE COMMAND LINE, NOT ON THE FILE. A `in y` search over the
    # whole workflow passes on the PROSE around the step - this file's first
    # draft did exactly that, and stayed green after the `-e` flag and the `:ro`
    # were deleted, because the comment explaining them still named both.
    runs = [l for l in y.splitlines() if "docker run" in l and ":/w" in l]
    check("exactly one docker run mounts the checkout", len(runs) == 1)
    line = runs[0] if runs else ""
    check("the checkout is mounted read-only", ':/w:ro"' in line)
    check("PYTHONDONTWRITEBYTECODE=1 is set on that run", "-e PYTHONDONTWRITEBYTECODE=1" in line)
    check("the workspace is removed before each checkout", 'rm -rf "$WS"' in y)
    check("no sleep was introduced to dodge the mtime resolution", "sleep " not in y)

for d in tmpdirs:
    shutil.rmtree(d, ignore_errors=True)

print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")