test_patch.py

#!/usr/bin/env python3
"""Self-check for the patcher. Run: python3 test_patch.py

Uses a synthetic fixture rather than a real opencode binary, so it runs
anywhere. The fixture reproduces the exact byte shape the bundler emits.
"""

import importlib.machinery
import importlib.util
import json
import os
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
spec = importlib.util.spec_from_loader(
    "patcher", importlib.machinery.SourceFileLoader("patcher", os.path.join(HERE, "opencode-retry-patch"))
)
patcher = importlib.util.module_from_spec(spec)
spec.loader.exec_module(patcher)

# Byte-for-byte the shape found in opencode 1.18.21, with surrounding context.
PRISTINE = (
    b'var ih="Free usage exceeded, subscribe to Go",hh="https://opencode.ai/go",'
    b"ah=2000,rh=2,th=0.25,dh=30000,yh=2147483647,ch=5,"
    b"Jd=[/429|500|502|503|504|524/i,/rate limit/i];function cl(e){return Math.min(e,yh)}"
)
# Same shape, different minified names, to prove the matcher is not name-bound.
RENAMED = PRISTINE.replace(b"ah=", b"Q3=").replace(b"rh=", b"W8=").replace(b"th=", b"Zz=") \
                  .replace(b"dh=", b"Kp=").replace(b"yh=", b"Lm=").replace(b"ch=", b"Rv=")


class Opts:
    delay, factor, jitter, cap, retries = 1500, 1.15, 0.2, 5000, 20


# Executable *and* carrying the pattern, so the whole flow can be exercised:
# the script reads a version from it, patches it, and checks it still runs.
def fake_opencode(path, version):
    with open(path, "wb") as fh:
        fh.write(b"#!/bin/sh\necho " + version.encode() + b"\n# " + PRISTINE + b"\n")
    os.chmod(path, 0o755)


def check(condition, label):
    print(("  ok   " if condition else "  FAIL ") + label)
    if not condition:
        sys.exit(1)


def patch_region(data):
    start, end, names, values = patcher.find(data)
    region, _ = patcher.build(names, Opts, end - start)
    return data[:start] + region + data[end:], names, values, region


