Skip to content

Commit a7dfcff

Browse files
committed
Add RFC compliance metrics with grading system
- Add ComplianceMetrics class with compliance score calculation - Add compliance grade system (A-F) with configurable thresholds - Add severity level classification for test failures - Add severity-weighted compliance scoring - Add RFC section compliance breakdown by RFC 4271 sections - Add generate_compliance_report() method - Update test report to show compliance summary, grades, and RFC compliance - Add comprehensive tests for all compliance metrics functionality - All 118 tests pass
1 parent 569c08a commit a7dfcff

2 files changed

Lines changed: 327 additions & 5 deletions

File tree

src/bgp_test_framework/runner.py

Lines changed: 187 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,162 @@ def get_log(self) -> List[Dict[str, Any]]:
9999
return self.entries
100100

101101

102+
class ComplianceMetrics:
103+
COMPLIANCE_GRADE_THRESHOLDS = {
104+
"A": 95.0,
105+
"B": 85.0,
106+
"C": 70.0,
107+
"D": 50.0,
108+
"F": 0.0,
109+
}
110+
111+
SEVERITY_LEVELS = {
112+
"CRITICAL": 10,
113+
"HIGH": 7,
114+
"MEDIUM": 4,
115+
"LOW": 2,
116+
"INFO": 1,
117+
}
118+
119+
@staticmethod
120+
def calculate_compliance_score(total: int, passed: int) -> float:
121+
if total == 0:
122+
return 0.0
123+
return (passed / total) * 100.0
124+
125+
@classmethod
126+
def get_compliance_grade(cls, score: float) -> str:
127+
if score >= cls.COMPLIANCE_GRADE_THRESHOLDS["A"]:
128+
return "A"
129+
elif score >= cls.COMPLIANCE_GRADE_THRESHOLDS["B"]:
130+
return "B"
131+
elif score >= cls.COMPLIANCE_GRADE_THRESHOLDS["C"]:
132+
return "C"
133+
elif score >= cls.COMPLIANCE_GRADE_THRESHOLDS["D"]:
134+
return "D"
135+
return "F"
136+
137+
@staticmethod
138+
def get_severity_level(test_id: str) -> str:
139+
if test_id.startswith("MH-"):
140+
return "CRITICAL"
141+
elif test_id.startswith("OM-"):
142+
return "HIGH"
143+
elif test_id.startswith("UM-"):
144+
return "HIGH"
145+
elif test_id.startswith("AT-"):
146+
return "MEDIUM"
147+
elif test_id.startswith("FSM-"):
148+
return "HIGH"
149+
elif test_id.startswith("TM-"):
150+
return "MEDIUM"
151+
elif test_id.startswith("SEC-"):
152+
return "CRITICAL"
153+
elif test_id.startswith("RA-"):
154+
return "MEDIUM"
155+
elif test_id.startswith("DEC-"):
156+
return "LOW"
157+
return "INFO"
158+
159+
@staticmethod
160+
def calculate_severity_score(failed_tests: List[TestResult]) -> Dict[str, int]:
161+
severity_counts: Dict[str, int] = {
162+
level: 0 for level in ComplianceMetrics.SEVERITY_LEVELS
163+
}
164+
for result in failed_tests:
165+
severity = ComplianceMetrics.get_severity_level(result.test_id)
166+
severity_counts[severity] += 1
167+
return severity_counts
168+
169+
@staticmethod
170+
def calculate_severity_weighted_score(
171+
total: int, passed: int, failed_tests: List[TestResult]
172+
) -> float:
173+
if total == 0:
174+
return 0.0
175+
176+
max_score = sum(
177+
ComplianceMetrics.SEVERITY_LEVELS[
178+
ComplianceMetrics.get_severity_level(r.test_id)
179+
]
180+
for r in failed_tests
181+
) + (ComplianceMetrics.SEVERITY_LEVELS["INFO"] * passed if passed > 0 else 0)
182+
183+
failed_score = sum(
184+
ComplianceMetrics.SEVERITY_LEVELS[
185+
ComplianceMetrics.get_severity_level(r.test_id)
186+
]
187+
for r in failed_tests
188+
)
189+
190+
if max_score == 0:
191+
return 100.0
192+
193+
return ((max_score - failed_score) / max_score) * 100.0
194+
195+
@staticmethod
196+
def get_rfc_section_compliance(
197+
results: List[TestResult],
198+
) -> Dict[str, Dict[str, Any]]:
199+
rfc_sections = {
200+
"RFC 4271 Section 4.1": "message_header",
201+
"RFC 4271 Section 4.2": "open_message",
202+
"RFC 4271 Section 4.3": "update_message",
203+
"RFC 4271 Section 5": "attribute",
204+
"RFC 4271 Section 8": "fsm",
205+
"RFC 4271 Section 4.4": "timing",
206+
"RFC 4271 Section 6": "security",
207+
"RFC 4271 Section 9.2": "route_aggregation",
208+
"RFC 4271 Section 9.1": "decision_process",
209+
}
210+
211+
compliance: Dict[str, Dict[str, Any]] = {}
212+
for section, category in rfc_sections.items():
213+
section_results = [r for r in results if r.category.value == category]
214+
total = len(section_results)
215+
passed_count = sum(1 for r in section_results if r.passed)
216+
score = ComplianceMetrics.calculate_compliance_score(total, passed_count)
217+
218+
compliance[section] = {
219+
"total": total,
220+
"passed": passed_count,
221+
"failed": total - passed_count,
222+
"score": score,
223+
"grade": ComplianceMetrics.get_compliance_grade(score),
224+
}
225+
226+
return compliance
227+
228+
@staticmethod
229+
def generate_compliance_report(results: List[TestResult]) -> Dict[str, Any]:
230+
total = len(results)
231+
passed = sum(1 for r in results if r.passed)
232+
failed = total - passed
233+
score = ComplianceMetrics.calculate_compliance_score(total, passed)
234+
failed_tests = [r for r in results if not r.passed]
235+
236+
return {
237+
"compliance_score": round(score, 2),
238+
"compliance_grade": ComplianceMetrics.get_compliance_grade(score),
239+
"total_tests": total,
240+
"tests_passed": passed,
241+
"tests_failed": failed,
242+
"pass_rate": f"{score:.1f}%",
243+
"severity_distribution": ComplianceMetrics.calculate_severity_score(
244+
failed_tests
245+
),
246+
"weighted_score": round(
247+
ComplianceMetrics.calculate_severity_weighted_score(
248+
total, passed, failed_tests
249+
),
250+
2,
251+
),
252+
"rfc_section_compliance": ComplianceMetrics.get_rfc_section_compliance(
253+
results
254+
),
255+
}
256+
257+
102258
class TestRunner:
103259
def __init__(self, config: TestConfiguration):
104260
self.config = config
@@ -722,6 +878,8 @@ def get_summary(self) -> Dict[str, Any]:
722878
else:
723879
by_category[cat]["failed"] += 1
724880

