audit-labs/audit-tools

  1"""Tests for the TUI's GitHub audit orchestration.
  2
  3These stub out the network-bound collectors and verify run_audit's wiring:
  4argument dispatch, the shared collaborator cache, per-check error handling,
  5and that the CSV package (per-check files + summary) is written.
  6"""
  7
  8import csv
  9import os
 10
 11import pytest
 12
 13from tui import github_runner as r
 14
 15
 16@pytest.fixture
 17def fake_checks(monkeypatch):
 18    """Replace the real registry with stubbed collectors and record calls."""
 19    calls = {}
 20
 21    def base_fn(org, cfg):
 22        calls["base"] = (org, cfg)
 23        return [{"login": "alice"}, {"login": "bob"}]
 24
 25    def collabs_fn(org, cfg, repo_collabs):
 26        calls["collabs"] = (org, cfg, repo_collabs)
 27        return [{"repo": e["repo"]} for e in repo_collabs]
 28
 29    def branch_fn(org, cfg, branch):
 30        calls["branch"] = (org, cfg, branch)
 31        return [{"branch": branch}]
 32
 33    def boom_fn(org, cfg):
 34        raise RuntimeError("kaboom")
 35
 36    checks = [
 37        r.Check("base", "Base", base_fn, "base.csv"),
 38        r.Check("collabs", "Collabs", collabs_fn, "collabs.csv", arg="collabs"),
 39        r.Check("branch", "Branch", branch_fn, "branch.csv", arg="branch"),
 40        r.Check("boom", "Boom", boom_fn, "boom.csv"),
 41    ]
 42    monkeypatch.setattr(r, "CHECKS", checks)
 43
 44    fetch_calls = []
 45
 46    def fake_fetch(org, cfg):
 47        fetch_calls.append(org)
 48        return [{"repo": "repo1", "collaborators": []}]
 49
 50    monkeypatch.setattr(r.members, "fetch_repo_collaborators", fake_fetch)
 51
 52    return calls, fetch_calls
 53
 54
 55def run(tmp_path, keys, fake_checks, branch="main"):
 56    events = []
 57    sections = r.run_audit(
 58        org="acme",
 59        token="tok",
 60        output_dir=str(tmp_path),
 61        branch=branch,
 62        selected_keys=keys,
 63        on_event=events.append,
 64    )
 65    return events, sections
 66
 67
 68def test_argument_dispatch_and_files(tmp_path, fake_checks):
 69    calls, _ = fake_checks
 70    run(tmp_path, ["base", "collabs", "branch"], fake_checks, "dev")
 71
 72    # Each collector was called with the right signature.
 73    assert calls["base"][0] == "acme"
 74    assert calls["collabs"][2] == [{"repo": "repo1", "collaborators": []}]
 75    assert calls["branch"][2] == "dev"
 76
 77    # CSV files were written for each check, plus the summary.
 78    for name in ("base.csv", "collabs.csv", "branch.csv", "summary.txt"):
 79        assert os.path.exists(tmp_path / name), name
 80
 81    with open(tmp_path / "base.csv", newline="") as f:
 82        assert len(list(csv.DictReader(f))) == 2
 83
 84
 85def test_collab_cache_fetched_once(tmp_path, fake_checks):
 86    _, fetch_calls = fake_checks
 87    run(tmp_path, ["collabs", "base"], fake_checks)
 88    assert fetch_calls == ["acme"]  # fetched exactly once
 89
 90
 91def test_collab_cache_skipped_when_not_needed(tmp_path, fake_checks):
 92    _, fetch_calls = fake_checks
 93    run(tmp_path, ["base"], fake_checks)
 94    assert fetch_calls == []  # no collabs check selected -> no fetch
 95
 96
 97def test_failing_check_does_not_abort_run(tmp_path, fake_checks):
 98    events, sections = run(tmp_path, ["boom", "base"], fake_checks)
 99
100    kinds = [(e.kind, e.label) for e in events]
101    assert ("error", "Boom") in kinds
102    assert ("done", "Base") in kinds  # base still ran after boom failed
103
104    labels = dict(sections)
105    assert labels["Boom"] == 0
106    assert labels["Base"] == 2
107
108
109def test_summary_event_totals_rows(tmp_path, fake_checks):
110    events, _ = run(tmp_path, ["base", "branch"], fake_checks)
111    summary = [e for e in events if e.kind == "summary"]
112    assert len(summary) == 1
113    assert summary[0].count == 3  # 2 base + 1 branch
114    assert summary[0].label == str(tmp_path)