def main():
    print("js_number")
    check(patcher.js_number(2000) == "2e3", "2000 -> 2e3")
    check(patcher.js_number(250) == "250", "250 -> 250 (25e1 is not shorter)")
    check(patcher.js_number(5) == "5", "5 -> 5")
    check(patcher.js_number(0.2) == ".2", "0.2 -> .2")
    check(patcher.js_number(1) == "1", "1 -> 1")

    print("matching")
    start, end, names, values = patcher.find(PRISTINE)
    check(names == ["ah", "rh", "th", "dh", "yh", "ch"], "reads the minified names")
    check(values == ["2000", "2", "0.25", "30000", "2147483647", "5"], "reads the upstream values")
    check(PRISTINE[start:end] == b"ah=2000,rh=2,th=0.25,dh=30000,yh=2147483647,ch=5", "spans only the constants")

    _, renamed_names, renamed_values, _ = patch_region(RENAMED)
    check(renamed_names == ["Q3", "W8", "Zz", "Kp", "Lm", "Rv"], "survives a minifier rename")
    check(renamed_values == values, "reads the same values after rename")

    print("patching")
    patched, _, _, region = patch_region(PRISTINE)
    check(len(patched) == len(PRISTINE), "total length is preserved")
    check(len(region) == end - start, "region length is preserved")
    check(b"ah=1500,rh=1.15,th=.2,dh=5e3,yh=2147483647,ch=20" in patched, "writes the new values")
    check(b"Jd=[/429|500|502|503|504|524/i" in patched, "leaves the anchor untouched")
    check(patched.endswith(b"return Math.min(e,yh)}"), "leaves trailing code untouched")

    print("idempotence")
    twice, _, _, _ = patch_region(patched)
    check(twice == patched, "patching twice changes nothing")

    print("refusal")
    doubled = PRISTINE + PRISTINE
    try:
        patcher.find(doubled)
        check(False, "refuses an ambiguous match")
    except SystemExit:
        check(True, "refuses an ambiguous match")

    class TooBig(Opts):
        retries = 99999999999999999999
    try:
        patcher.build(names, TooBig, end - start)
        check(False, "refuses values that would grow the region")
    except SystemExit:
        check(True, "refuses values that would grow the region")

    print("platform")
    check(patcher.default_binary("Windows").endswith("opencode.exe"), "windows default is the .exe")
    check(patcher.default_binary("Linux").endswith("opencode"), "linux default has no extension")
    check(patcher.default_binary("Darwin") == patcher.default_binary("Linux"), "macos matches linux")
    for system, marker in (("Linux", 'command opencode "$@"'), ("Windows", "opencode.exe")):
        snippet = patcher.HOOKS["Windows" if system == "Windows" else "posix"].format(script="/x/y")
        check(marker in snippet, f"{system.lower()} hook reaches the real binary")
        check("/x/y" in snippet, f"{system.lower()} hook names the patcher")

    print("cli")
    script = os.path.join(HERE, "opencode-retry-patch")
    run = lambda *a: subprocess.run([sys.executable, script, *a], capture_output=True, text=True)

    with tempfile.TemporaryDirectory() as tmp:
        binary = os.path.join(tmp, "opencode")
        fake_opencode(binary, "1.18.23")
        before = open(binary, "rb").read()

        r = run("--binary", binary, "--dry-run")
        check(r.returncode == 0, "--dry-run exits clean")
        listing = sorted(os.listdir(tmp))
        check(open(binary, "rb").read() == before, "--dry-run leaves the binary alone")
        check(sorted(os.listdir(tmp)) == listing, "--dry-run creates no files")
        check("->" in r.stdout and "max retries" in r.stdout, "--dry-run shows the change")

        r = run("--binary", binary)
        check(r.returncode == 0, "patch exits clean")
        check(not os.path.exists(binary + ".1.18.23.orig"), "no 144MB copy is made to undo 48 bytes")
        check("still starts" in r.stdout, "patched fixture still runs")
        check(b"ch=20" in open(binary, "rb").read(), "patch reached the fixture")

        check(run("--binary", binary).stdout.strip() == "already patched, nothing to do", "second run is a no-op")

        state = json.load(open(binary + ".retrypatch.json"))
        check(state["names"] == ["ah", "rh", "th", "dh", "yh", "ch"], "state records the minified names")
        check(state["region"].startswith("ah=1500"), "state records what was written")
        check(state["pristine"] == "ah=2000,rh=2,th=0.25,dh=30000,yh=2147483647,ch=5",
              "state records the bytes the patch overwrote")
        blob = open(binary, "rb").read()
        check(blob[state["offset"]:state["offset"] + state["width"]] == state["region"].encode(),
              "the recorded offset points at the patched region")
        check(run("--binary", binary, "--quiet").stdout == "", "--quiet says nothing when there is nothing to do")

        # a replaced binary must not be trusted just because a state file exists
        fake_opencode(binary, "1.18.23")
        r = run("--binary", binary, "--quiet")
        check("patched" in r.stdout, "a replaced binary is re-patched despite stale state")
        check(b"ch=20" in open(binary, "rb").read(), "the replacement really got patched")

        inode_before = os.stat(binary).st_ino
        r = run("--binary", binary, "--restore")
        check(r.returncode == 0, "--restore exits clean")
        check(open(binary, "rb").read() == before, "--restore rebuilds the original byte for byte, with no backup file")
        check("recorded original bytes" in r.stdout, "--restore says where it got them")
        check(not os.path.exists(binary + ".retrypatch.json"), "--restore forgets the recorded offset")
        check(os.stat(binary).st_ino != inode_before, "--restore renames in, it does not overwrite in place")
        check(not os.path.exists(binary + ".swap"), "--restore leaves no staging file behind")

    # the bug this replaced: one unversioned backup, restored after an upgrade,
    # silently reinstated the previous version
    with tempfile.TemporaryDirectory() as tmp:
        binary = os.path.join(tmp, "opencode")
        fake_opencode(binary, "1.18.23")
        fake_opencode(binary + ".1.18.21.orig", "1.18.21")
        r = run("--binary", binary, "--restore")
        check(r.returncode != 0, "--restore refuses a backup from another version")
        check("downgrade" in r.stderr, "--restore explains why it refused")
        check(open(binary, "rb").read().find(b"1.18.23") > 0, "the installed version is left alone")

    # upgrading the script on a binary an older release already patched: the
    # factory bytes must come from that release's backup, not from the binary
    with tempfile.TemporaryDirectory() as tmp:
        binary = os.path.join(tmp, "opencode")
        fake_opencode(binary + ".1.18.23.orig", "1.18.23")
        fake_opencode(binary, "1.18.23")
        blob = open(binary, "rb").read()
        st, en, _, _ = patcher.find(blob)
        with open(binary, "wb") as fh:
            fh.write(blob[:st] + b"ah=500,rh=1,th=.2,dh=2e3,yh=2147483647,ch=250" .ljust(en - st) + blob[en:])
        run("--binary", binary)
        state = json.load(open(binary + ".retrypatch.json"))
        check(state["pristine"].startswith("ah=2000"), "factory bytes are taken from the old backup, not the patch")

        # same upgrade, but the old release had already applied these very values
        os.remove(binary + ".retrypatch.json")
        r = run("--binary", binary)
        check(r.returncode == 0, "adopting an already-correct binary exits clean")
        state = json.load(open(binary + ".retrypatch.json"))
        check(state["pristine"].startswith("ah=2000"), "and still records the factory bytes, not the current ones")
        run("--binary", binary, "--restore")
        check(b"ah=2000,rh=2,th=0.25" in open(binary, "rb").read(), "restore really undoes it after that upgrade")

    with tempfile.TemporaryDirectory() as tmp:
        binary = os.path.join(tmp, "opencode")
        fake_opencode(binary, "1.18.23")
        fake_opencode(binary + ".orig", "1.18.23")
        check(run("--binary", binary, "--dry-run").returncode == 0, "--dry-run tolerates a legacy backup")
        check(not os.path.exists(binary + ".1.18.23.orig"), "--dry-run does not migrate it either")
        run("--binary", binary)
        check(os.path.exists(binary + ".1.18.23.orig"), "a legacy unversioned backup is migrated on patch")

    print("\nall checks passed")


if __name__ == "__main__":
    main()