e8ff75cbfe
5 Backlog-Items aus dem Multi-Site-Briefing in einem Sprint:
1. B13 B2C-Soft-Hints — Versicherungs/Tarif/Buchungs-Marker
_B2C_WEAK erweitert um "Reiseversicherung", "Tarifrechner",
"Online-Antrag", "Flug buchen", "Stromtarif" etc.
Fängt Allianz-Reise-Chatbot (vorher False-Negative).
2. Chatbot-Policy-Discovery (chatbot_policy_discovery.py)
Probt 14 Standard-Slugs (privacypolicychatbot, chatbot-datenschutz,
ai-policy, ki-datenschutz, ...) × 5 Lang-Prefixe auf jeder
submitted Origin. Successful >300-Wort-Findings werden in
doc_texts['dse'] gemerged. Audit-Trail über
doc_entries[dse].chatbot_policy_sources.
Hebt Westfield-iAdvize-Lücke.
3. API-Response-Payload erweitert
phase_f_persist.response um extra_findings, audit_walk und
html_blocks erweitert. B-Wiring-Output (B1, B3-B18) ist nicht
mehr nur im Mail-HTML versteckt — externe Aufrufer sehen jeden
Finding. Schema additiv, legacy clients ignorieren neue Felder.
4. Plausibility-LLM Empty-Response-Fix
Resilienz-Strategie A→B→C→D:
A) format='json' (strict, default)
B) format='' (loose, _try_extract_json mit ```json-fence + prose-
wrap-Unterstützung)
C) Split-Batch-Recursion (vorhanden)
D) Give up, leeres dict (callers behandeln als skipped)
Plus _post_llm() als isolierter LLM-Call-Helper, catched
Network-Errors.
5. Specialist-Agents Phase 2 LLM (MVP) — Impressum-Agent
impressum_agent_llm.py: qwen3:30b-a3b mit § 5 TMG System-Prompt,
business_scope-hints aus profile_dict. Output identisches Schema
wie pattern-agent für ein Merge ohne API-Bruch.
_b18_wiring.py orchestriert beide Agents + deduplet nach
field_id, rendert lila V2-Block mit KB/LLM-Tags pro Finding.
Pattern-first im Dedup (deterministisch + stable).
Tests: 107/107 grün (7 Test-Suites + chatbot-discovery + b18).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
"""Tests for B18 Impressum-Specialist-Agent (Pattern + LLM)."""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from compliance.api.agent_check._b18_wiring import _render, run_b18
|
|
from compliance.services.specialist_agents.impressum_agent_llm import (
|
|
_parse_response,
|
|
)
|
|
|
|
|
|
_GOOD_IMPRESSUM = """
|
|
Acme GmbH
|
|
Musterstraße 1
|
|
10115 Berlin
|
|
|
|
Handelsregister: HRB 12345 Berlin
|
|
USt-IdNr: DE123456789
|
|
Geschäftsführer: Max Mustermann
|
|
|
|
Telefon: +49 30 12345
|
|
E-Mail: info@acme.example
|
|
"""
|
|
|
|
_BAD_IMPRESSUM = (
|
|
"Acme GmbH, Musterstraße 1, 10115 Berlin. "
|
|
"Kontakt: info@acme.example. "
|
|
"Wir freuen uns ueber Ihren Besuch auf unserer Website "
|
|
"und ueber Ihr Interesse an unserem Unternehmen und unseren "
|
|
"Produkten. Bitte beachten Sie auch unsere weiteren Hinweise."
|
|
)
|
|
|
|
|
|
class TestParseResponse:
|
|
def test_pure_json(self):
|
|
out = _parse_response('{"findings":[{"field_id":"foo","severity":"HIGH"}]}')
|
|
assert len(out) == 1
|
|
assert out[0]["field_id"] == "foo"
|
|
|
|
def test_markdown_fenced_json(self):
|
|
out = _parse_response('```json\n{"findings":[{"field_id":"x"}]}\n```')
|
|
assert len(out) == 1
|
|
|
|
def test_prose_wrapped(self):
|
|
out = _parse_response(
|
|
'Hier ist die Analyse: {"findings":[{"field_id":"y"}]} Ende.'
|
|
)
|
|
assert len(out) == 1
|
|
|
|
def test_empty(self):
|
|
assert _parse_response("") == []
|
|
|
|
def test_garbage(self):
|
|
assert _parse_response("not json at all") == []
|
|
|
|
|
|
class TestRunB18Wiring:
|
|
def test_short_impressum_skipped(self):
|
|
state = {"doc_texts": {"impressum": "tiny"}}
|
|
asyncio.run(run_b18(state))
|
|
assert "impressum_agent_html" not in state
|
|
|
|
def test_no_impressum_skipped(self):
|
|
asyncio.run(run_b18({"doc_texts": {}}))
|
|
|
|
def test_merges_pattern_and_llm(self):
|
|
# Pattern-agent will likely find no gaps in _GOOD_IMPRESSUM.
|
|
# Mock the LLM to return a fake additional finding.
|
|
async def fake_llm(text, scope):
|
|
return [{
|
|
"check_id": "IMPRESSUM-AGENT-LLM-DPO",
|
|
"agent": "impressum_agent_v2_llm",
|
|
"field_id": "dpo",
|
|
"severity": "MEDIUM",
|
|
"title": "DSB-Verweis fehlt",
|
|
"norm": "§ 5 TMG / DDG (LLM)",
|
|
"evidence": "kein Hinweis auf DSB",
|
|
"action": "DSB im Impressum verlinken",
|
|
}]
|
|
with patch(
|
|
"compliance.api.agent_check._b18_wiring.evaluate_llm",
|
|
new=fake_llm,
|
|
):
|
|
state = {"doc_texts": {"impressum": _GOOD_IMPRESSUM},
|
|
"profile_dict": {}}
|
|
asyncio.run(run_b18(state))
|
|
assert "impressum_agent_html" in state
|
|
extras = state.get("extra_findings") or []
|
|
ids = [f.get("check_id") for f in extras]
|
|
assert any("LLM-DPO" in i for i in ids)
|
|
|
|
def test_dedup_pattern_vs_llm_same_field(self):
|
|
# Pattern agent returns ust_id; mocked LLM also returns ust_id —
|
|
# only one should survive the dedup.
|
|
async def fake_llm(text, scope):
|
|
return [{
|
|
"check_id": "IMPRESSUM-AGENT-LLM-UST_ID",
|
|
"agent": "impressum_agent_v2_llm",
|
|
"field_id": "ust_id",
|
|
"severity": "HIGH",
|
|
"title": "duplicate ust_id finding",
|
|
"norm": "§ 5 TMG",
|
|
"evidence": "—",
|
|
"action": "—",
|
|
}]
|
|
with patch(
|
|
"compliance.api.agent_check._b18_wiring.evaluate_llm",
|
|
new=fake_llm,
|
|
):
|
|
state = {"doc_texts": {"impressum": _BAD_IMPRESSUM},
|
|
"profile_dict": {}}
|
|
asyncio.run(run_b18(state))
|
|
ust_findings = [
|
|
f for f in state.get("extra_findings") or []
|
|
if (f.get("field_id") or "").lower() == "ust_id"
|
|
]
|
|
assert len(ust_findings) == 1
|
|
|
|
|
|
class TestRender:
|
|
def test_render_with_two_findings(self):
|
|
merged = [
|
|
{"check_id": "X", "title": "A", "severity": "HIGH",
|
|
"agent": "impressum_agent_v1", "norm": "n", "action": "do"},
|
|
{"check_id": "Y", "title": "B", "severity": "MEDIUM",
|
|
"agent": "impressum_agent_v2_llm", "norm": "n", "action": "do"},
|
|
]
|
|
html = _render(merged, merged[:1], merged[1:])
|
|
assert "KB" in html # pattern tag
|
|
assert "LLM" in html # llm tag
|
|
assert "Pattern-Match: 1" in html
|
|
assert "LLM-Analyse: 1" in html
|