feat(dse): obligation shadow telemetry

Verdrahtet die Obligation Aggregation Engine als Layer 4 (SHADOW) in v3_engine: erzeugt aus den results zusätzlich Obligation-Ergebnisse AUSSCHLIESSLICH für die Telemetrie. Greift NICHT in results ein — nutzer-sichtbare Findings unverändert. - _obligation_shadow.py: fetch_obligation_markers (legal_obligations + applicability) + compute_obligation_shadow (pure): legacy_control_findings, obligation_shadow_results, collapse_factor, na_count, met_failed_delta, top_collapsed_obligations - met-Signal = Legacy-passed (kein zusätzlicher Prüfer-Call/Key) E2E (3 Firmen, echte Engine): 57 Control-Findings → 14 Obligation-Findings (4,1×); Redundanz kollabiert wo Evidenz existiert, echte Lücken bleiben FAILED. 6 Unit-Tests grün. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-24 12:59:52 +02:00
parent 7ec29999a2
commit c3542f7dfe
3 changed files with 173 additions and 0 deletions
@@ -0,0 +1,52 @@
+"""Unit-Tests für die DSE Shadow-Verdrahtung (compute_obligation_shadow, pure)."""
+from compliance.services.specialist_agents.dse._obligation_shadow import (
+    compute_obligation_shadow,
+)
+
+
+def _markers(n, ob, cond=None):
+    return {f"C{i}": {"obl": [ob], "cond": cond} for i in range(n)}
+
+
+class TestComputeShadow:
+    def test_collapse_and_delta(self):
+        results = [{"control_id": f"C{i}", "passed": False} for i in range(5)]
+        s = compute_obligation_shadow(results, "x", _markers(5, "recipients_disclosed"))
+        assert s["legacy_control_findings"] == 5
+        assert s["obligation_findings"] == 1          # 5 → 1
+        assert s["collapse_factor"] == 5.0
+        assert s["met_failed_delta"] == 4
+        top = s["top_collapsed_obligations"][0]
+        assert top["obligation"] == "recipients_disclosed" and top["fehlt"] == 5
+
+    def test_fp_correction_one_passed_collapses_to_met(self):
+        results = [{"control_id": f"C{i}", "passed": i == 0} for i in range(5)]
+        s = compute_obligation_shadow(results, "x", _markers(5, "recipients_disclosed"))
+        assert s["legacy_control_findings"] == 4
+        assert s["obligation_findings"] == 0          # MET (anderswo erfüllt)
+        assert s["met_failed_delta"] == 4
+
+    def test_na_when_predicate_false(self):
+        results = [{"control_id": "C0", "passed": False}]
+        m = {"C0": {"obl": ["third_country_transfer_disclosed"],
+                    "cond": "has_third_country_transfer"}}
+        s = compute_obligation_shadow(results, "nur innerhalb der eu", m)
+        assert s["na_count"] == 1
+        assert s["obligation_findings"] == 0          # NA statt FEHLT
+
+    def test_na_predicate_true_keeps_finding(self):
+        results = [{"control_id": "C0", "passed": False}]
+        m = {"C0": {"obl": ["third_country_transfer_disclosed"],
+                    "cond": "has_third_country_transfer"}}
+        s = compute_obligation_shadow(results, "übermittlung in ein drittland", m)
+        assert s["na_count"] == 0
+        assert s["obligation_findings"] == 1
+
+    def test_no_markers_returns_status(self):
+        s = compute_obligation_shadow([{"control_id": "C0", "passed": False}], "x", {})
+        assert "no obligation" in s["status"]
+
+    def test_does_not_mutate_results(self):
+        results = [{"control_id": "C0", "passed": False}]
+        compute_obligation_shadow(results, "x", _markers(1, "recipients_disclosed"))
+        assert set(results[0].keys()) == {"control_id", "passed"}   # unverändert