"""Ontologie: Modell, Export, Abgleich mit n8n (synthetische Workflows), Übernahme, Service-Routine.""" import json import pytest from knaster.ontology import Ontology, apply_decisions, build_snapshot, check, to_graph, to_turtle from knaster.ontology.n8n import is_closed, literal_values, parse_know_code from knaster.settings import REPO_ROOT KNOW_CODE = r""" const allowed=['setup_schema','store_mail_event','upsert_hint','promote_hint','find_mail']; if(operation==='setup_schema'){ query=`CREATE TABLE IF NOT EXISTS ${t.mail_events} (id BIGSERIAL PRIMARY KEY, message_id TEXT UNIQUE NOT NULL, category TEXT, sender TEXT); CREATE TABLE IF NOT EXISTS ${t.hints} (hint_key TEXT UNIQUE NOT NULL, hint_type TEXT NOT NULL, quality_status TEXT NOT NULL DEFAULT 'observed'); CREATE TABLE IF NOT EXISTS ${t.config} (config_key TEXT UNIQUE NOT NULL, weight NUMERIC); CREATE TABLE IF NOT EXISTS ${t.orphans} (id BIGSERIAL PRIMARY KEY, x TEXT); ALTER TABLE ${t.mail_events} ADD COLUMN IF NOT EXISTS received_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS mood TEXT; INSERT INTO ${t.config} (config_key, weight) VALUES ('mahnung', 3) ON CONFLICT (config_key) DO NOTHING; INSERT INTO ${t.config} (config_key, weight) VALUES ('default', 0) ON CONFLICT (config_key) DO NOTHING;`;} if(operation==='store_mail_event'){ query=`INSERT INTO ${t.mail_events} (message_id,category,sender) VALUES (${sqlString(payload.message_id)},${sqlString(payload.category)},${sqlString(payload.sender)}) ON CONFLICT (message_id) DO UPDATE SET category=EXCLUDED.category RETURNING *`;} if(operation==='upsert_hint'){ query=`INSERT INTO ${t.hints} (hint_key,hint_type) VALUES (${k},${ht}) ON CONFLICT (hint_key) DO UPDATE SET quality_status=CASE WHEN x THEN 'proposed' ELSE ${t.hints}.quality_status END`;} if(operation==='promote_hint'){ query=`UPDATE ${t.hints} SET quality_status='confirmed' WHERE hint_key=${k}`;} if(operation==='find_mail'){ query=`SELECT * FROM ${t.mail_events}`;} """ ONTO = { "id": "t", "namespace": "urn:test#", "modules": {"m": "Mail"}, "classes": { "Mail": {"label": "Mail", "module": "m", "table": "mail_events", "key": "message_id", "attributes": {"id": "BIGSERIAL", "message_id": "TEXT", "category": "TEXT", "sender": "TEXT", "received_at": "TIMESTAMPTZ"}}, "Hinweis": {"label": "Hinweis", "module": "m", "table": "hints", "attributes": {"hint_key": "TEXT", "hint_type": "TEXT", "quality_status": "TEXT"}}, "Kategorie": {"label": "Kategorie", "module": "m", "table": "config", "attributes": {"config_key": "TEXT", "weight": "NUMERIC"}}, "Alt": {"label": "Alt", "module": "m", "table": "removed_table", "attributes": {"a": "TEXT"}}, }, "relations": [{"source": "Hinweis", "name": "hatKategorie", "target": "Kategorie", "via": "hint_type"}], "vocabularies": { "Kat": {"label": "Kategorie", "class": "Kategorie", "columns": ["mail_events.category", "hints.hint_type", "config.config_key"], "concepts": {"mahnung": {"label": "Mahnung"}, "rechnung": {"label": "Rechnung", "alt": ["Buchhaltung"]}, "sonstiges": {}, "default": {"technical": True}}}, "QS": {"label": "Qualität", "columns": ["hints.quality_status"], "concepts": {"observed": {}, "proposed": {}, "confirmed": {}, "rejected": {}}}, }, } CAT_PAYLOAD = ("={{ JSON.stringify({ message_id: $json.id, sender: ($('Vorfilter').item.json.from||'').toLowerCase(), " "category: ({'Buchhaltung':'rechnung','Bauvorhaben':'bauvorhaben','Sonstiges':'sonstiges'}" "[$json.output?.category]||'sonstiges'), raw_signals: { category: 'x' } }) }}") def _call(name, op, payload="", node="KNOW"): return {"name": node, "type": "n8n-nodes-base.executeWorkflow", "parameters": {"workflowId": {"value": "know1"}, "workflowInputs": {"value": {"operation": op, "payload_json": payload}}}} WORKFLOWS = [ {"id": "know1", "name": "KNOW", "active": True, "nodes": [{"name": "Normalize", "type": "n8n-nodes-base.code", "parameters": {"jsCode": KNOW_CODE}}]}, {"id": "w1", "name": "CAT", "active": True, "nodes": [ _call("CAT", "store_mail_event", CAT_PAYLOAD), {"name": "Mark", "type": "n8n-nodes-base.postgres", "parameters": {"query": "UPDATE knowledge.hints SET x=1 WHERE hint_key='{{ $json.k }}'"}}]}, {"id": "w2", "name": "MPA", "active": True, "nodes": [ _call("MPA", "store_mail_event", '={{ JSON.stringify({ message_id: $json.id, category: "MPA" }) }}'), _call("MPA", "log_decision", "{}", node="Log1"), _call("MPA", "log_decision", "{}", node="Log2")]}, {"id": "w3", "name": "Dyn", "active": True, "nodes": [ {"name": "Build", "type": "n8n-nodes-base.code", "parameters": {"jsCode": "return {operation: 'upsert_hint'}"}}, _call("Dyn", "={{ $json.operation }}")]}, {"id": "w4", "name": "Old", "active": False, "nodes": [ _call("Old", "store_mail_event", "={{ JSON.stringify({ category: 'legacy_kat' }) }}")]}, ] @pytest.fixture def onto(): return Ontology(**json.loads(json.dumps(ONTO))) def test_repo_ontology_is_valid_and_turtle_current(): path = REPO_ROOT / "ontology" / "know.yaml" o = Ontology.from_file(path) assert o.validate_model() == [] assert len(o.classes) >= 49 and "MailKategorie" in o.vocabularies assert (REPO_ROOT / "ontology" / "know.ttl").read_text(encoding="utf-8") == to_turtle(o) def test_model_validation_finds_errors(onto): onto.relations.append(onto.relations[0].model_copy(update={"target": "Gibtsnicht"})) onto.vocabularies["Kat"].concepts["sonstiges"].alt = ["Mahnung"] onto.vocabularies["Kat"].columns.append("mail_events.fehlt") onto.vocabularies["Kat"].rejected = ["rechnung"] errors = "\n".join(onto.validate_model()) assert "Gibtsnicht" in errors and "„Mahnung“" in errors and "mail_events.fehlt" in errors assert "zugleich verworfen" in errors def test_vocabulary_resolve(onto): v = onto.vocabularies["Kat"] assert v.resolve("Buchhaltung") == "rechnung" and v.resolve("MAHNUNG") == "mahnung" and v.resolve("x") is None def test_exports(onto): ttl = to_turtle(onto) assert "t:Mail a owl:Class" in ttl and "skos:altLabel \"Buchhaltung\"@de" in ttl assert "t:hatKategorie a owl:ObjectProperty" in ttl g = to_graph(onto, generated="2026-10-13") mail = next(c for c in g["classes"] if c["id"] == "Mail") assert "id" not in [a[0] for a in mail["attrs"]] hint = next(c for c in g["classes"] if c["id"] == "Hinweis") assert hint["enums"]["quality_status"] == ["observed", "proposed", "confirmed", "rejected"] assert "Mahnung" in next(c for c in g["classes"] if c["id"] == "Kategorie")["individuals"] def test_parse_know_code(): spec = parse_know_code(KNOW_CODE) assert spec.allowed == {"setup_schema", "store_mail_event", "upsert_hint", "promote_hint", "find_mail"} assert set(spec.schema["mail_events"]) == {"id", "message_id", "category", "sender", "received_at", "mood"} assert spec.writes["store_mail_event"]["mail_events"] == {"message_id", "category", "sender"} assert spec.writes["promote_hint"]["hints"] == {"quality_status"} assert spec.seeds["config.config_key"] == {"mahnung", "default"} assert "config" not in spec.writes.get("setup_schema", {}) # Seeds gehören zu keiner Operation assert spec.assigned["upsert_hint"]["hints.quality_status"] == {"proposed"} assert spec.assigned["promote_hint"]["hints.quality_status"] == {"confirmed"} @pytest.mark.parametrize("expr,key,values,closed", [ (CAT_PAYLOAD, "category", {"rechnung", "bauvorhaben", "sonstiges"}, True), (CAT_PAYLOAD, "sender", set(), False), ('{ category: "MPA", x: 1 }', "category", {"MPA"}, True), ("{ category: $json.cat || 'info' }", "category", {"info"}, False), ("={{ JSON.stringify({ x: { category: 'tief' } }) }}", "category", set(), False), ]) def test_literal_values_and_closed(expr, key, values, closed): assert literal_values(expr, key) == values assert is_closed(expr, key) is closed def test_check_against_synthetic_n8n(onto): snap = build_snapshot(WORKFLOWS) assert snap.know and snap.know.workflow_id == "know1" dyn = [c for c in snap.calls if c.workflow == "Dyn"] assert [(c.operation, c.inferred) for c in dyn] == [("upsert_hint", True)] findings = check(onto, snap) by_kind = {} for f in findings: by_kind.setdefault(f.kind, []).append(f) # unbekannte Operation: zwei Knoten → ein Befund [unknown] = by_kind["unknown_operation"] assert unknown.severity == "high" and "Log1, Log2" in unknown.detail [sql] = by_kind["direct_sql"] assert sql.severity == "medium" and "hints" in sql.summary vocab = {f.change["value"]: f for f in by_kind["vocab_value"]} assert set(vocab) == {"bauvorhaben", "MPA", "legacy_kat"} assert vocab["legacy_kat"].severity == "low" and vocab["MPA"].severity == "medium" # mahnung wird von keiner festen Zuordnung erzeugt, default ist technisch [unused] = [f for f in by_kind["concept_without_producer"] if f.key == "unused:Kat"] assert "mahnung" in unused.summary and "default" not in unused.summary # Qualitätsstatus: proposed/confirmed setzt KNOW selbst (upsert_hint ist aktiv abgeleitet, promote_hint nicht) qs = [f for f in by_kind.get("concept_without_producer", []) if f.key == "unused:QS"] assert not qs # keine feste Zuordnung aus Workflows → keine Aussage assert {f.change["column"] for f in by_kind["unmapped_column"]} == {"mood"} assert by_kind["unmapped_table"][0].change["table"] == "orphans" assert by_kind["missing_table"][0].key == "gone:removed_table" assert "no_write_operation" not in by_kind # orphans ist keiner Klasse zugeordnet, config ist Seed-Tabelle assert "nowriter:config" not in {f.key for f in findings} # Seed-Tabelle # entschiedene bzw. ignorierte Schlüssel verschwinden onto.ignore.append("table:orphans") keys = {f.key for f in check(onto, snap, decided={"vocab:Kat:mpa"})} assert "vocab:Kat:mpa" not in keys and "table:orphans" not in keys def test_apply_decisions(onto): decisions = [ {"key": "vocab:Kat:bauvorhaben", "decision": "accept", "params": {"type": "vocab_value", "vocabulary": "Kat", "value": "Bauvorhaben"}}, {"key": "vocab:Kat:buchhaltg", "decision": "map", "params": {"type": "vocab_value", "vocabulary": "Kat", "value": "Buchhaltg", "concept": "rechnung"}}, {"key": "vocab:Kat:mpa", "decision": "reject", "params": {"type": "vocab_value", "vocabulary": "Kat", "value": "MPA"}}, {"key": "column:mail_events.mood", "decision": "accept", "params": {"type": "add_attribute", "class": "Mail", "column": "mood", "sql_type": "TEXT"}}, {"key": "table:orphans", "decision": "accept", "params": {"type": "add_class", "table": "orphans", "attributes": {"x": "TEXT"}}}, {"key": "sql:CAT:Mark", "decision": "reject", "params": {"type": "x"}}, {"key": "vocab:Kat:kaputt", "decision": "map", "params": {"type": "vocab_value", "vocabulary": "Kat", "value": "k", "concept": "gibtsnicht"}}, ] applied, messages = apply_decisions(onto, decisions) v = onto.vocabularies["Kat"] assert v.resolve("Bauvorhaben") == "bauvorhaben" and v.resolve("Buchhaltg") == "rechnung" assert v.is_rejected("mpa") and "mood" in onto.classes["Mail"].attributes assert onto.classes["Orphans"].module == "neu" and "sql:CAT:Mark" in onto.ignore assert "vocab:Kat:kaputt" not in applied and any("nicht übernommen" in m for m in messages) assert onto.version == 2 and onto.validate_model() == [] def test_routine_proposes_and_remembers_decisions(make_system, tmp_path, onto): path = tmp_path / "know.yaml" onto.save(path) sysm = make_system(ontology_path=path, ontology_source=lambda: WORKFLOWS) m = sysm.maintenance report = m.run_routine() assert any("log_decision" in f for f in report.findings) vocab = {p.params["value"]: p for p in report.proposals if p.kind == "ontology_change" and p.params["type"] == "vocab_value"} assert {"bauvorhaben", "MPA", "legacy_kat"} <= set(vocab) assert all(p.state == "open" for p in vocab.values()) # Ontologie-Änderungen nie automatisch # zuordnen, ablehnen, aufnehmen assert "executed" in m.handle_command(f"ordne {vocab['legacy_kat'].id} sonstiges zu", user="robert").text assert "rejected" in m.handle_command(f"lehne {vocab['MPA'].id} ab: Workflow-Name, keine Kategorie").text assert "executed" in m.handle_command(f"genehmige {vocab['bauvorhaben'].id}").text assert "Fehler" in m.handle_command("ordne P-999 sonstiges zu").text decisions = {d["key"]: d for d in sysm.store.list_ontology_decisions()} assert decisions["vocab:Kat:legacy_kat"]["decision"] == "map" assert decisions["vocab:Kat:legacy_kat"]["decided_by"] == "robert" assert decisions["vocab:Kat:mpa"]["decision"] == "reject" # zweiter Lauf: entschiedene Werte kommen nicht wieder again = m.run_routine() values = {p.params.get("value") for p in again.proposals if p.kind == "ontology_change"} assert not values & {"bauvorhaben", "MPA", "legacy_kat"} # Übernahme in die Datei pending = sysm.store.list_ontology_decisions(applied=False) applied, _ = apply_decisions(onto, pending) assert set(applied) == {"vocab:Kat:legacy_kat", "vocab:Kat:mpa", "vocab:Kat:bauvorhaben"} assert onto.vocabularies["Kat"].resolve("legacy_kat") == "sonstiges" def test_routine_survives_unreachable_n8n(make_system, tmp_path, onto): path = tmp_path / "know.yaml" onto.save(path) def broken(): raise ConnectionError("n8n nicht erreichbar") report = make_system(ontology_path=path, ontology_source=broken).maintenance.run_routine() assert any("Ontologie-Abgleich nicht möglich" in f for f in report.findings) def test_cli_check_with_snapshot(tmp_path, onto, monkeypatch, capsys): from knaster.cli import main path = tmp_path / "know.yaml" onto.save(path) snap = tmp_path / "wf.json" snap.write_text(json.dumps({"data": WORKFLOWS})) monkeypatch.setenv("KNASTER_DB_URL", "sqlite://") assert main(["ontology", "--path", str(path), "check", "--snapshot", str(snap)]) == 1 # high-Befund out = capsys.readouterr().out assert "log_decision" in out and "Änderungsvorschläge" in out assert main(["ontology", "--path", str(path), "validate"]) == 0 html = tmp_path / "o.html" assert main(["ontology", "--path", str(path), "export", "--html", str(html), "--template", str(REPO_ROOT / "ontology" / "viewer.html.tmpl")]) == 0 assert "__DATA__" not in html.read_text() and "owl:Class" in html.read_text()