repo-manager/tests/test_custodian_legacy_identity.py

from pathlib import Path

from repo_manager.identifiers import ensure_missing_work_record_identifiers
from repo_manager.record_identity import scan_record_identities


def test_custodian_scoped_legacy_ids_are_grandfathered_and_parent_scoped(tmp_path: Path) -> None:
    root = tmp_path
    for number, workplan_id in enumerate(
        ("CUST-WP-0000c", "CUST-WP-0010b", "CUST-WP-0017-LEGACY", "CUST-WP-0053-LEGACY"), start=1
    ):
        _write_workplan(
            root / "workplans" / f"{number}.md",
            workplan_id,
            f"{workplan_id}-T01",
            f"{number:08d}-1111-4111-8111-111111111111",
        )
    report = scan_record_identities(root)
    assert report["invalid_identifiers"] == []
    assert report["identity_collisions"] == []
    assert report["duplicate_source_occurrences"] == []


def test_full_legacy_bindings_dry_run_has_no_assignments(tmp_path: Path) -> None:
    for number, workplan_id in enumerate(
        ("CUST-WP-0000c", "CUST-WP-0010b", "CUST-WP-0017-LEGACY", "CUST-WP-0053-LEGACY"), start=1
    ):
        _write_workplan(
            tmp_path / "workplans" / f"{number}.md",
            workplan_id,
            f"{workplan_id}-T01",
            f"{number:08d}-1111-4111-8111-111111111111",
        )
    result = ensure_missing_work_record_identifiers(tmp_path, execute=False)
    assert result["assignments"] == []
    assert result["files_changed"] == []


def test_unqualified_task_remains_invalid_outside_allowlisted_parents(tmp_path: Path) -> None:
    workplan = tmp_path / "workplans" / "CUST-WP-0099.md"
    workplan.parent.mkdir()
    workplan.write_text(
        '---\nid: CUST-WP-0099\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---\n'
        '\n```task\nid: T01\nstatus: done\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```\n',
        encoding="utf-8",
    )
    report = scan_record_identities(tmp_path)
    assert report["invalid_identifiers"][0]["id"] == "T01"


def _write_workplan(path: Path, workplan_id: str, task_id: str, task_uuid: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        f'---\nid: {workplan_id}\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---\n'
        f'\n```task\nid: {task_id}\nstatus: done\nstate_hub_task_id: "{task_uuid}"\n```\n',
        encoding="utf-8",
    )


def test_canonical_qualified_task_collision_remains_global(tmp_path: Path) -> None:
    _write_workplan(
        tmp_path / "workplans" / "a.md",
        "CUST-WP-0099",
        "CUST-WP-0099-T01",
        "22222222-2222-4222-8222-222222222222",
    )
    _write_workplan(
        tmp_path / "workplans" / "b.md",
        "CUST-WP-0100",
        "CUST-WP-0099-T01",
        "33333333-3333-4333-8333-333333333333",
    )
    report = scan_record_identities(tmp_path)
    assert report["identity_collisions"][0]["id"] == "CUST-WP-0099-T01"


def test_same_parent_local_alias_collision_fails_closed(tmp_path: Path) -> None:
    path = tmp_path / "workplans" / "CUST-WP-0006.md"
    path.parent.mkdir()
    path.write_text(
        '---\nid: CUST-WP-0006\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---\n'
        '\n```task\nid: T01\nstatus: done\nstate_hub_task_id: "22222222-2222-4222-8222-222222222222"\n```\n'
        '\n```task\nid: T01\nstatus: done\nstate_hub_task_id: "33333333-3333-4333-8333-333333333333"\n```\n',
        encoding="utf-8",
    )
    report = scan_record_identities(tmp_path)
    assert report["identity_collisions"][0]["id"] == "T01"


def test_unknown_alias_and_missing_managed_uuid_are_invalid(tmp_path: Path) -> None:
    _write_workplan(
        tmp_path / "workplans" / "unknown.md",
        "CUST-WP-0006",
        "T99",
        "22222222-2222-4222-8222-222222222222",
    )
    _write_workplan(tmp_path / "workplans" / "missing.md", "CUST-WP-0006", "T01", "")
    report = scan_record_identities(tmp_path)
    assert {item["id"] for item in report["invalid_identifiers"]} == {"T99", "T01"}


def test_all_38_local_aliases_preserve_binding_and_source_bytes(tmp_path):
    before = {}
    for parent, count in (
        ("CUST-WP-0006", 5),
        ("CUST-WP-0007", 17),
        ("CUST-WP-0030", 9),
        ("CUST-WP-0031", 7),
    ):
        path = tmp_path / "workplans" / (parent + ".md")
        path.parent.mkdir(exist_ok=True)
        body = f'---\nid: {parent}\nstate_hub_workstream_id: "11111111-1111-4111-8111-111111111111"\n---\n'
        for n in range(1, count + 1):
            body += f'\n```task\nid: T{n:02d}\nstatus: done\nstate_hub_task_id: "{int(parent[-4:]):08d}-1111-4111-8111-{n:012d}"\n```\n'
        path.write_text(body)
        before[path] = path.read_bytes()
    assert scan_record_identities(tmp_path)["ok"]
    result = ensure_missing_work_record_identifiers(tmp_path, execute=True)
    assert result["assignments"] == [] and result["files_changed"] == []
    assert all(path.read_bytes() == body for path, body in before.items())


def test_new_suffix_task_is_not_grandfathered(tmp_path):
    _write_workplan(
        tmp_path / "workplans" / "new.md",
        "CUST-WP-0000c",
        "CUST-WP-0000c-T99",
        "22222222-2222-4222-8222-222222222222",
    )
    assert not scan_record_identities(tmp_path)["ok"]