881+
compliance_report = ComplianceMetrics.generate_compliance_report(self.results)
882+
725883
return {
726884
"total": total,
727885
"passed": passed,
@@ -730,25 +888,51 @@ def get_summary(self) -> Dict[str, Any]:
730888
"by_category": by_category,
731889
"target": f"{self.config.target_host}:{self.config.target_port}",
732890
"source_as": self.config.source_as,
891+
"compliance": compliance_report,
733892
}
734893

735894
def generate_report(self) -> str:
736895
summary = self.get_summary()
896+
compliance = summary.get("compliance", {})
737897
lines = [
738898
"=" * 80,
739-
"BGPv4 Adversarial Test Report",
899+
"BGPv4 RFC Compliance Test Report",
740900
"=" * 80,
741901
f"Target: {summary['target']}",
742902
f"Source AS: {summary['source_as']}",
743903
"-" * 80,
904+
"Compliance Summary:",
905+
f" Compliance Score: {compliance.get('compliance_score', 0):.2f}%",
906+
f" Compliance Grade: {compliance.get('compliance_grade', 'N/A')}",
907+
f" Weighted Score: {compliance.get('weighted_score', 0):.2f}%",
908+
"-" * 80,
744909
f"Total Tests: {summary['total']}",
745910
f"Passed: {summary['passed']}",
746911
f"Failed: {summary['failed']}",
747912
f"Pass Rate: {summary['pass_rate']}",
748-
"-" * 80,
749-
"Results by Category:",
750913
]
751914

