audit-labs/audit-tools

  1"""Tests for the TUI's AWS audit orchestration.
  2
  3Stub the boto3 session and the collectors, then verify run_audit's wiring:
  4each check is called with the config, per-check errors don't abort the run, a
  5failed session build is reported cleanly, and the CSV package is written.
  6"""
  7
  8import csv
  9import os
 10
 11import pytest
 12
 13from tui import aws_runner as r
 14
 15
 16@pytest.fixture
 17def fake_checks(monkeypatch):
 18    calls = {}
 19
 20    def users_fn(cfg):
 21        calls["users"] = cfg
 22        return [{"user": "alice"}, {"user": "bob"}]
 23
 24    def policy_fn(cfg):
 25        calls["policy"] = cfg
 26        return [{"minimum_length": 14}]
 27
 28    def boom_fn(cfg):
 29        raise RuntimeError("kaboom")
 30
 31    checks = [
 32        r.Check("users", "Users", users_fn, "users.csv"),
 33        r.Check("policy", "Policy", policy_fn, "policy.csv"),
 34        r.Check("boom", "Boom", boom_fn, "boom.csv"),
 35    ]
 36    monkeypatch.setattr(r, "CHECKS", checks)
 37
 38    # Replace build_cfg so no real boto3 session is created.
 39    monkeypatch.setattr(
 40        r.api,
 41        "build_cfg",
 42        lambda profile, region, account: {
 43            "session": "SESSION",
 44            "profile": profile,
 45            "region": region,
 46            "account": account,
 47        },
 48    )
 49    return calls
 50
 51
 52def run(tmp_path, keys, profile="", region="us-east-1", account=""):
 53    events = []
 54    sections = r.run_audit(
 55        profile=profile,
 56        region=region,
 57        account=account,
 58        output_dir=str(tmp_path),
 59        selected_keys=keys,
 60        on_event=events.append,
 61    )
 62    return events, sections
 63
 64
 65def test_dispatch_and_files(tmp_path, fake_checks):
 66    calls = fake_checks
 67    run(tmp_path, ["users", "policy"], region="eu-west-1")
 68
 69    # Each collector received the config dict.
 70    assert calls["users"]["region"] == "eu-west-1"
 71    assert calls["policy"]["session"] == "SESSION"
 72
 73    for name in ("users.csv", "policy.csv", "summary.txt"):
 74        assert os.path.exists(tmp_path / name), name
 75
 76    with open(tmp_path / "users.csv", newline="") as f:
 77        assert len(list(csv.DictReader(f))) == 2
 78
 79
 80def test_failing_check_does_not_abort_run(tmp_path, fake_checks):
 81    events, sections = run(tmp_path, ["boom", "users"])
 82
 83    kinds = [(e.kind, e.label) for e in events]
 84    assert ("error", "Boom") in kinds
 85    assert ("done", "Users") in kinds
 86
 87    labels = dict(sections)
 88    assert labels["Boom"] == 0
 89    assert labels["Users"] == 2
 90
 91
 92def test_session_build_failure_is_reported(tmp_path, fake_checks, monkeypatch):
 93    def boom_cfg(profile, region, account):
 94        raise RuntimeError("ProfileNotFound")
 95
 96    monkeypatch.setattr(r.api, "build_cfg", boom_cfg)
 97
 98    events, sections = run(tmp_path, ["users"], profile="ghost")
 99
100    kinds = [(e.kind, e.label) for e in events]
101    assert ("error", "AWS session") in kinds
102    # Run still ends with a summary and writes the (empty) package.
103    summary = [e for e in events if e.kind == "summary"]
104    assert summary
105    assert summary[0].count == 0
106    assert sections == []
107    assert os.path.exists(tmp_path / "summary.txt")
108
109
110def test_subject_defaults_to_default(tmp_path, fake_checks):
111    run(tmp_path, ["users"], profile="")
112    # An empty profile is folder-named "default".
113    assert r.default_output_dir("./out", "") == r.default_output_dir("./out", "default")
114
115
116def test_sso_off_by_default():
117    assert "sso_assignments" not in r.DEFAULT_SELECTION
118    assert "iam_users" in r.DEFAULT_SELECTION