tegwick
ad621d6c0d
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from repo_manager.identifiers import (
|
|
derive_work_record_uuid,
|
|
plan_identifier_migration,
|
|
scan_live_identifier_collisions,
|
|
)
|
|
|
|
|
|
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"])
|
|
|
|
|
|
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"])
|