915+
severity = compliance.get("severity_distribution", {})
916+
if severity:
917+
lines.append("-" * 80)
918+
lines.append("Severity Distribution of Failures:")
919+
for level in ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]:
920+
count = severity.get(level, 0)
921+
if count > 0:
922+
lines.append(f" {level}: {count}")
923+
924+
lines.append("-" * 80)
925+
lines.append("RFC Section Compliance:")
926+
rfc_compliance = compliance.get("rfc_section_compliance", {})
927+
for section, stats in rfc_compliance.items():
928+
if stats["total"] > 0:
929+
lines.append(
930+
f" {section}: {stats['score']:.1f}% (Grade: {stats['grade']}) "
931+
f"[{stats['passed']}/{stats['total']}]"
932+
)
933+
934+
lines.append("-" * 80)
935+
lines.append("Results by Category:")
752936
for cat, stats in summary["by_category"].items():
753937
lines.append(f" {cat}: {stats['passed']}/{stats['total']} passed")
754938

tests/functional/test_runner.py

Lines changed: 140 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,12 @@
44

55
from unittest.mock import Mock, patch
66

7-
from bgp_test_framework.runner import TestRunner, TestConfiguration, BGPLogger
7+
from bgp_test_framework.runner import (
8+
TestRunner,
9+
TestConfiguration,
10+
BGPLogger,
11+
ComplianceMetrics,
12+
)
813
from bgp_test_framework.tests import TestCategory, TestResult
914

1015

@@ -230,7 +235,9 @@ def test_generate_report(self):
230235
),
231236
]
232237
report = runner.generate_report()
233-
assert "BGPv4 Adversarial Test Report" in report
238+
assert "BGPv4 RFC Compliance Test Report" in report
239+
assert "Compliance Score:" in report
240+
assert "Compliance Grade:" in report
234241
assert "MH-001" in report
235242
assert "PASS" in report
236243

