tegwick
d103955217
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""E2E vertical slice: observe → index → command → git evidence → rebuild."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
from repo_manager.commands.task_status import update_task_status
|
||
from repo_manager.gitops import head_sha
|
||
from repo_manager.index_store import default_index_path, load_index
|
||
from repo_manager.observe import observe_repository
|
||
|
||
WORKPLAN = """---
|
||
id: DEMO-WP-0001
|
||
type: workplan
|
||
title: "Demo workplan"
|
||
status: active
|
||
---
|
||
|
||
# Demo workplan
|
||
|
||
## First task
|
||
|
||
```task
|
||
id: DEMO-WP-0001-T01
|
||
status: todo
|
||
priority: high
|
||
```
|
||
|
||
Do the thing.
|
||
"""
|
||
|
||
|
||
def _git(repo: Path, *args: str) -> None:
|
||
subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True)
|
||
|
||
|
||
def _init_fixture(tmp_path: Path) -> Path:
|
||
repo = tmp_path / "demo-repo"
|
||
repo.mkdir()
|
||
_git(repo, "init")
|
||
_git(repo, "config", "user.email", "test@example.com")
|
||
_git(repo, "config", "user.name", "Test")
|
||
(repo / "workplans").mkdir()
|
||
(repo / "workplans" / "DEMO-WP-0001-demo.md").write_text(WORKPLAN, encoding="utf-8")
|
||
(repo / ".repo-classification.yaml").write_text(
|
||
"repo_classification:\n category: experimental\n domain: infotech\n",
|
||
encoding="utf-8",
|
||
)
|
||
(repo / "INTENT.md").write_text("# INTENT\n\nDemo fixture repository.\n", encoding="utf-8")
|
||
_git(repo, "add", ".")
|
||
_git(repo, "commit", "-m", "seed")
|
||
return repo
|
||
|
||
|
||
def test_e2e_vertical_slice(tmp_path: Path):
|
||
repo = _init_fixture(tmp_path)
|
||
before = head_sha(repo)
|
||
assert before
|
||
|
||
# 1–2. Observe + index declared work
|
||
snap, index = observe_repository(repo, slug="demo-repo")
|
||
assert snap["slug"] == "demo-repo"
|
||
assert snap["classification"]["category"] == "experimental"
|
||
assert snap["index"]["workplan_count"] == 1
|
||
assert snap["index"]["task_count"] == 1
|
||
tasks = [r for r in index.work_records if r.kind == "task"]
|
||
assert tasks[0].id == "DEMO-WP-0001-T01"
|
||
assert tasks[0].status == "todo"
|
||
|
||
# 3–4. Apply governed file-backed transition + git evidence
|
||
result = update_task_status(
|
||
repo,
|
||
"DEMO-WP-0001-T01",
|
||
"progress",
|
||
correlation_id="00000000-0000-4000-8000-000000000001",
|
||
reason="e2e proof",
|
||
)
|
||
assert result.status == "applied", result
|
||
assert result.evidence.get("git_sha")
|
||
assert result.evidence["git_sha"] != before
|
||
assert result.evidence["files_touched"] == ["workplans/DEMO-WP-0001-demo.md"]
|
||
|
||
# File content updated
|
||
text = (repo / "workplans" / "DEMO-WP-0001-demo.md").read_text(encoding="utf-8")
|
||
assert "status: progress" in text
|
||
|
||
# 5. Rebuilt projection
|
||
snap2, index2 = observe_repository(repo, slug="demo-repo")
|
||
task2 = next(r for r in index2.work_records if r.id == "DEMO-WP-0001-T01")
|
||
assert task2.status == "progress"
|
||
assert snap2["revision"]["head_sha"] == result.evidence["git_sha"]
|
||
|
||
# Index on disk includes command event (normalized change)
|
||
idx_path = default_index_path(repo)
|
||
assert idx_path.is_file()
|
||
stored = load_index(idx_path)
|
||
types = [e.get("type") for e in stored.events]
|
||
assert "repo.command.applied" in types
|
||
applied = next(e for e in stored.events if e.get("type") == "repo.command.applied")
|
||
assert applied["correlation_id"] == "00000000-0000-4000-8000-000000000001"
|
||
assert applied["git_sha"] == result.evidence["git_sha"]
|
||
|
||
# Proof artifact for humans
|
||
proof = {
|
||
"slice": "RMGR-WP-0001-T05",
|
||
"repo_slug": "demo-repo",
|
||
"before_sha": before,
|
||
"after_sha": result.evidence["git_sha"],
|
||
"command": result.to_dict(),
|
||
"snapshot_after": snap2,
|
||
}
|
||
proof_path = tmp_path / "e2e-proof.json"
|
||
proof_path.write_text(json.dumps(proof, indent=2), encoding="utf-8")
|
||
assert proof_path.stat().st_size > 0
|