117 lines
3.7 KiB
Python
117 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
|