@@ -302,3 +309,134 @@ def test_test_filtering_by_id(self):
302309
filtered = [(tid, cat) for tid, cat in all_tests if tid in target_ids]
303310
assert len(filtered) == 2
304311
assert all(tid in target_ids for tid, _ in filtered)
312+
313+
314+
class TestComplianceMetrics:
315+
def test_calculate_compliance_score(self):
316+
assert ComplianceMetrics.calculate_compliance_score(100, 95) == 95.0
317+
assert ComplianceMetrics.calculate_compliance_score(10, 10) == 100.0
318+
assert ComplianceMetrics.calculate_compliance_score(10, 0) == 0.0
319+
assert ComplianceMetrics.calculate_compliance_score(0, 0) == 0.0
320+
321+
def test_get_compliance_grade(self):
322+
assert ComplianceMetrics.get_compliance_grade(100) == "A"
323+
assert ComplianceMetrics.get_compliance_grade(95) == "A"
324+
assert ComplianceMetrics.get_compliance_grade(94) == "B"
325+
assert ComplianceMetrics.get_compliance_grade(85) == "B"
326+
assert ComplianceMetrics.get_compliance_grade(84) == "C"
327+
assert ComplianceMetrics.get_compliance_grade(70) == "C"
328+
assert ComplianceMetrics.get_compliance_grade(69) == "D"
329+
assert ComplianceMetrics.get_compliance_grade(50) == "D"
330+
assert ComplianceMetrics.get_compliance_grade(49) == "F"
331+
assert ComplianceMetrics.get_compliance_grade(0) == "F"
332+
333+
def test_get_severity_level(self):
334+
assert ComplianceMetrics.get_severity_level("MH-001") == "CRITICAL"
335+
assert ComplianceMetrics.get_severity_level("OM-001") == "HIGH"
336+
assert ComplianceMetrics.get_severity_level("UM-001") == "HIGH"
337+
assert ComplianceMetrics.get_severity_level("AT-001") == "MEDIUM"
338+
assert ComplianceMetrics.get_severity_level("FSM-001") == "HIGH"
339+
assert ComplianceMetrics.get_severity_level("TM-001") == "MEDIUM"
340+
assert ComplianceMetrics.get_severity_level("SEC-001") == "CRITICAL"
341+
assert ComplianceMetrics.get_severity_level("RA-001") == "MEDIUM"
342+
assert ComplianceMetrics.get_severity_level("DEC-001") == "LOW"
343+
assert ComplianceMetrics.get_severity_level("OTHER-001") == "INFO"
344+
345+
def test_calculate_severity_score(self):
346+
failed = [
347+
TestResult(
348+
"MH-001",
349+
"Test 1",
350+
TestCategory.MESSAGE_HEADER,
351+
False,
352+
"Expected",
353+
"Actual",
354+
),
355+
TestResult(
356+
"SEC-001",
357+
"Test 2",
358+
TestCategory.SECURITY,
359+
False,
360+
"Expected",
361+
"Actual",
362+
),
363+
]
364+
severity = ComplianceMetrics.calculate_severity_score(failed)
365+
assert severity["CRITICAL"] == 2
366+
assert severity["HIGH"] == 0
367+
assert severity["MEDIUM"] == 0
368+
369+
def test_calculate_severity_weighted_score(self):
370+
failed = [
371+
TestResult(
372+
"MH-001",
373+
"Test 1",
374+
TestCategory.MESSAGE_HEADER,
375+
False,
376+
"Expected",
377+
"Actual",
378+
),
379+
]
380+
score = ComplianceMetrics.calculate_severity_weighted_score(10, 9, failed)
381+
assert score > 0 and score <= 100
382+
383+
def test_get_rfc_section_compliance(self):
384+
results = [
385+
TestResult(
386+
"MH-001",
387+
"Test 1",
388+
TestCategory.MESSAGE_HEADER,
389+
True,
390+
"Expected",
391+
"Actual",
392+
),
393+
TestResult(
394+
"MH-002",
395+
"Test 2",
396+
TestCategory.MESSAGE_HEADER,
397+
False,
398+
"Expected",
399+
"Actual",
400+
),
401+
]
402+
compliance = ComplianceMetrics.get_rfc_section_compliance(results)
403+
assert "RFC 4271 Section 4.1" in compliance
404+
assert compliance["RFC 4271 Section 4.1"]["total"] == 2
405+
assert compliance["RFC 4271 Section 4.1"]["passed"] == 1
406+
assert compliance["RFC 4271 Section 4.1"]["failed"] == 1
407+
408+
def test_generate_compliance_report(self):
409+
results = [
410+
TestResult(
411+
"MH-001",
412+
"Test 1",
413+
TestCategory.MESSAGE_HEADER,
414+
True,
415+
"Expected",
416+
"Actual",
417+
),
418+
TestResult(
419+
"MH-002",
420+
"Test 2",
421+
TestCategory.MESSAGE_HEADER,
422+
False,
423+
"Expected",
424+
"Actual",
425+
),
426+
TestResult(
427+
"OM-001",
428+
"Test 3",
429+
TestCategory.OPEN_MESSAGE,
430+
True,
431+
"Expected",
432+
"Actual",
433+
),
434+
]
435+
report = ComplianceMetrics.generate_compliance_report(results)
436+
assert report["total_tests"] == 3
437+
assert report["tests_passed"] == 2
438+
assert report["tests_failed"] == 1
439+
assert report["compliance_score"] > 0
440+
assert "compliance_grade" in report
441+
assert "severity_distribution" in report
442+
assert "rfc_section_compliance" in report

0 commit comments

Comments
 (0)