7b94512abc
Feedback rules now support value/field/entity/stream scope, require expiry for broad suppressions, and can be revoked individually. Adds an active-rules panel, detector precision metrics, and diagnostics.
53 lines
2.3 KiB
Python
53 lines
2.3 KiB
Python
import unittest
|
|
|
|
from fgai.metrics import prometheus_metrics
|
|
|
|
|
|
class MetricsTests(unittest.TestCase):
|
|
def test_renders_low_cardinality_metrics(self):
|
|
output = prometheus_metrics({"summary": {"total": 10}, "anomaly_summary": {"high": 2}, "baseline": {"sources_ready": 3}, "capabilities": {"graylog_mcp": {"status": "connected", "events_fetched": 9}}})
|
|
self.assertIn('fgai_events_total{kind="total"} 10', output)
|
|
self.assertIn("fgai_graylog_mcp_connected 1", output)
|
|
|
|
def test_feedback_rules_counted_by_scope(self):
|
|
status = {
|
|
"feedback": [
|
|
{"scope": "entity", "status": "expected"},
|
|
{"scope": "value", "status": "confirmed"},
|
|
],
|
|
"field_deviations": {},
|
|
}
|
|
text = prometheus_metrics(status)
|
|
self.assertIn('fgai_feedback_rules{scope="entity"} 1', text)
|
|
self.assertIn('fgai_feedback_rules{scope="value"} 1', text)
|
|
self.assertIn('fgai_feedback_rules{scope="field"} 0', text)
|
|
self.assertIn('fgai_feedback_rules{scope="stream"} 0', text)
|
|
|
|
def test_suppressed_findings_and_detector_precision(self):
|
|
status = {
|
|
"feedback": [],
|
|
"field_deviations": {
|
|
"alice": [
|
|
{"detector": "rare_value", "feedback": "confirmed"},
|
|
{"detector": "rare_value", "feedback": "false_positive"},
|
|
{"detector": "rare_value", "feedback": "expected"},
|
|
{"detector": "auth_failure_burst", "feedback": "unreviewed"},
|
|
]
|
|
},
|
|
}
|
|
text = prometheus_metrics(status)
|
|
self.assertIn("fgai_feedback_suppressed_findings 2", text)
|
|
self.assertIn('fgai_detector_feedback_total{detector="rare_value",verdict="confirmed"} 1', text)
|
|
self.assertIn('fgai_detector_feedback_total{detector="rare_value",verdict="false_positive"} 1', text)
|
|
self.assertIn('fgai_detector_precision{detector="rare_value"} 0.5000', text)
|
|
self.assertNotIn('detector="auth_failure_burst"', text)
|
|
|
|
def test_metrics_handle_empty_status(self):
|
|
text = prometheus_metrics({})
|
|
self.assertIn('fgai_feedback_rules{scope="value"} 0', text)
|
|
self.assertIn("fgai_feedback_suppressed_findings 0", text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|