tegwick
06bd08cfa5
Some checks failed
Governed runtime contract / contract (push) Failing after 10s
Executes activity-core assembly-checkpoint-* runs: fast-forward the assembly-economics checkout, run its deterministic checkpoints/run_checkpoint.py (data via web-fetcher, optional llm-connect assessment), commit only checkpoints/, push fast-forward (rebase once, never force; publish grant limited to checkpoints/), post assembly_checkpoint. Out-of-window fires are no-ops (checkpoints/schedule.md is authoritative). Selected by label 'assembly-checkpoint'. Founder-approved 2026-10-12. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 1494246@bnt-lap001 Assistant-Session: caf6de6f-f9e4-47ce-a584-49baa271fa4b
86 lines
3.9 KiB
Python
86 lines
3.9 KiB
Python
"""Tests for the assembly-checkpoint approach (offline; local bare origin)."""
|
|
|
|
import subprocess
|
|
import textwrap
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rein_aharness import approaches as ap
|
|
from rein_aharness.assembly_checkpoint import kind_from_title, run_assembly_checkpoint
|
|
|
|
RUNNER = textwrap.dedent('''
|
|
import json, sys, pathlib
|
|
kind = sys.argv[sys.argv.index("--kind") + 1]
|
|
mode = pathlib.Path("checkpoints/MODE").read_text().strip()
|
|
if mode == "out":
|
|
print(json.dumps({"status": "out-of-window", "date": "2026-10-12", "kind": kind})); sys.exit(3)
|
|
if mode == "outside":
|
|
pathlib.Path("README.md").write_text("tampered")
|
|
p = pathlib.Path("checkpoints/2026/2026-10-14-" + kind + ".md")
|
|
p.parent.mkdir(parents=True, exist_ok=True); p.write_text("# record")
|
|
print(json.dumps({"status": "ok", "date": "2026-10-14", "kind": kind, "path": str(p), "ok": 5,
|
|
"not_found": 4, "error": 0, "assessment": False, "phase": "setup"}))
|
|
''')
|
|
|
|
|
|
def _git(cwd, *args):
|
|
subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(tmp_path: Path) -> Path:
|
|
origin = tmp_path / "origin.git"
|
|
_git(tmp_path, "init", "-q", "--bare", "-b", "main", str(origin))
|
|
work = tmp_path / "work"
|
|
_git(tmp_path, "clone", "-q", str(origin), str(work))
|
|
_git(work, "config", "user.email", "t@t"), _git(work, "config", "user.name", "t")
|
|
(work / "checkpoints").mkdir()
|
|
(work / "checkpoints" / "run_checkpoint.py").write_text(RUNNER)
|
|
(work / "checkpoints" / "MODE").write_text("in")
|
|
(work / "README.md").write_text("x")
|
|
_git(work, "add", "-A"), _git(work, "commit", "-qm", "init"), _git(work, "push", "-q", "origin", "HEAD:main")
|
|
return work
|
|
|
|
|
|
def test_kind_from_title():
|
|
assert kind_from_title("Assembly Economics reality checkpoint (full) — steady phase") == "full"
|
|
assert kind_from_title("anything") == "light"
|
|
|
|
|
|
def test_in_window_commits_and_pushes(repo):
|
|
r = run_assembly_checkpoint(repo, kind="light", report_to_hub=False)
|
|
assert r.ok and r.committed and r.pushed and r.path.endswith("2026-10-14-light.md")
|
|
origin_head = subprocess.run(["git", "-C", str(repo), "rev-parse", "origin/main"],
|
|
capture_output=True, text=True).stdout.strip()
|
|
assert origin_head == r.origin_sha
|
|
|
|
|
|
def test_out_of_window_is_noop(repo):
|
|
(repo / "checkpoints" / "MODE").write_text("out")
|
|
_git(repo, "commit", "-qam", "mode out"), _git(repo, "push", "-q", "origin", "HEAD:main")
|
|
r = run_assembly_checkpoint(repo, kind="light", report_to_hub=False)
|
|
assert r.ok and r.out_of_window and not r.committed and not r.pushed
|
|
|
|
|
|
def test_refuses_paths_outside_checkpoints(repo):
|
|
(repo / "checkpoints" / "MODE").write_text("outside")
|
|
_git(repo, "commit", "-qam", "mode outside"), _git(repo, "push", "-q", "origin", "HEAD:main")
|
|
r = run_assembly_checkpoint(repo, kind="light", report_to_hub=False)
|
|
assert r.ok # README change is not staged: only checkpoints/ is added
|
|
changed = subprocess.run(["git", "-C", str(repo), "show", "--name-only", "--format=", "HEAD"],
|
|
capture_output=True, text=True).stdout.split()
|
|
assert all(p.startswith("checkpoints/") for p in changed)
|
|
|
|
|
|
def test_label_selects_assembly_checkpoint():
|
|
from rein_aharness.ops_run_client import OpsRun
|
|
import inspect
|
|
kw = {p: None for p in inspect.signature(OpsRun).parameters}
|
|
for k, p in inspect.signature(OpsRun).parameters.items():
|
|
if p.default is not inspect.Parameter.empty:
|
|
kw[k] = p.default
|
|
kw.update(id="x", title="Assembly Economics reality checkpoint (light) — setup phase", description="d",
|
|
labels=["assembly-checkpoint", "research", "automated"], target_repo="assembly-economics",
|
|
activity_definition_id="assembly-checkpoint-setup-daily")
|
|
assert ap.select_approach(OpsRun(**kw)) == ap.APPROACH_ASSEMBLY_CHECKPOINT
|