tegwick
956efbb7ae
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
138 lines
4.8 KiB
Python
138 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from repo_manager.identifiers import (
|
|
derive_work_record_uuid,
|
|
load_fleet_namespace,
|
|
plan_identifier_migration,
|
|
scan_live_identifier_collisions,
|
|
verify_identifier_migration_plan,
|
|
)
|
|
|
|
|
|
def test_declared_fleet_namespace_is_helixforge() -> None:
|
|
assert load_fleet_namespace() == "helixforge"
|
|
|
|
|
|
def _workplan(path: Path, identifier: str, status: str, task_status: str = "todo") -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
f"""---
|
|
id: {identifier}
|
|
title: Test
|
|
status: {status}
|
|
---
|
|
|
|
## Task
|
|
|
|
```task
|
|
id: {identifier}-T01
|
|
status: {task_status}
|
|
```
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_uuid_derivation_is_stable_and_namespace_scoped() -> None:
|
|
first = derive_work_record_uuid("helixforge", "RMGR-WP-0005")
|
|
assert str(first) == "6dcf854e-9229-569d-90f0-2d9df235a61d"
|
|
assert derive_work_record_uuid("helixforge", "RMGR-WP-0005") == first
|
|
assert derive_work_record_uuid("client-a", "RMGR-WP-0005") != first
|
|
with pytest.raises(ValueError):
|
|
derive_work_record_uuid("Client A", "RMGR-WP-0005")
|
|
|
|
|
|
def test_preflight_blocks_live_collision_but_ignores_archived_history(tmp_path: Path) -> None:
|
|
one = tmp_path / "one"
|
|
two = tmp_path / "two"
|
|
_workplan(one / "workplans" / "one.md", "SHARED-WP-0001", "active")
|
|
_workplan(two / "workplans" / "two.md", "SHARED-WP-0001", "ready")
|
|
report = scan_live_identifier_collisions(tmp_path)
|
|
assert report["ok"] is False
|
|
assert "SHARED-WP-0001" in report["collisions"]
|
|
|
|
_workplan(two / "workplans" / "two.md", "SHARED-WP-0001", "archived")
|
|
report = scan_live_identifier_collisions(tmp_path)
|
|
assert report["ok"] is True
|
|
|
|
|
|
def test_migration_plan_preserves_mapping_and_is_atomic_per_repo(tmp_path: Path) -> None:
|
|
one = tmp_path / "one"
|
|
two = tmp_path / "two"
|
|
_workplan(one / "workplans" / "one.md", "ONE-WP-0001", "active")
|
|
_workplan(two / "workplans" / "two.md", "TWO-WP-0001", "ready", task_status="done")
|
|
|
|
report = plan_identifier_migration(tmp_path, "fleet-a")
|
|
|
|
assert report["ok"] is True
|
|
assert report["apply_policy"] == "all-or-nothing per repository"
|
|
assert report["totals"] == {
|
|
"repositories": 2,
|
|
"eligible": 2,
|
|
"skipped": 0,
|
|
"records": 3,
|
|
"replace": 0,
|
|
"assign": 3,
|
|
"unchanged": 0,
|
|
}
|
|
one_plan = next(item for item in report["repositories"] if item["repo"] == "one")
|
|
assert one_plan["atomic_unit"] is True
|
|
assert {item["record_id"] for item in one_plan["mappings"]} == {
|
|
"ONE-WP-0001",
|
|
"ONE-WP-0001-T01",
|
|
}
|
|
assert all(item["current_uuid"] is None for item in one_plan["mappings"])
|
|
assert len(report["plan_sha256"]) == 64
|
|
assert report["generated_at"].endswith("Z")
|
|
|
|
|
|
def test_migration_plan_skips_entire_repo_affected_by_collision(tmp_path: Path) -> None:
|
|
one = tmp_path / "one"
|
|
two = tmp_path / "two"
|
|
_workplan(one / "workplans" / "one.md", "SHARED-WP-0001", "active")
|
|
_workplan(one / "workplans" / "safe.md", "ONE-WP-0002", "active")
|
|
_workplan(two / "workplans" / "two.md", "SHARED-WP-0001", "ready")
|
|
|
|
report = plan_identifier_migration(tmp_path, "fleet-a")
|
|
|
|
assert report["ok"] is False
|
|
assert report["totals"]["skipped"] == 2
|
|
one_plan = next(item for item in report["repositories"] if item["repo"] == "one")
|
|
assert one_plan["eligible"] is False
|
|
assert any(item["record_id"] == "ONE-WP-0002" for item in one_plan["mappings"])
|
|
assert any(item["reason"] == "live identifier collision" for item in one_plan["blockers"])
|
|
|
|
|
|
def test_migration_verification_detects_tampering_and_source_drift(tmp_path: Path) -> None:
|
|
repo = tmp_path / "one"
|
|
path = repo / "workplans" / "one.md"
|
|
_workplan(path, "ONE-WP-0001", "active")
|
|
subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
|
|
subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-m", "seed"],
|
|
cwd=repo,
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
plan = plan_identifier_migration(tmp_path, "helixforge")
|
|
assert verify_identifier_migration_plan(plan)["ok"] is True
|
|
|
|
path.write_text(path.read_text(encoding="utf-8") + "\nchanged\n", encoding="utf-8")
|
|
verification = verify_identifier_migration_plan(plan)
|
|
assert verification["ok"] is False
|
|
assert any(
|
|
error["reason"] == "authoritative source changed after planning"
|
|
for error in verification["errors"]
|
|
)
|
|
|
|
plan["namespace"] = "tampered"
|
|
assert any(
|
|
error["reason"] == "plan SHA-256 mismatch"
|
|
for error in verify_identifier_migration_plan(plan)["errors"]
|
|
)
|