diff --git a/packages/darnit/src/darnit/tools/audit_org.py b/packages/darnit/src/darnit/tools/audit_org.py index b363b883..e3d54f9b 100644 --- a/packages/darnit/src/darnit/tools/audit_org.py +++ b/packages/darnit/src/darnit/tools/audit_org.py @@ -273,28 +273,31 @@ def aggregate_org_results( }) continue - pass_count = summary.get("PASS", 0) - fail_count = summary.get("FAIL", 0) - warn_count = summary.get("WARN", 0) - total = summary.get("total", 0) + check_results = result.get("results") or [] + from darnit.tools.audit import calculate_compliance, summarize_results - # A repo is compliant only if all non-N/A controls pass - na_count = summary.get("N/A", 0) - is_compliant = fail_count == 0 and warn_count == 0 and (pass_count + na_count == total) + compliance = calculate_compliance(check_results, level) + is_compliant = bool(compliance) and all( + compliance.get(lvl, False) for lvl in range(1, level + 1) + ) + if not summary: + summary = summarize_results(check_results) if is_compliant: org_summary["compliant_repos"] += 1 else: org_summary["non_compliant_repos"] += 1 - org_summary["repos"].append({ + repo_entry: dict[str, Any] = { "repo": repo_name, "status": "COMPLIANT" if is_compliant else "NON_COMPLIANT", - "pass": pass_count, - "fail": fail_count, - "warn": warn_count, - "total": total, - }) + "pass": summary.get("PASS", 0), + "fail": summary.get("FAIL", 0), + "warn": summary.get("WARN", 0), + "total": summary.get("total", 0), + "compliance": compliance, + } + org_summary["repos"].append(repo_entry) return org_summary diff --git a/tests/darnit/test_audit_org.py b/tests/darnit/test_audit_org.py index 93b5106a..4f1eccf7 100644 --- a/tests/darnit/test_audit_org.py +++ b/tests/darnit/test_audit_org.py @@ -206,13 +206,17 @@ def test_all_pass(self): { "repo": "a", "status": "OK", - "results": [], + "results": [ + {"id": f"C-{i}", "status": "PASS", "level": 1} for i in range(5) + ], "summary": {"PASS": 5, "FAIL": 0, "WARN": 0, "N/A": 0, "total": 5}, }, { "repo": "b", "status": "OK", - "results": [], + "results": [ + {"id": f"C-{i}", "status": "PASS", "level": 1} for i in range(5) + ], "summary": {"PASS": 5, "FAIL": 0, "WARN": 0, "N/A": 0, "total": 5}, }, ] @@ -227,13 +231,21 @@ def test_mixed_results(self): { "repo": "good", "status": "OK", - "results": [], + "results": [ + {"id": f"C-{i}", "status": "PASS", "level": 1} for i in range(5) + ], "summary": {"PASS": 5, "FAIL": 0, "WARN": 0, "N/A": 0, "total": 5}, }, { "repo": "bad", "status": "OK", - "results": [], + "results": [ + {"id": "C-1", "status": "PASS", "level": 1}, + {"id": "C-2", "status": "PASS", "level": 1}, + {"id": "C-3", "status": "PASS", "level": 1}, + {"id": "C-4", "status": "FAIL", "level": 1}, + {"id": "C-5", "status": "FAIL", "level": 1}, + ], "summary": {"PASS": 3, "FAIL": 2, "WARN": 0, "N/A": 0, "total": 5}, }, ] @@ -247,7 +259,9 @@ def test_error_repos(self): { "repo": "good", "status": "OK", - "results": [], + "results": [ + {"id": f"C-{i}", "status": "PASS", "level": 1} for i in range(5) + ], "summary": {"PASS": 5, "FAIL": 0, "WARN": 0, "N/A": 0, "total": 5}, }, { @@ -269,7 +283,13 @@ def test_warn_counts_as_non_compliant(self): { "repo": "warned", "status": "OK", - "results": [], + "results": [ + {"id": "C-1", "status": "PASS", "level": 1}, + {"id": "C-2", "status": "PASS", "level": 1}, + {"id": "C-3", "status": "PASS", "level": 1}, + {"id": "C-4", "status": "PASS", "level": 1}, + {"id": "C-5", "status": "WARN", "level": 1}, + ], "summary": {"PASS": 4, "FAIL": 0, "WARN": 1, "N/A": 0, "total": 5}, }, ] @@ -277,6 +297,117 @@ def test_warn_counts_as_non_compliant(self): assert summary["non_compliant_repos"] == 1 assert summary["compliant_repos"] == 0 + def test_results_with_warn_error_and_pending_is_non_compliant(self): + """Results with WARN, ERROR, and PENDING controls are non-compliant.""" + results = [ + { + "repo": "troubled", + "status": "OK", + "results": [ + {"id": "OSPS-AC-01.01", "status": "PASS", "level": 1}, + {"id": "OSPS-BR-01.01", "status": "WARN", "level": 1}, + {"id": "OSPS-QA-01.01", "status": "ERROR", "level": 1}, + {"id": "OSPS-DO-01.01", "status": "PENDING", "level": 1}, + ], + "summary": { + "PASS": 1, + "FAIL": 0, + "WARN": 1, + "N/A": 0, + "ERROR": 1, + "PENDING": 1, + "total": 4, + }, + }, + ] + summary = aggregate_org_results("org", results, 1) + assert summary["compliant_repos"] == 0 + assert summary["non_compliant_repos"] == 1 + repo = summary["repos"][0] + assert repo["status"] == "NON_COMPLIANT" + assert repo["compliance"] == {1: False} + + def test_results_with_only_pass_and_na_is_compliant(self): + """Results with only PASS and N/A controls are compliant.""" + results = [ + { + "repo": "clean", + "status": "OK", + "results": [ + {"id": "OSPS-AC-01.01", "status": "PASS", "level": 1}, + {"id": "OSPS-AC-02.01", "status": "PASS", "level": 1}, + {"id": "OSPS-BR-01.01", "status": "N/A", "level": 1}, + ], + "summary": {"PASS": 2, "FAIL": 0, "WARN": 0, "N/A": 1, "total": 3}, + }, + ] + summary = aggregate_org_results("org", results, 1) + assert summary["compliant_repos"] == 1 + assert summary["non_compliant_repos"] == 0 + repo = summary["repos"][0] + assert repo["status"] == "COMPLIANT" + assert repo["compliance"] == {1: True} + + def test_multi_level_compliance_calculation(self): + """Multi-level audit correctly calculates per-level compliance.""" + results = [ + { + "repo": "partially-compliant", + "status": "OK", + "results": [ + {"id": "L1-01", "status": "PASS", "level": 1}, + {"id": "L2-01", "status": "FAIL", "level": 2}, + ], + }, + ] + summary_l1 = aggregate_org_results("org", results, 1) + assert summary_l1["compliant_repos"] == 1 + assert summary_l1["repos"][0]["status"] == "COMPLIANT" + + summary_l2 = aggregate_org_results("org", results, 2) + assert summary_l2["compliant_repos"] == 0 + assert summary_l2["non_compliant_repos"] == 1 + assert summary_l2["repos"][0]["status"] == "NON_COMPLIANT" + assert summary_l2["repos"][0]["compliance"] == {1: True, 2: False} + + def test_level_2_audit_requires_level_1_to_pass(self): + """A repo with L1 FAIL and L2 PASS is non-compliant when audited at level 2.""" + results = [ + { + "repo": "l1-fail-l2-pass", + "status": "OK", + "results": [ + {"id": "L1-01", "status": "FAIL", "level": 1}, + {"id": "L2-01", "status": "PASS", "level": 2}, + ], + }, + ] + summary = aggregate_org_results("org", results, 2) + assert summary["compliant_repos"] == 0 + assert summary["non_compliant_repos"] == 1 + repo = summary["repos"][0] + assert repo["status"] == "NON_COMPLIANT" + assert repo["compliance"] == {1: False, 2: True} + + def test_level_2_audit_passes_when_all_levels_pass(self): + """A repo with L1 PASS and L2 PASS is compliant when audited at level 2.""" + results = [ + { + "repo": "fully-compliant", + "status": "OK", + "results": [ + {"id": "L1-01", "status": "PASS", "level": 1}, + {"id": "L2-01", "status": "PASS", "level": 2}, + ], + }, + ] + summary = aggregate_org_results("org", results, 2) + assert summary["compliant_repos"] == 1 + assert summary["non_compliant_repos"] == 0 + repo = summary["repos"][0] + assert repo["status"] == "COMPLIANT" + assert repo["compliance"] == {1: True, 2: True} + class TestFormatOrgResults: """Tests for format_org_results_markdown and format_org_results_json."""