rorappandClaude Opus 5.5
9442cc49d1
ci / test (push) Successful in 27s
- ontology/know.yaml: 49 Klassen, 66 Beziehungen, 12 Vokabulare (72 Konzepte); erzeugt: know.ttl (OWL + SKOS), Ansicht aus viewer.html.tmpl - knaster.ontology: Modell + Prüfung, Export, Momentaufnahme der n8n-Workflows (nur GET; KNOW-Spezifikation aus dem Workflow-Code, Aufrufe inkl. Laufzeit- Operationen, direktes SQL), Abgleich, Übernahme von Entscheidungen - Service-Routine Schritt 8: Verstöße als Befunde, Ontologie-Änderungen als Vorschläge ontology_change (immer Freigabe); „ordne P-x <konzept> zu“; Entscheidungen in ontology_decisions, `knaster ontology apply` schreibt sie in die YAML (Version +1) - CLI: knaster ontology validate|export|snapshot|check|apply; knaster validate prüft Modell und Aktualität der Turtle-Datei - Gegen echten n8n-Export geprüft: neue Befunde B31–B33 (verified/confirmed, KAN ruft unbekannte Operation, MPA ordnet Projektereignisse nie ein) - SPEC-ontology, BETRIEB §6, Router-Image enthält die Ontologie Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
268 lines
15 KiB
Python
268 lines
15 KiB
Python
"""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()
|