tegwick
c90f70122c
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
361 lines
12 KiB
Python
361 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from repo_manager.commands import registrar_reconcile as rr
|
|
|
|
|
|
def _git(repo: Path, *args: str) -> None:
|
|
subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True)
|
|
|
|
|
|
def _fixture(tmp_path: Path) -> Path:
|
|
bare = tmp_path / "remote.git"
|
|
repo = tmp_path / "demo"
|
|
_git(tmp_path, "init", "--bare", str(bare))
|
|
_git(tmp_path, "clone", str(bare), str(repo))
|
|
_git(repo, "config", "user.name", "Test")
|
|
_git(repo, "config", "user.email", "test@example.com")
|
|
(repo / "workplans").mkdir()
|
|
(repo / "workplans" / "DEMO-WP-0001.md").write_text(
|
|
"""---
|
|
id: DEMO-WP-0001
|
|
type: workplan
|
|
title: Demo
|
|
status: active
|
|
---
|
|
|
|
## Task
|
|
|
|
```task
|
|
id: DEMO-WP-0001-T01
|
|
status: todo
|
|
priority: high
|
|
```
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
_git(repo, "add", ".")
|
|
_git(repo, "commit", "-m", "seed")
|
|
_git(repo, "push", "-u", "origin", "HEAD")
|
|
return repo
|
|
|
|
|
|
def test_requires_explicit_primary_confirmation(tmp_path: Path) -> None:
|
|
repo = _fixture(tmp_path)
|
|
result = rr.registrar_reconcile(repo)
|
|
assert result.status == "rejected"
|
|
assert result.error and result.error["code"] == "confirmation_required"
|
|
|
|
|
|
def test_unlinked_closed_workplan_is_registrar_work(tmp_path: Path) -> None:
|
|
repo = _fixture(tmp_path)
|
|
workplan = repo / "workplans" / "DEMO-WP-0001.md"
|
|
workplan.write_text(
|
|
workplan.read_text(encoding="utf-8").replace("status: active", "status: finished"),
|
|
encoding="utf-8",
|
|
)
|
|
assert rr._missing_identifiers(repo) == {
|
|
"workplans": ["DEMO-WP-0001"],
|
|
"tasks": ["DEMO-WP-0001-T01"],
|
|
"intakes": [],
|
|
"decisions": [],
|
|
}
|
|
|
|
|
|
def test_linked_closed_workplan_does_not_reopen_historical_task_gaps(tmp_path: Path) -> None:
|
|
repo = _fixture(tmp_path)
|
|
workplan = repo / "workplans" / "DEMO-WP-0001.md"
|
|
text = workplan.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
"status: active\n---",
|
|
'status: finished\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
|
|
)
|
|
workplan.write_text(text, encoding="utf-8")
|
|
assert rr._missing_identifiers(repo) == {
|
|
"workplans": [],
|
|
"tasks": [],
|
|
"intakes": [],
|
|
"decisions": [],
|
|
}
|
|
|
|
|
|
def test_rejects_dirty_or_unsynced_repository(tmp_path: Path, monkeypatch) -> None:
|
|
repo = _fixture(tmp_path)
|
|
monkeypatch.setattr(rr, "_check_primary", lambda _api: ({"status": "ok", "db": "connected"}, None))
|
|
(repo / "note.txt").write_text("dirty", encoding="utf-8")
|
|
dirty = rr.registrar_reconcile(repo, confirm_primary=True)
|
|
assert dirty.status == "rejected"
|
|
assert dirty.error and dirty.error["code"] == "git_precondition_failed"
|
|
|
|
|
|
def test_missing_scan_includes_intakes_and_decisions(tmp_path: Path) -> None:
|
|
repo = _fixture(tmp_path)
|
|
records = repo / "intakes" / "records.md"
|
|
records.parent.mkdir()
|
|
records.write_text(
|
|
"""# Records
|
|
|
|
```yaml
|
|
id: DEMO-IN-0001
|
|
kind: intake
|
|
title: Intake
|
|
status: open
|
|
```
|
|
|
|
```yaml
|
|
id: DEMO-DEC-0001
|
|
kind: decision
|
|
title: Decision
|
|
status: open
|
|
```
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
missing = rr._missing_identifiers(repo)
|
|
assert missing["intakes"] == ["DEMO-IN-0001"]
|
|
assert missing["decisions"] == ["DEMO-DEC-0001"]
|
|
|
|
|
|
def test_scopes_registrar_env_and_commits_assigned_ids(tmp_path: Path, monkeypatch) -> None:
|
|
repo = _fixture(tmp_path)
|
|
monkeypatch.setattr(rr, "_check_primary", lambda _api: ({"status": "ok", "db": "connected"}, None))
|
|
|
|
def fake_run(command, *, env):
|
|
assert command[1:3] == ["fix-consistency", "--path"]
|
|
assert env["STATEHUB_REGISTRAR"] == "1"
|
|
workplan = repo / "workplans" / "DEMO-WP-0001.md"
|
|
text = workplan.read_text(encoding="utf-8")
|
|
text = text.replace("status: active\n---", 'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---')
|
|
text = text.replace("priority: high\n```", 'priority: high\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```')
|
|
workplan.write_text(text, encoding="utf-8")
|
|
return subprocess.CompletedProcess(command, 0, "ok", "")
|
|
|
|
monkeypatch.setattr(rr, "_run_statehub", fake_run)
|
|
|
|
result = rr.registrar_reconcile(
|
|
repo,
|
|
statehub_bin="statehub",
|
|
confirm_primary=True,
|
|
)
|
|
|
|
assert result.status == "applied"
|
|
assert result.evidence["missing_after"] == {
|
|
"workplans": [],
|
|
"tasks": [],
|
|
"intakes": [],
|
|
"decisions": [],
|
|
}
|
|
subject = subprocess.run(
|
|
["git", "log", "-1", "--format=%s"], cwd=repo, capture_output=True, text=True, check=True
|
|
).stdout.strip()
|
|
assert subject == "chore(registrar): assign State Hub identifiers"
|
|
|
|
|
|
def test_accepts_unrelated_assessment_fail_after_exact_requested_verification(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
repo = _fixture(tmp_path)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_check_primary",
|
|
lambda _api: ({"status": "ok", "db": "connected"}, None),
|
|
)
|
|
|
|
def fake_run(command, *, env):
|
|
workplan = repo / "workplans" / "DEMO-WP-0001.md"
|
|
text = workplan.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
"status: active\n---",
|
|
'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
|
|
)
|
|
text = text.replace(
|
|
"priority: high\n```",
|
|
'priority: high\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```',
|
|
)
|
|
workplan.write_text(text, encoding="utf-8")
|
|
return subprocess.CompletedProcess(command, 1, "unrelated C-03 remains", "")
|
|
|
|
monkeypatch.setattr(rr, "_run_statehub", fake_run)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_verify_full_projection",
|
|
lambda _api, workplans, tasks, records: (
|
|
{
|
|
"expected_workplans": len(workplans),
|
|
"expected_tasks": len(tasks),
|
|
"expected_intakes": len(records["intake"]),
|
|
"expected_decisions": len(records["decision"]),
|
|
"missing_workplans": [],
|
|
"missing_tasks": [],
|
|
"missing_intakes": [],
|
|
"missing_decisions": [],
|
|
},
|
|
None,
|
|
),
|
|
)
|
|
|
|
result = rr.registrar_reconcile(
|
|
repo,
|
|
statehub_bin="statehub",
|
|
confirm_primary=True,
|
|
)
|
|
|
|
assert result.status == "applied"
|
|
assert result.evidence["requested_projection_verified"] is True
|
|
assert result.evidence["requested_projection"]["expected_workplans"] == 1
|
|
assert result.evidence["requested_projection"]["expected_tasks"] == 1
|
|
|
|
|
|
def test_repairs_an_already_identified_workplan_projection(tmp_path: Path, monkeypatch) -> None:
|
|
repo = _fixture(tmp_path)
|
|
workplan = repo / "workplans" / "DEMO-WP-0001.md"
|
|
text = workplan.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
"status: active\n---",
|
|
'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
|
|
)
|
|
text = text.replace(
|
|
"priority: high\n```",
|
|
'priority: high\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```',
|
|
)
|
|
workplan.write_text(text, encoding="utf-8")
|
|
brief = repo / ".custodian-brief.md"
|
|
brief.write_text("authoritative local brief\n", encoding="utf-8")
|
|
_git(repo, "add", ".")
|
|
_git(repo, "commit", "-m", "add authoritative identifiers")
|
|
_git(repo, "push")
|
|
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_check_primary",
|
|
lambda _api: ({"status": "ok", "db": "connected"}, None),
|
|
)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_projection_exists",
|
|
lambda _api, projection_id: ({"id": projection_id}, None),
|
|
)
|
|
|
|
def fake_run(command, *, env):
|
|
assert env["STATEHUB_REGISTRAR"] == "1"
|
|
brief.write_text("generated from partial projection\n", encoding="utf-8")
|
|
return subprocess.CompletedProcess(command, 1, "legacy stale references remain", "")
|
|
|
|
monkeypatch.setattr(rr, "_run_statehub", fake_run)
|
|
|
|
result = rr.registrar_reconcile(
|
|
repo,
|
|
statehub_bin="statehub",
|
|
confirm_primary=True,
|
|
repair_workplan="DEMO-WP-0001",
|
|
)
|
|
|
|
assert result.status == "applied"
|
|
assert result.evidence["repair_projection_verified"] is True
|
|
assert result.evidence["repair_projection_id"] == "11111111-1111-4111-8111-111111111111"
|
|
assert result.evidence["restored_generated_paths"] == [".custodian-brief.md"]
|
|
assert brief.read_text(encoding="utf-8") == "authoritative local brief\n"
|
|
|
|
|
|
def test_bootstraps_and_verifies_a_completely_empty_projection(tmp_path: Path, monkeypatch) -> None:
|
|
repo = _fixture(tmp_path)
|
|
workplan = repo / "workplans" / "DEMO-WP-0001.md"
|
|
text = workplan.read_text(encoding="utf-8")
|
|
text = text.replace(
|
|
"status: active\n---",
|
|
'status: active\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---',
|
|
)
|
|
text = text.replace(
|
|
"priority: high\n```",
|
|
'priority: high\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```',
|
|
)
|
|
workplan.write_text(text, encoding="utf-8")
|
|
_git(repo, "add", ".")
|
|
_git(repo, "commit", "-m", "add authoritative identifiers")
|
|
_git(repo, "push")
|
|
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_check_primary",
|
|
lambda _api: ({"status": "ok", "db": "connected"}, None),
|
|
)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_check_empty_repo_projection",
|
|
lambda _api, _slug: ({"repo_id": "repo-1", "workplan_count": 0}, None),
|
|
)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_verify_full_projection",
|
|
lambda _api, workplans, tasks, records: (
|
|
{
|
|
"expected_workplans": len(workplans),
|
|
"expected_tasks": len(tasks),
|
|
"expected_intakes": len(records["intake"]),
|
|
"expected_decisions": len(records["decision"]),
|
|
"missing_workplans": [],
|
|
"missing_tasks": [],
|
|
"missing_intakes": [],
|
|
"missing_decisions": [],
|
|
},
|
|
None,
|
|
),
|
|
)
|
|
|
|
def fake_run(command, *, env):
|
|
assert command[-1] == "--bootstrap-empty-projection"
|
|
assert env["STATEHUB_REGISTRAR"] == "1"
|
|
return subprocess.CompletedProcess(command, 2, "bootstrap warnings", "")
|
|
|
|
monkeypatch.setattr(rr, "_run_statehub", fake_run)
|
|
|
|
result = rr.registrar_reconcile(
|
|
repo,
|
|
statehub_bin="statehub",
|
|
confirm_primary=True,
|
|
bootstrap_empty_projection=True,
|
|
)
|
|
|
|
assert result.status == "applied"
|
|
assert result.evidence["bootstrap_source"] == {
|
|
"workplans": 1,
|
|
"tasks": 1,
|
|
"intakes": 0,
|
|
"decisions": 0,
|
|
}
|
|
assert result.evidence["bootstrap_projection_verified"] is True
|
|
|
|
|
|
def test_empty_projection_bootstrap_refuses_existing_rows(tmp_path: Path, monkeypatch) -> None:
|
|
repo = _fixture(tmp_path)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_check_primary",
|
|
lambda _api: ({"status": "ok", "db": "connected"}, None),
|
|
)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_check_empty_repo_projection",
|
|
lambda _api, _slug: (
|
|
{"repo_id": "repo-1", "workplan_count": 1},
|
|
"target repository projection is not empty",
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
rr,
|
|
"_run_statehub",
|
|
lambda *_args, **_kwargs: pytest.fail("non-empty bootstrap must not run State Hub"),
|
|
)
|
|
|
|
result = rr.registrar_reconcile(
|
|
repo,
|
|
confirm_primary=True,
|
|
bootstrap_empty_projection=True,
|
|
)
|
|
|
|
assert result.status == "rejected"
|
|
assert result.error and result.error["code"] == "bootstrap_precondition_failed"
|