feat(agents): Specialist-Agents Phase 2 Foundation + Cookie-Policy-Agent

Sprint 1 — Foundation (User-Vorgabe 2026-06-08): Foundation: - _base.py: BaseSpecialistAgent ABC + Pydantic Contract (AgentInput/AgentOutput/Finding/Recommendation/McCoverage/EscalationLog). - _base.lint_output(): Disclaimer-Linter verbietet "rechtssicher" / "garantiert" / "gesetzeskonform" — scrubbed inline + Log in notes. - _registry.py: AgentRegistry mit MC-Owner-Mapping (verhindert Doppel-Ownership). - _escalation.py: cascade(local → ovh). qwen2.5:7b default, OVH 120b als Stage-2 (deaktiviert wenn OVH_URL leer). - _rollup.py: deterministisches Dedup ähnlicher actions zu Recommendations mit related_finding_ids[]. - _evidence_vault.py: Pro-Run File-Vault für Playwright-Videos, Screenshots, CSV. SHA256 + manifest.json. DSR-tauglich (delete_run). Agenten: - ImpressumAgent v2 (impressum/agent.py + mcs.py) — konsolidiert v1-Pattern-Match + v2-LLM-MVP unter dem neuen Contract. 12 MCs. - CookiePolicyAgent v1 (cookie_policy/agent.py + mcs.py) — 12 MCs zu Cookie-Richtlinie-Vollständigkeit + KB-Layer für CMP-Vendor-Cross-Check. Tests: 25/25 grün (10 Impressum + 9 Vault + 6 Cookie-Policy). Roadmap: SSE-Test-Endpoint + Frontend-Tab → DSE/AGB-Agents → Cookie-Banner-Themen-Agent → Cross-Doc-Konsistenz-Agent. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-08 17:40:05 +02:00
parent d6b8bf87c2
commit f4357a2e9b
15 changed files with 2364 additions and 10 deletions
@@ -0,0 +1,185 @@
+"""Tests für Impressum-Agent v2 (BaseSpecialistAgent)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from compliance.services.specialist_agents import (
+    REGISTRY,
+    AgentInput,
+    AgentOutput,
+    ImpressumAgent,
+    Severity,
+)
+from compliance.services.specialist_agents._base import (
+    FORBIDDEN_OUTPUT_TERMS,
+    lint_output,
+    stable_recommendation_id,
+)
+from compliance.services.specialist_agents._rollup import rollup
+
+
+TESLA_IMPRESSUM = (
+    "Tesla Germany GmbH\n"
+    "Ludwig-Prandtl-Strasse 25-29\n"
+    "12526 Berlin\n"
+    "Deutschland\n\n"
+    "Email: kontakt@tesla.com\n"
+    "Telefon: +49 89 1250 16 800\n\n"
+    "Management:\n"
+    "Elon Musk\n\n"
+    "Handelsregister: HRB 218904 B, Amtsgericht Charlottenburg\n"
+)
+
+FULL_IMPRESSUM = (
+    TESLA_IMPRESSUM
+    + "\nUSt-IdNr: DE123456789\nGeschäftsführer: Max Mustermann\n"
+)
+
+
+def _run(coro):
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+
+def test_agent_is_registered():
+    agent = REGISTRY.get("impressum")
+    assert agent is not None
+    assert agent.doc_type == "impressum"
+    assert len(agent.owned_mc_ids) >= 10
+
+
+def test_short_text_skipped():
+    agent = ImpressumAgent()
+    out = _run(agent.evaluate(AgentInput(doc_type="impressum", text="x")))
+    assert out.mc_total > 0
+    assert all(c.status == "skipped" for c in out.mc_coverage)
+    assert not out.findings
+
+
+def test_tesla_missing_german_label(monkeypatch):
+    # Skip LLM escalation for unit test (no Ollama in CI)
+    async def _no_cascade(*a, **kw):
+        return None, []
+    monkeypatch.setattr(
+        "compliance.services.specialist_agents.impressum.agent.cascade",
+        _no_cascade,
+    )
+    agent = ImpressumAgent()
+    out = _run(agent.evaluate(AgentInput(
+        doc_type="impressum", text=TESLA_IMPRESSUM,
+    )))
+    field_ids = {f.field_id for f in out.findings}
+    # Tesla pattern: "Management:" matches IMP-MC-006 → present
+    # But IMP-MC-007 (deutsches Label) MUSS fehlen
+    assert "vertretungsberechtigte_label_korrekt" in field_ids
+    # USt fehlt
+    assert "ust_id" in field_ids
+
+
+def test_full_impressum_has_no_basic_findings(monkeypatch):
+    async def _no_cascade(*a, **kw):
+        return None, []
+    monkeypatch.setattr(
+        "compliance.services.specialist_agents.impressum.agent.cascade",
+        _no_cascade,
+    )
+    agent = ImpressumAgent()
+    out = _run(agent.evaluate(AgentInput(
+        doc_type="impressum", text=FULL_IMPRESSUM,
+    )))
+    # nur scope-dependent fields fehlen (vsbg, odr, redaktion)
+    high = [f for f in out.findings if f.severity == Severity.HIGH.value]
+    assert not high, f"unexpected HIGH findings: {[f.field_id for f in high]}"
+
+
+def test_b2c_scope_adds_vsbg(monkeypatch):
+    async def _no_cascade(*a, **kw):
+        return None, []
+    monkeypatch.setattr(
+        "compliance.services.specialist_agents.impressum.agent.cascade",
+        _no_cascade,
+    )
+    agent = ImpressumAgent()
+    out = _run(agent.evaluate(AgentInput(
+        doc_type="impressum", text=TESLA_IMPRESSUM,
+        business_scope=["b2c", "ecommerce"],
+    )))
+    field_ids = {f.field_id for f in out.findings}
+    assert "verbraucher_streitbeilegung" in field_ids
+    assert "odr_link" in field_ids
+
+
+def test_automotive_scope_auto_detected(monkeypatch):
+    async def _no_cascade(*a, **kw):
+        return None, []
+    monkeypatch.setattr(
+        "compliance.services.specialist_agents.impressum.agent.cascade",
+        _no_cascade,
+    )
+    agent = ImpressumAgent()
+    out = _run(agent.evaluate(AgentInput(
+        doc_type="impressum", text=TESLA_IMPRESSUM,
+    )))
+    field_ids = {f.field_id for f in out.findings}
+    assert "aufsichtsbehoerde" in field_ids
+    # Action MUSS KBA-Hint enthalten
+    aufsicht = next(f for f in out.findings
+                    if f.field_id == "aufsichtsbehoerde")
+    assert "KBA" in aufsicht.action or "Kraftfahrt" in aufsicht.action
+
+
+def test_disclaimer_linter_scrubs_forbidden():
+    from compliance.services.specialist_agents._base import Finding
+    from datetime import datetime, timezone
+    f = Finding(
+        check_id="X", agent="t", agent_version="1",
+        severity=Severity.HIGH,
+        title="Diese Lösung ist rechtssicher und garantiert konform",
+        action="Voll konform machen",
+    )
+    out = AgentOutput(
+        agent="t", agent_version="1",
+        started_at=datetime.now(timezone.utc),
+        finished_at=datetime.now(timezone.utc),
+        duration_ms=0, findings=[f],
+    )
+    cleaned = lint_output(out)
+    assert "rechtssicher" not in cleaned.findings[0].title.lower()
+    assert "garantiert" not in cleaned.findings[0].title.lower()
+    assert "linter scrubbed" in cleaned.notes.lower()
+
+
+def test_rollup_bundles_same_action():
+    from compliance.services.specialist_agents._base import Finding
+    fs = [
+        Finding(check_id="A", agent="t", agent_version="1",
+                severity=Severity.HIGH,
+                title="Lücke 1", action="AVV mit Anbieter X abschließen"),
+        Finding(check_id="B", agent="t", agent_version="1",
+                severity=Severity.MEDIUM,
+                title="Lücke 2", action="AVV mit Anbieter X abschließen."),
+        Finding(check_id="C", agent="t", agent_version="1",
+                severity=Severity.LOW,
+                title="Lücke 3", action="Etwas anderes machen"),
+    ]
+    recs = rollup(fs)
+    assert len(recs) == 2
+    bundled = next(r for r in recs if len(r.related_finding_ids) == 2)
+    assert bundled.severity == Severity.HIGH.value
+    assert set(bundled.related_finding_ids) == {"A", "B"}
+
+
+def test_stable_recommendation_id_is_deterministic():
+    a = stable_recommendation_id("AVV mit Anbieter X abschließen")
+    b = stable_recommendation_id("avv mit anbieter x abschliessen")
+    # case insensitive aber Diakritika strict (deutsch ß ≠ ss)
+    assert len(a) == 16
+    assert len(b) == 16
+
+
+def test_forbidden_terms_complete():
+    """Sanity-Test, dass alle wichtigen Wörter im Linter sind."""
+    for term in ("rechtssicher", "garantiert", "gesetzeskonform"):
+        assert any(term in t for t in FORBIDDEN_OUTPUT_TERMS)