126 lines
4.7 KiB
Python
126 lines
4.7 KiB
Python
"""Admin identity registry stays local. Run from backend/: python tests/test_identity_registry.py"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
os.environ["KANSHO_FAKE_PROVIDER"] = "1"
|
|
os.environ["KANSHO_FAKE_DETECT"] = "1"
|
|
|
|
from fastapi.testclient import TestClient
|
|
from identity_store import STATUS_LEGACY, list_confirmed_identities, list_registry
|
|
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()
|
|
import sqlite3
|
|
from db import _IDENTITY_MAPPING_COLUMNS, _ensure_columns
|
|
from identity_store import migrate_legacy_identity_rows as migrate_legacy
|
|
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE identity_mappings (
|
|
id TEXT PRIMARY KEY,
|
|
profile_id TEXT NOT NULL,
|
|
token TEXT NOT NULL,
|
|
local_label TEXT NOT NULL DEFAULT '',
|
|
created TEXT NOT NULL DEFAULT '2026-08-01 00:00:00'
|
|
);
|
|
INSERT INTO identity_mappings (id, profile_id, token, local_label)
|
|
VALUES ('m1', 'p1', 'PERSON:01', 'Altbau');
|
|
"""
|
|
)
|
|
_ensure_columns(conn, "identity_mappings", _IDENTITY_MAPPING_COLUMNS)
|
|
migrate_legacy(conn)
|
|
legacy = dict(conn.execute("SELECT * FROM identity_mappings WHERE id = 'm1'").fetchone())
|
|
expect(legacy["status"] == STATUS_LEGACY, "existing rows become legacy_review_required")
|
|
expect(legacy["origin"] == "legacy_auto", "existing rows keep legacy origin")
|
|
expect(legacy["canonical_label"] == "Altbau", "canonical_label is copied from local_label")
|
|
expect(bool(legacy["updated"]), "updated is populated after ALTER")
|
|
conn.close()
|
|
|
|
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"]}
|
|
profile_id = setup.json()["profile_id"]
|
|
|
|
from db import get_db
|
|
import uuid
|
|
|
|
legacy_id = str(uuid.uuid4())
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO identity_mappings
|
|
(id, profile_id, token, local_label, canonical_label, entity_type, status, origin)
|
|
VALUES (?, ?, ?, ?, ?, 'PERSON', ?, 'legacy_auto')
|
|
""",
|
|
(legacy_id, profile_id, "PERSON:70", "Altbau", "Altbau", STATUS_LEGACY),
|
|
)
|
|
expect(
|
|
not any(item["id"] == legacy_id for item in list_confirmed_identities(profile_id)),
|
|
"legacy row is not confirmed",
|
|
)
|
|
expect(any(item["id"] == legacy_id for item in list_registry(profile_id)), "legacy row remains reviewable")
|
|
|
|
patched = client.patch(
|
|
f"/api/admin/identities/{legacy_id}",
|
|
headers=headers,
|
|
json={"status": "confirmed", "entity_type": "PLACE", "canonical_label": "Altbau"},
|
|
)
|
|
expect(patched.status_code == 200, patched.text)
|
|
expect(patched.json()["status"] == "confirmed", "admin can confirm a legacy row")
|
|
expect(patched.json()["entity_type"] == "PLACE", "admin can correct the type")
|
|
|
|
created = client.post(
|
|
"/api/admin/identities",
|
|
headers=headers,
|
|
json={"canonical_label": "Nordwerk", "entity_type": "ORG", "aliases": ["Nordwerk AG"]},
|
|
)
|
|
expect(created.status_code == 200, created.text)
|
|
identity_id = created.json()["id"]
|
|
aliases = client.patch(
|
|
f"/api/admin/identities/{identity_id}",
|
|
headers=headers,
|
|
json={"aliases": ["Nordwerk AG", "Nordwerk"]},
|
|
)
|
|
expect(aliases.status_code == 200, aliases.text)
|
|
expect("Nordwerk AG" in (aliases.json().get("aliases") or []), "confirmed alias can be stored")
|
|
|
|
calls = {"n": 0}
|
|
|
|
def boom(*_args, **_kwargs):
|
|
calls["n"] += 1
|
|
raise AssertionError("no egress")
|
|
|
|
with patch("entity_detect.complete_chat", boom), patch("providers.complete_chat", boom):
|
|
gone = client.delete(f"/api/admin/identities/{identity_id}", headers=headers)
|
|
expect(gone.status_code == 200, "admin delete stays local")
|
|
expect(calls["n"] == 0, "registry actions do not call detect or generate")
|
|
expect(not any(item["id"] == identity_id for item in list_registry(profile_id)), "deleted row is gone")
|
|
|
|
print("All identity registry tests passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|