Lars/Kansho
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
53b1eff833
Kansho/backend/tests/test_privacy_detect.py
Lars 53b1eff833 MVP 0.92
2026-08-26 10:42:20 +02:00

277 lines
14 KiB
Python
Raw Blame History

"""Detection + mask + demask. Run from backend/: python tests/test_privacy_detect.py"""
from __future__ import annotations
import os
import sys
import tempfile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
os.environ["KANSHO_DB_PATH"] = str(Path(tempfile.gettempdir()) / "kansho-privacy-detect-test.sqlite")
os.environ["KANSHO_FAKE_PROVIDER"] = "1"
os.environ["KANSHO_FAKE_DETECT"] = "1"
Path(os.environ["KANSHO_DB_PATH"]).unlink(missing_ok=True)
from fastapi.testclient import TestClient
from identity_store import list_mappings
from main import app
from privacy_gateway import reset_debug
def expect(ok: bool, message: str) -> None:
if not ok:
raise SystemExit(f"FAIL: {message}")
print(f"OK {message}")
def main() -> None:
reset_debug()
with TestClient(app) as client:
setup = client.post(
"/api/auth/setup",
json={"email": "lars@example.test", "name": "Lars", "password": "test-pass"},
)
headers = {"X-Auth-Token": setup.json()["token"]}
space = client.post("/api/journal/spaces", headers=headers, json={"title": "Alltag"})
day = client.post(
f"/api/journal/spaces/{space.json()['id']}/days",
headers=headers,
json={"calendar_date": "2026-08-22"},
)
conv = client.post(
f"/api/journal/days/{day.json()['day']['id']}/conversations",
headers=headers,
json={"title": "Gespräch"},
)
turn = client.post(
f"/api/journal/conversations/{conv.json()['id']}/turn",
headers=headers,
json={"body": "Ich war mit Anna in Hamburg und danach allein."},
)
expect(turn.status_code == 200, f"turn {turn.text}")
trace = turn.json().get("trace") or {}
expect(trace, "gateway returned request-scoped trace")
expect(trace["detect_provider"] == "fake-detect", "detect provider is separate")
expect("Anna" not in (trace.get("egress") or ""), "Anna masked before generate")
expect("Hamburg" not in (trace.get("egress") or ""), "Hamburg masked before generate")
expect("[[PERSON:" in (trace.get("egress") or "") or "[[PLACE:" in (trace.get("egress") or ""), "placeholders in egress")
expect("Anna" in turn.json()["user"]["body"], "source keeps cleartext")
mappings = list_mappings(setup.json()["profile_id"])
labels = {item["local_label"] for item in mappings}
expect("Anna" in labels, "Anna stored in local dictionary")
expect("Hamburg" in labels, "Hamburg stored in local dictionary")
expect(all(not (item.get("token") or "").startswith("[[") or True for item in mappings), "tokens stored")
reset_debug()
again = client.post(
f"/api/journal/conversations/{conv.json()['id']}/turn",
headers=headers,
json={"body": "Anna war ruhig."},
)
expect(again.status_code == 200, "second turn")
last2 = again.json().get("trace") or {}
expect("nicht" in (last2.get("egress") or "").lower() or "Anna" not in (last2.get("egress") or ""), "dictionary remasks Anna without new invention")
expect(last2.get("mapping_count", 0) >= 2, "dictionary reused")
from entity_detect import _pattern_entities
from identity_store import is_maskable_label
from privacy_gateway import _mask
patterned = {item["text"] for item in _pattern_entities("Ich war mit Anna in Hamburg.", set())}
expect("Anna" in patterned and "Hamburg" in patterned, "pattern finds proper names")
expect("Himmel" not in {item["text"] for item in _pattern_entities("Der Himmel war klar.", set())}, "pattern skips weather")
expect(not is_maskable_label("ich"), "pronoun ich is not a mapping")
expect(not is_maskable_label("user"), "role user is not a mapping")
expect(not is_maskable_label("Himmel"), "weather is not identity")
expect(not is_maskable_label("Der Himmel"), "determiner plus weather is not identity")
expect(not is_maskable_label("6:00 Uhr"), "clock time is not identity")
expect(not is_maskable_label("meinen Tee"), "tea is not identity")
expect(not is_maskable_label("Relativ"), "sentence adverb is not identity")
expect(not is_maskable_label("Allerdings"), "conjunction is not identity")
expect(not is_maskable_label("Tages"), "common noun is not identity")
expect(not is_maskable_label("Brote"), "food is not identity")
expect(not is_maskable_label("Überlegen"), "common noun is not identity")
expect(not is_maskable_label("Ortes"), "place word is not identity")
expect(not is_maskable_label("Zwiebeln"), "food is not identity")
expect(not is_maskable_label("letzten"), "function word letzten is not identity")
expect(not is_maskable_label("Getöse"), "collective noun is not identity")
expect(not is_maskable_label("Schiff"), "vessel is not identity")
expect(not is_maskable_label("Faden"), "prompt word is not identity")
expect("Getöse" not in {item["text"] for item in _pattern_entities("Zweimaster mit Getöse im Hafen.", set())}, "mit plus common noun is not a name")
expect("Anna" in {item["text"] for item in _pattern_entities("Ich war mit Anna im Hafen.", set())}, "mit plus given name stays")
expect(is_maskable_label("Sushi"), "given name stays maskable")
expect(is_maskable_label("meine Frau Sushi"), "kinship stripped, name remains")
from identity_store import coalesce_mappings, normalize_token, remember_mapping
from db import get_db
import uuid
expect(normalize_token("PERSON:KURZ") is None, "template token KURZ rejected")
first = remember_mapping(setup.json()["profile_id"], "meine Frau Sushi", "PERSON:08")
second = remember_mapping(setup.json()["profile_id"], "Sushi", "PERSON:KURZ")
expect(first["token"] == second["token"], "alias reuses token")
expect(second["local_label"] == "Sushi", "canonical label is the name")
extra_id = str(uuid.uuid4())
with get_db() as conn:
conn.execute(
"""
INSERT INTO identity_mappings (id, profile_id, token, local_label)
VALUES (?, ?, ?, ?)
""",
(extra_id, setup.json()["profile_id"], "PERSON:KURZ", "meine Frau Sushi"),
)
coalesce_mappings(setup.json()["profile_id"])
sushi_rows = [
item
for item in list_mappings(setup.json()["profile_id"])
if (item.get("local_label") or "").lower() == "sushi"
]
expect(len(sushi_rows) == 1, "coalesce keeps one Sushi row")
expect(sushi_rows[0]["token"].upper() != "PERSON:KURZ", "stable token wins over KURZ")
joined = _mask(
"meine Frau Sushi kam. Sushi stellte fest.",
sushi_rows,
)
expect(joined.count(f"[[{sushi_rows[0]['token']}]]") == 2, "both mentions share one placeholder")
expect("Sushi" not in joined, "name fully masked")
expect("meine Frau " in joined, "kinship stays visible")
homonym = _mask(
"Ich esse gerne Sushi mit meiner Frau Sushi.",
sushi_rows,
)
expect("esse gerne Sushi" in homonym, "food homonym stays")
expect("Frau [[" in homonym or f"Frau [[{sushi_rows[0]['token']}]]" in homonym, "kinship name is masked")
expect(homonym.count("[[") == 1, "only the person mention is masked")
eating = _mask("Heute nur von Sushi essen erzählt.", sushi_rows)
expect("Sushi essen" in eating, "noun before eat stays food")
expect("[[" not in eating, "food-only sentence is not the person")
from pronoun_bind import bind_user_lines, bind_pronouns
from privacy_gateway import mask_for_egress
anna = [{"local_label": "Anna", "token": "PERSON:01"}]
bound = bind_user_lines(
_mask(
"user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.",
anna,
)
)
expect(bound.count("[[PERSON:01]]") == 3, "name and both pronouns share one token")
expect("Sie" not in bound and "ihr" not in bound, "pronouns rebound to person")
journal_masked = mask_for_egress(
"user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.",
anna,
"journal_generate",
)
expect(journal_masked.count("[[PERSON:01]]") == 1, "journal masks the name only")
expect("Sie setzte" in journal_masked and "ihr Tee" in journal_masked, "journal keeps anaphora")
dialogue_masked = mask_for_egress(
"user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.",
anna,
"dialogue_turn",
)
expect(dialogue_masked.count("[[PERSON:01]]") == 3, "dialogue still binds pronouns")
instruction = bind_user_lines(
"user: Anna kam.\n"
"Der Impuls darf eine Frage sein. Er darf keine neue Station setzen.\n"
)
expect("Er darf keine neue Station" in instruction, "instruction pronoun stays")
instruction_name = _mask(
"Nenne Berg nicht in der Anweisung.\nuser: Berg kam vorbei.",
[{"local_label": "Berg", "token": "PERSON:02"}],
personal_lines_only=True,
)
expect("Nenne Berg nicht" in instruction_name, "instruction line is not remasked")
expect("[[PERSON:02]]" in instruction_name, "user line still remasked")
herr = [{"local_label": "Berg", "token": "PERSON:02"}]
male, _ = bind_pronouns(_mask("Herr Berg kam. Er setzte sich.", herr))
expect(male.count("[[PERSON:02]]") == 2, "masculine pronoun follows last man")
address, _ = bind_pronouns(_mask("Anna kam. Seid ihr schon da?", anna))
expect("ihr" in address, "address ihr is not a person")
plural, _ = bind_pronouns(_mask("Anna kam. Sie waren zu dritt.", anna))
expect("Sie waren" in plural or "sie waren" in plural.lower(), "plural sie stays")
croatia = (
"Heute bin ich ca. gegen 6:00 Uhr aufgestanden. Relativ früh gesellte sich "
"auch meine Frau Sushi zu mir. Allerdings sollte die Bootstour erst um 11:00 Uhr starten."
)
patterned_live = {item["text"] for item in _pattern_entities(croatia, set())}
expect("Sushi" in patterned_live, "pattern keeps given name")
expect(
patterned_live <= {"Sushi"},
f"pattern does not treat German function words as names: {patterned_live}",
)
from entity_detect import uses_llm_detect
from providers import ProviderConfig
remote = ProviderConfig(
role="detect",
name="openrouter",
mode="http",
url="https://openrouter.ai/api/v1/chat/completions",
model="x",
key="sk-test",
local=False,
zdr=True,
no_train=True,
)
expect(uses_llm_detect(remote), "development allows remote detect")
os.environ["KANSHO_ENV"] = "production"
expect(not uses_llm_detect(remote), "production blocks remote detect")
local = ProviderConfig(
role="detect",
name="ollama",
mode="http",
url="http://127.0.0.1:11434/v1/chat/completions",
model="x",
key="",
local=True,
zdr=True,
no_train=True,
)
expect(uses_llm_detect(local), "production still allows local detect")
os.environ.pop("KANSHO_ENV", None)
poisoned = _mask(
"Nicht erwähnt ist nicht geschehen. Ich kochte Tee.",
[{"local_label": "ich", "token": "PERSON:05"}],
)
expect("N[[PERSON:05]]t" not in poisoned, "ich does not split nicht")
expect("Nicht erwähnt" in poisoned, "instruction word bleibt")
from privacy_gateway import PrivacyGatewayError, _demask, _validate_response
mapped = [{"local_label": "Anna", "token": "PERSON:01"}]
kept = _validate_response("[[PERSON:01]] war ruhig.", mapped)
expect(kept.startswith("[[PERSON:01]]"), "placeholders pass validation")
expect(_demask(kept, mapped) == "Anna war ruhig.", "demask after validation")
blocked = False
try:
_validate_response("Anna war ruhig.", mapped)
except PrivacyGatewayError as exc:
blocked = exc.code == "response_validation_failed"
expect(blocked, "plaintext identity is blocked before demask")
sushi_map = [{"local_label": "Sushi", "token": "PERSON:01"}]
food_reply = _validate_response("Heute nur von Sushi essen erzählt.", sushi_map)
expect("Sushi essen" in food_reply, "food homonym in a reply is not treated as a leak")
mixed = mask_for_egress(
"Belegte Vorhaben (nur user-Quellen):\n- Morgen wollen wir mit Anna fahren.\nuser: Heute Markt.",
mapped,
"dialogue_turn",
)
expect("Anna" not in mixed, "opening context names are masked for dialogue egress")
expect("[[PERSON:01]]" in mixed, "opening context uses the person token")
expect("Heute Markt" in mixed, "current user wording remains")
excerpt = mask_for_egress(
"Frühere Einträge im Space (nur Hinweise, keine Gewissheit):\n2026-08-01: Anna war am Hafen.\nuser: Heute windstill.",
mapped,
"dialogue_turn",
)
expect("Anna" not in excerpt, "space excerpts are masked for dialogue egress")
expect("[[PERSON:01]]" in excerpt, "space excerpt uses the person token")
print("All privacy detect tests passed.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink