464 lines
18 KiB
Python
464 lines
18 KiB
Python
"""Unit-Tests für die ContextBridge (Kontextbrücke).
|
|
|
|
Testet:
|
|
- Sensitive-Content-Erkennung (Credentials, Endpoints, PII)
|
|
- Freigabe-Ablauf mit Nutzerbestätigung
|
|
- Verweigerung bei sensitivem Inhalt
|
|
- Widerruf von Freigaben
|
|
- Update-Propagierung
|
|
- Registry-Persistierung
|
|
|
|
Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from monorepo.bridge import ContextBridge, SensitiveMatch, ShareResult, SharedArtifactEntry
|
|
from monorepo.knowledge.index import IndexEntry, YAMLIndex
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_knowledge_base(tmp_path: Path) -> Path:
|
|
"""Erstellt einen temporären Wissensspeicher mit Test-Artefakten."""
|
|
kb_path = tmp_path / "knowledge-store"
|
|
kb_path.mkdir()
|
|
return kb_path
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_artifact_file(tmp_knowledge_base: Path) -> Path:
|
|
"""Erstellt eine Test-Artefakt-Datei ohne sensitive Inhalte."""
|
|
artifact_dir = tmp_knowledge_base / "bahn" / "decision"
|
|
artifact_dir.mkdir(parents=True)
|
|
artifact_file = artifact_dir / "api-design.md"
|
|
artifact_file.write_text(
|
|
"---\n"
|
|
"type: decision\n"
|
|
"title: API Design Entscheidung\n"
|
|
"tags: [api, design]\n"
|
|
"source_context: bahn\n"
|
|
"shareable: true\n"
|
|
"content_hash: sha256:abc123\n"
|
|
"---\n"
|
|
"# API Design\n\n"
|
|
"Wir verwenden REST für die externe API.\n",
|
|
encoding="utf-8",
|
|
)
|
|
return artifact_file
|
|
|
|
|
|
@pytest.fixture
|
|
def sensitive_artifact_file(tmp_knowledge_base: Path) -> Path:
|
|
"""Erstellt eine Test-Artefakt-Datei MIT sensitivem Inhalt."""
|
|
artifact_dir = tmp_knowledge_base / "dhive" / "note"
|
|
artifact_dir.mkdir(parents=True)
|
|
artifact_file = artifact_dir / "service-config.md"
|
|
artifact_file.write_text(
|
|
"---\n"
|
|
"type: note\n"
|
|
"title: Service Konfiguration\n"
|
|
"tags: [config]\n"
|
|
"source_context: dhive\n"
|
|
"shareable: false\n"
|
|
"content_hash: sha256:def456\n"
|
|
"---\n"
|
|
"# Service Config\n\n"
|
|
"api_key: sk-12345abcdef\n"
|
|
"endpoint: https://internal.dhive.example.com/api\n"
|
|
"Kontakt: admin@dhive-internal.de\n",
|
|
encoding="utf-8",
|
|
)
|
|
return artifact_file
|
|
|
|
|
|
@pytest.fixture
|
|
def yaml_index(tmp_knowledge_base: Path, sample_artifact_file: Path) -> YAMLIndex:
|
|
"""Erstellt einen YAML-Index mit einem Test-Eintrag."""
|
|
index = YAMLIndex(tmp_knowledge_base / "_index.yaml")
|
|
index.load()
|
|
|
|
entry = IndexEntry(
|
|
id="bahn/decision/api-design",
|
|
title="API Design Entscheidung",
|
|
type="decision",
|
|
tags=["api", "design"],
|
|
scope="bahn",
|
|
summary="REST API Entscheidung",
|
|
path="bahn/decision/api-design.md",
|
|
content_hash="sha256:abc123",
|
|
)
|
|
index.update_entry(entry)
|
|
return index
|
|
|
|
|
|
@pytest.fixture
|
|
def yaml_index_with_sensitive(
|
|
yaml_index: YAMLIndex, sensitive_artifact_file: Path
|
|
) -> YAMLIndex:
|
|
"""YAML-Index mit einem sensitiven Artefakt."""
|
|
entry = IndexEntry(
|
|
id="dhive/note/service-config",
|
|
title="Service Konfiguration",
|
|
type="note",
|
|
tags=["config"],
|
|
scope="dhive",
|
|
summary="Service-Konfiguration mit Secrets",
|
|
path="dhive/note/service-config.md",
|
|
content_hash="sha256:def456",
|
|
)
|
|
yaml_index.update_entry(entry)
|
|
return yaml_index
|
|
|
|
|
|
@pytest.fixture
|
|
def bridge(tmp_knowledge_base: Path, yaml_index: YAMLIndex, tmp_path: Path) -> ContextBridge:
|
|
"""Erstellt eine ContextBridge-Instanz mit Registry-Pfad."""
|
|
registry_path = tmp_path / "config" / "shared-artifacts.yaml"
|
|
return ContextBridge(
|
|
knowledge_base_path=tmp_knowledge_base,
|
|
index=yaml_index,
|
|
registry_path=registry_path,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def bridge_with_sensitive(
|
|
tmp_knowledge_base: Path, yaml_index_with_sensitive: YAMLIndex, tmp_path: Path
|
|
) -> ContextBridge:
|
|
"""ContextBridge mit sensitiven Artefakten im Index."""
|
|
registry_path = tmp_path / "config" / "shared-artifacts.yaml"
|
|
return ContextBridge(
|
|
knowledge_base_path=tmp_knowledge_base,
|
|
index=yaml_index_with_sensitive,
|
|
registry_path=registry_path,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: check_sensitive_content (Req 5.1, 5.3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCheckSensitiveContent:
|
|
"""Tests für die Sensitive-Content-Erkennung."""
|
|
|
|
def test_no_sensitive_content(self, bridge: ContextBridge) -> None:
|
|
"""Kein sensitiver Inhalt wird als leere Liste erkannt."""
|
|
content = "# API Design\n\nWir verwenden REST.\nDetails folgen."
|
|
result = bridge.check_sensitive_content(content)
|
|
assert result == []
|
|
|
|
def test_detects_api_key(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt API-Key-Muster."""
|
|
content = "config:\n api_key: sk-12345abcdef\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert any(m.pattern_name == "credentials" for m in result)
|
|
|
|
def test_detects_token(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt Token-Muster."""
|
|
content = "Authorization:\n token= ghp_abc123xyz\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert any(m.pattern_name == "credentials" for m in result)
|
|
|
|
def test_detects_password(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt Passwort-Muster."""
|
|
content = "db_password: supersecret123\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert any(m.pattern_name == "credentials" for m in result)
|
|
|
|
def test_detects_secret(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt Secret-Muster."""
|
|
content = "client_secret= abc123\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert any(m.pattern_name == "credentials" for m in result)
|
|
|
|
def test_detects_endpoint_url(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt Endpoint-URL-Muster."""
|
|
content = "Konfiguration:\n endpoint: https://api.internal.example.com/v2\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert any(m.pattern_name == "endpoint" for m in result)
|
|
|
|
def test_detects_email_pii(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt E-Mail-Adressen als PII."""
|
|
content = "Ansprechpartner: max.mustermann@example.de\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert any(m.pattern_name == "pii" for m in result)
|
|
|
|
def test_multiple_sensitive_items(self, bridge: ContextBridge) -> None:
|
|
"""Erkennt mehrere sensitive Elemente."""
|
|
content = (
|
|
"# Config\n"
|
|
"api_key: abc123\n"
|
|
"endpoint: https://internal.example.com/api\n"
|
|
"admin: admin@example.org\n"
|
|
)
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 3
|
|
pattern_names = {m.pattern_name for m in result}
|
|
assert "credentials" in pattern_names
|
|
assert "endpoint" in pattern_names
|
|
assert "pii" in pattern_names
|
|
|
|
def test_line_numbers_correct(self, bridge: ContextBridge) -> None:
|
|
"""Zeilennummern sind korrekt (1-basiert)."""
|
|
content = "Zeile 1\nZeile 2\napi_key: geheim\nZeile 4\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 1
|
|
assert result[0].line_number == 3
|
|
|
|
def test_empty_content(self, bridge: ContextBridge) -> None:
|
|
"""Leerer Inhalt ergibt keine Treffer."""
|
|
result = bridge.check_sensitive_content("")
|
|
assert result == []
|
|
|
|
def test_case_insensitive_credentials(self, bridge: ContextBridge) -> None:
|
|
"""Credential-Erkennung ist case-insensitive."""
|
|
content = "API_KEY: value123\nToken: xyz\nPASSWORD= abc\n"
|
|
result = bridge.check_sensitive_content(content)
|
|
assert len(result) >= 3
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: share_artifact (Req 5.1, 5.2, 5.3, 5.4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestShareArtifact:
|
|
"""Tests für die Artefakt-Freigabe."""
|
|
|
|
def test_share_requires_user_confirmation(self, bridge: ContextBridge) -> None:
|
|
"""Req 5.4: Freigabe ohne Nutzerbestätigung wird verweigert."""
|
|
result = bridge.share_artifact("bahn/decision/api-design", user_confirmed=False)
|
|
assert result.success is False
|
|
assert "Nutzerbestätigung" in result.errors[0]
|
|
|
|
def test_share_nonexistent_artifact(self, bridge: ContextBridge) -> None:
|
|
"""Freigabe eines nicht existierenden Artefakts schlägt fehl."""
|
|
result = bridge.share_artifact("nonexistent/artifact", user_confirmed=True)
|
|
assert result.success is False
|
|
assert "nicht im Index" in result.errors[0]
|
|
|
|
def test_share_clean_artifact_succeeds(self, bridge: ContextBridge) -> None:
|
|
"""Req 5.2: Freigabe eines sauberen Artefakts gelingt."""
|
|
result = bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
assert result.success is True
|
|
assert result.artifact_id == "bahn/decision/api-design"
|
|
assert len(result.shared_to) > 0
|
|
# Bahn-Artefakt sollte in privat, dhive, shared sichtbar sein
|
|
assert "bahn" not in result.shared_to
|
|
assert "privat" in result.shared_to
|
|
assert "dhive" in result.shared_to
|
|
assert "shared" in result.shared_to
|
|
|
|
def test_share_sensitive_artifact_refused(
|
|
self, bridge_with_sensitive: ContextBridge
|
|
) -> None:
|
|
"""Req 5.3: Freigabe eines Artefakts mit sensitivem Inhalt wird verweigert."""
|
|
result = bridge_with_sensitive.share_artifact(
|
|
"dhive/note/service-config", user_confirmed=True
|
|
)
|
|
assert result.success is False
|
|
assert any("sensible Inhalte" in e for e in result.errors)
|
|
|
|
def test_share_marks_as_shared(self, bridge: ContextBridge) -> None:
|
|
"""Nach Freigabe ist das Artefakt als geteilt markiert."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
assert bridge.is_shared("bahn/decision/api-design") is True
|
|
|
|
def test_unshared_artifact_not_marked(self, bridge: ContextBridge) -> None:
|
|
"""Nicht freigegebene Artefakte sind nicht als geteilt markiert."""
|
|
assert bridge.is_shared("bahn/decision/api-design") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: revoke_share (Req 5.6)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRevokeShare:
|
|
"""Tests für den Widerruf der Freigabe."""
|
|
|
|
def test_revoke_removes_from_registry(self, bridge: ContextBridge) -> None:
|
|
"""Req 5.6: Widerruf entfernt Artefakt aus der Registry."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
assert bridge.is_shared("bahn/decision/api-design") is True
|
|
|
|
bridge.revoke_share("bahn/decision/api-design")
|
|
assert bridge.is_shared("bahn/decision/api-design") is False
|
|
|
|
def test_revoke_nonexistent_raises(self, bridge: ContextBridge) -> None:
|
|
"""Widerruf eines nicht freigegebenen Artefakts wirft ValueError."""
|
|
with pytest.raises(ValueError, match="nicht freigegeben"):
|
|
bridge.revoke_share("nonexistent/artifact")
|
|
|
|
def test_revoke_removes_from_target_contexts(self, bridge: ContextBridge) -> None:
|
|
"""Nach Widerruf ist das Artefakt in keinem Zielkontext mehr sichtbar."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
bridge.revoke_share("bahn/decision/api-design")
|
|
|
|
assert bridge.get_shared_artifacts("privat") == []
|
|
assert bridge.get_shared_artifacts("dhive") == []
|
|
assert bridge.get_shared_artifacts("shared") == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: get_shared_content (Update-Propagierung, Req 5.5)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetSharedContent:
|
|
"""Tests für Lesezugriff und Update-Propagierung."""
|
|
|
|
def test_read_shared_artifact(self, bridge: ContextBridge) -> None:
|
|
"""Req 5.2: Freigegebenes Artefakt ist in Zielkontext lesbar."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
content = bridge.get_shared_content("bahn/decision/api-design", "privat")
|
|
assert content is not None
|
|
assert "REST" in content
|
|
|
|
def test_read_not_shared_returns_none(self, bridge: ContextBridge) -> None:
|
|
"""Nicht freigegebenes Artefakt liefert None."""
|
|
content = bridge.get_shared_content("bahn/decision/api-design", "privat")
|
|
assert content is None
|
|
|
|
def test_read_from_unauthorized_context_returns_none(
|
|
self, bridge: ContextBridge
|
|
) -> None:
|
|
"""Lesezugriff aus Quellkontext (nicht Zielkontext) liefert None."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
# Bahn ist der Quellkontext, nicht ein Zielkontext
|
|
content = bridge.get_shared_content("bahn/decision/api-design", "bahn")
|
|
assert content is None
|
|
|
|
def test_update_propagation(
|
|
self, bridge: ContextBridge, tmp_knowledge_base: Path
|
|
) -> None:
|
|
"""Req 5.5: Änderungen im Quellkontext sind beim nächsten Lesen sichtbar."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
|
|
# Inhalt aktualisieren
|
|
artifact_path = tmp_knowledge_base / "bahn" / "decision" / "api-design.md"
|
|
artifact_path.write_text(
|
|
"---\n"
|
|
"type: decision\n"
|
|
"title: API Design Entscheidung\n"
|
|
"tags: [api, design]\n"
|
|
"source_context: bahn\n"
|
|
"shareable: true\n"
|
|
"content_hash: sha256:updated\n"
|
|
"---\n"
|
|
"# API Design v2\n\n"
|
|
"Wir verwenden GraphQL für die externe API.\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Nächster Lesezugriff sieht den neuen Inhalt
|
|
content = bridge.get_shared_content("bahn/decision/api-design", "privat")
|
|
assert content is not None
|
|
assert "GraphQL" in content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: get_shared_artifacts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetSharedArtifacts:
|
|
"""Tests für die Abfrage sichtbarer Artefakte pro Kontext."""
|
|
|
|
def test_returns_shared_artifacts_for_context(self, bridge: ContextBridge) -> None:
|
|
"""Listet freigegebene Artefakte für einen Zielkontext."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
artifacts = bridge.get_shared_artifacts("privat")
|
|
assert "bahn/decision/api-design" in artifacts
|
|
|
|
def test_empty_for_source_context(self, bridge: ContextBridge) -> None:
|
|
"""Quellkontext sieht sein eigenes Artefakt nicht in der Share-Liste."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
artifacts = bridge.get_shared_artifacts("bahn")
|
|
assert "bahn/decision/api-design" not in artifacts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Registry-Persistierung
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRegistryPersistence:
|
|
"""Tests für die YAML-Persistierung der Shared-Artifacts-Registry."""
|
|
|
|
def test_registry_persisted_after_share(
|
|
self, bridge: ContextBridge, tmp_path: Path
|
|
) -> None:
|
|
"""Registry wird nach Freigabe auf Festplatte geschrieben."""
|
|
bridge.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
|
|
registry_path = tmp_path / "config" / "shared-artifacts.yaml"
|
|
assert registry_path.exists()
|
|
|
|
def test_registry_loaded_on_init(
|
|
self,
|
|
tmp_knowledge_base: Path,
|
|
yaml_index: YAMLIndex,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Registry wird beim Instanzieren aus der Datei geladen."""
|
|
registry_path = tmp_path / "config" / "shared-artifacts.yaml"
|
|
|
|
# Erste Bridge: Artefakt freigeben
|
|
bridge1 = ContextBridge(
|
|
knowledge_base_path=tmp_knowledge_base,
|
|
index=yaml_index,
|
|
registry_path=registry_path,
|
|
)
|
|
bridge1.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
|
|
# Zweite Bridge: Registry sollte geladen sein
|
|
bridge2 = ContextBridge(
|
|
knowledge_base_path=tmp_knowledge_base,
|
|
index=yaml_index,
|
|
registry_path=registry_path,
|
|
)
|
|
assert bridge2.is_shared("bahn/decision/api-design") is True
|
|
|
|
def test_registry_updated_after_revoke(
|
|
self,
|
|
tmp_knowledge_base: Path,
|
|
yaml_index: YAMLIndex,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Registry wird nach Widerruf aktualisiert."""
|
|
registry_path = tmp_path / "config" / "shared-artifacts.yaml"
|
|
|
|
bridge1 = ContextBridge(
|
|
knowledge_base_path=tmp_knowledge_base,
|
|
index=yaml_index,
|
|
registry_path=registry_path,
|
|
)
|
|
bridge1.share_artifact("bahn/decision/api-design", user_confirmed=True)
|
|
bridge1.revoke_share("bahn/decision/api-design")
|
|
|
|
# Neue Bridge: Artefakt sollte nicht mehr geteilt sein
|
|
bridge2 = ContextBridge(
|
|
knowledge_base_path=tmp_knowledge_base,
|
|
index=yaml_index,
|
|
registry_path=registry_path,
|
|
)
|
|
assert bridge2.is_shared("bahn/decision/api-design") is False
|