the-custodian/tests/test_repo_classification_convergence.py

from __future__ import annotations

from pathlib import Path

from tools.repo_classification_convergence import analyze_repositories, render_text


def _record(slug: str, path: Path | None, *, category: str | None) -> dict:
    return {
        "slug": slug,
        "status": "active",
        "category": category,
        "local_path": str(path) if path else None,
    }


def _write_valid(path: Path) -> None:
    path.mkdir()
    (path / ".repo-classification.yaml").write_text(
        """repo_classification:
  category: project
  domain: infotech
  secondary_domains: []
  capability_tags: []
  business_stake: []
  business_mechanics: []
""",
        encoding="utf-8",
    )


def test_report_separates_owner_gap_from_projection_gap(tmp_path: Path) -> None:
    classified = tmp_path / "classified"
    unprojected = tmp_path / "unprojected"
    _write_valid(classified)
    _write_valid(unprojected)

    report = analyze_repositories(
        [
            _record("classified", classified, category="project"),
            _record("owner-gap", tmp_path / "owner-gap", category=None),
            _record("projection-gap", unprojected, category=None),
            {"slug": "archived", "status": "archived", "category": None},
        ]
    )

    assert report["registered"] == 4
    assert report["active"] == 3
    assert report["active_classified"] == 1
    assert report["missing_source"] == ["owner-gap"]
    assert report["present_unprojected"] == ["projection-gap"]
    assert report["converged"] is False


def test_report_flags_stale_projection_and_invalid_source(tmp_path: Path) -> None:
    invalid = tmp_path / "invalid"
    invalid.mkdir()
    (invalid / ".repo-classification.yaml").write_text(
        "repo_classification:\n  category: made-up\n  domain: custodian\n",
        encoding="utf-8",
    )

    report = analyze_repositories(
        [
            _record("stale", tmp_path / "stale", category="project"),
            _record("invalid", invalid, category="project"),
        ]
    )

    assert report["projected_without_source"] == ["stale"]
    assert "invalid" in report["invalid_source"]
    assert report["converged"] is False


def test_clean_active_fleet_converges_and_renders(tmp_path: Path) -> None:
    repo = tmp_path / "repo"
    _write_valid(repo)
    report = analyze_repositories([_record("repo", repo, category="project")])

    assert report["converged"] is True
    assert report["source_warning_count"] == 0
    assert "active classified: 1" in render_text(report)
    assert "converged: yes" in render_text(report)


def test_host_paths_for_this_machine_beat_other_host_local_path(tmp_path: Path) -> None:
    local = tmp_path / "repo-manager"
    _write_valid(local)
    report = analyze_repositories(
        [
            {
                "slug": "repo-manager",
                "status": "active",
                "category": "tooling",
                "local_path": "/home/tegwick/repo-manager",
                "host_paths": {
                    "bnt-lap001": str(local),
                    "239.62.205.92.host.secureserver.net": "/home/tegwick/repo-manager",
                },
            }
        ],
        hostname="bnt-lap001",
    )

    assert report["projected_without_source"] == []
    assert report["converged"] is True