Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 16 additions & 13 deletions packages/darnit/src/darnit/tools/audit_org.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
143 changes: 137 additions & 6 deletions tests/darnit/test_audit_org.py
Original file line number Diff line number Diff line change
Expand Up @@ -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},
},
]
Expand All @@ -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},
},
]
Expand All @@ -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},
},
{
Expand All @@ -269,14 +283,131 @@ 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},
},
]
summary = aggregate_org_results("org", results, 1)
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."""
Expand Down
Loading