Benjamin Admin
f4357a2e9b
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>
94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
"""Tests für Evidence-Vault."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_vault(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("EVIDENCE_VAULT_ROOT", str(tmp_path))
|
|
import compliance.services.specialist_agents._evidence_vault as v
|
|
yield v
|
|
|
|
|
|
def test_open_vault_creates_structure(tmp_vault):
|
|
vault = tmp_vault.open_vault("impressum", "2.0")
|
|
assert vault.root.exists()
|
|
for sub in ("screenshots", "videos", "csv", "findings", "raw"):
|
|
assert (vault.root / sub).is_dir()
|
|
assert vault.manifest_path.exists()
|
|
|
|
|
|
def test_put_bytes_appends_manifest(tmp_vault):
|
|
vault = tmp_vault.open_vault("impressum", "2.0")
|
|
rel = vault.put_bytes("screenshot", "url1", "test.png",
|
|
b"\x89PNG\r\n\x1a\n", mime="image/png")
|
|
assert rel.startswith("screenshots/")
|
|
assert (vault.root / rel).exists()
|
|
assets = vault.list_assets()
|
|
assert len(assets) == 1
|
|
assert assets[0]["sha256"]
|
|
assert assets[0]["mime"] == "image/png"
|
|
assert assets[0]["size_bytes"] == 8
|
|
|
|
|
|
def test_put_json_stores_finding(tmp_vault):
|
|
vault = tmp_vault.open_vault("cookie_policy", "1.0")
|
|
rel = vault.put_json("finding", "url1", "output.json",
|
|
{"findings": [{"check_id": "X"}]})
|
|
p = vault.root / rel
|
|
data = json.loads(p.read_text())
|
|
assert data["findings"][0]["check_id"] == "X"
|
|
|
|
|
|
def test_assets_for_slot_isolation(tmp_vault):
|
|
vault = tmp_vault.open_vault("agent", "1.0")
|
|
vault.put_bytes("screenshot", "url1", "a.png", b"a")
|
|
vault.put_bytes("screenshot", "url2", "b.png", b"b")
|
|
vault.put_bytes("video", "url1", "w.mp4", b"v")
|
|
assert len(vault.assets_for_slot("url1")) == 2
|
|
assert len(vault.assets_for_slot("url2")) == 1
|
|
|
|
|
|
def test_asset_path_blocks_traversal(tmp_vault):
|
|
vault = tmp_vault.open_vault("agent", "1.0")
|
|
p = vault.asset_path("../../../etc/passwd")
|
|
assert p is None
|
|
|
|
|
|
def test_finalize_writes_finished_at(tmp_vault):
|
|
vault = tmp_vault.open_vault("agent", "1.0")
|
|
snap = vault.finalize()
|
|
assert "finished_at" in snap
|
|
manifest = json.loads(vault.manifest_path.read_text())
|
|
assert "finished_at" in manifest
|
|
|
|
|
|
def test_list_runs_returns_recent(tmp_vault):
|
|
tmp_vault.open_vault("a", "1.0", run_id="run1")
|
|
tmp_vault.open_vault("b", "1.0", run_id="run2")
|
|
runs = tmp_vault.list_runs(limit=10)
|
|
ids = {r["run_id"] for r in runs}
|
|
assert {"run1", "run2"} <= ids
|
|
|
|
|
|
def test_delete_run_removes_dir(tmp_vault):
|
|
vault = tmp_vault.open_vault("a", "1.0", run_id="kill-me")
|
|
vault.put_bytes("screenshot", "u", "x.png", b"x")
|
|
assert tmp_vault.delete_run("kill-me")
|
|
assert not vault.root.exists()
|
|
assert not tmp_vault.delete_run("kill-me") # idempotent
|
|
|
|
|
|
def test_safe_filename_strips_path_chars(tmp_vault):
|
|
vault = tmp_vault.open_vault("a", "1.0")
|
|
rel = vault.put_bytes("raw", "slot",
|
|
"../../etc/passwd", b"x")
|
|
assert "passwd" in rel
|
|
assert ".." not in rel
|