"""Strukturtests für die Modelle unter .agents/skills/sn-bpmn/beispiele/ – reines Python/pytest. Die Referenzmodelle (referenz_*.md) sind echte, größere Prozesse, die Syntaxbeispiele (beispiel_*.md) zeigen gezielt einzelne Elemente. Der Byte-Vergleich in test_pruefe_prozesse.py meldet jede Änderung an der Ausgabe von md_to_bpmn.py; diese Tests prüfen zusätzlich, ob das Ergebnis danach noch den Modellierregeln und den Grenzen von PICTURE genügt – auch wenn die .bpmn bewusst neu erzeugt wurde. """ import math import sys from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] SKILL = ROOT / ".agents" / "skills" / "sn-bpmn" sys.path.insert(0, str(SKILL)) import md_to_bpmn as m # noqa: E402 from test_md_to_bpmn import _liegt_in_rautenflaeche # noqa: E402 (dieselbe Rautengeometrie wie dort) MODELLE = sorted((SKILL / "beispiele").glob("*.md")) REFERENZEN = [md for md in MODELLE if md.stem.startswith("referenz_")] GATEWAYS = ("exclusiveGateway", "inclusiveGateway", "parallelGateway", "eventBasedGateway") EREIGNISSE = ("startEvent", "endEvent", "intermediateCatchEvent", "intermediateThrowEvent", "boundaryEvent") # Formen, durch die keine Kante laufen darf; Pools, Lanes, Gruppen und Hinweise zählen nicht als Hindernis HINDERNISSE = GATEWAYS + EREIGNISSE + ( "task", "manualTask", "userTask", "serviceTask", "scriptTask", "businessRuleTask", "sendTask", "receiveTask", "subProcess", "callActivity", "dataObjectReference", "dataStoreReference", ) KANTEN = ("sequenceFlow", "messageFlow") # Bekannte, per Issue erfasste Konverterfehler: (Modell, Kante, Element) -> Grund mit Issue-Nummer. # Jedes Paar bekommt einen eigenen xfail(strict=True)-Test; sobald der Fehler behoben ist, schlägt dieser als XPASS # fehl – dann den Eintrag hier entfernen. Nie verwenden, um neue Fehler zu verstecken (AGENTS.md). BEKANNTE_KREUZUNGEN = {} def baue(md): meta, steps, order, externals = m.parse_markdown(md.read_text(encoding="utf-8")) return m.build_bpmn(meta, steps, order, externals) def _liegt_innen(px, py, art, x, y, w, h, abstand=1.0): """True, wenn der Punkt im Inneren der sichtbaren Form liegt (Kontur und Umgebung zählen nicht).""" if art in GATEWAYS: return _liegt_in_rautenflaeche(px, py, x, y, w, h) if art in EREIGNISSE: r = min(w, h) / 2 return math.hypot(px - x - w / 2, py - y - h / 2) < r - abstand return x + abstand < px < x + w - abstand and y + abstand < py < y + h - abstand def kreuzungen(defs): """Alle Paare (Kante, Element), bei denen ein Sequenz- oder Nachrichtenfluss durchs Innere eines Elements läuft. Die Kante wird in Schritten von höchstens 4 Einheiten abgetastet, damit auch kleine Ereignisse nicht zwischen zwei Stichproben fallen.""" art = {el.get("id"): el.tag.rpartition("}")[2] for el in defs.iter() if el.get("id")} formen = {} for shape in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape"): eid = shape.get("bpmnElement") if art.get(eid) in HINDERNISSE: b = shape.find(f"{{{m.NS['dc']}}}Bounds") formen[eid] = tuple(float(b.get(a)) for a in ("x", "y", "width", "height")) funde = set() for edge in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNEdge"): kante = edge.get("bpmnElement") if art.get(kante) not in KANTEN: continue punkte = [(float(wp.get("x")), float(wp.get("y"))) for wp in edge.findall(f"{{{m.NS['di']}}}waypoint")] for eid, (x, y, w, h) in formen.items(): for (x1, y1), (x2, y2) in zip(punkte, punkte[1:]): schritte = max(8, int(math.hypot(x2 - x1, y2 - y1) / 4)) if any(_liegt_innen(x1 + (x2 - x1) * s / schritte, y1 + (y2 - y1) * s / schritte, art[eid], x, y, w, h) for s in range(1, schritte)): funde.add((kante, eid)) break return sorted(funde) def test_es_gibt_mindestens_ein_referenzmodell(): """Prüft, dass die Strukturtests tatsächlich an einem Referenzmodell laufen.""" assert REFERENZEN @pytest.mark.parametrize("md", REFERENZEN, ids=lambda p: p.stem) def test_jedes_element_hat_eine_zustaendigkeit(md): """Prüft, dass im Referenzmodell kein Element in der Lane 'Nicht zugeordnet' landet.""" lanes = [lane.get("name") for lane in baue(md).iter(f"{{{m.NS['bpmn']}}}lane")] assert lanes, "Referenzmodell ohne Lanes" assert m.UNASSIGNED_LANE not in lanes @pytest.mark.parametrize("md", REFERENZEN, ids=lambda p: p.stem) def test_kein_gateway_hat_mehr_als_vier_kanten(md): """Prüft Modellierregel 4: kein Gateway im Referenzmodell braucht mehr als die vier Anschlusspunkte von PICTURE.""" defs = baue(md) for tag in GATEWAYS: for gateway in defs.iter(f"{{{m.NS['bpmn']}}}{tag}"): kanten = gateway.findall(f"{{{m.NS['bpmn']}}}incoming") + gateway.findall(f"{{{m.NS['bpmn']}}}outgoing") assert len(kanten) <= 4, f"{gateway.get('id')} hat {len(kanten)} Kanten" @pytest.mark.parametrize("md", MODELLE, ids=lambda p: p.stem) def test_keine_unbekannte_kante_fuehrt_durch_ein_element(md): """Prüft, dass in keinem Modell ein Sequenz- oder Nachrichtenfluss durchs Innere einer Aufgabe, eines Ereignisses, eines Gateways oder eines Datenelements läuft – außer den per Issue erfassten Fällen. Meldet alle Funde auf einmal.""" bekannt = {(k, e) for (modell, k, e) in BEKANNTE_KREUZUNGEN if modell == md.stem} neu = [paar for paar in kreuzungen(baue(md)) if paar not in bekannt] assert neu == [], "Kante durch Element (Kante, Element): " + ", ".join(f"{k} → {e}" for k, e in neu) @pytest.mark.parametrize("modell, kante, element", [ pytest.param(*schluessel, id="-".join(schluessel), marks=pytest.mark.xfail(reason=grund, strict=True)) for schluessel, grund in BEKANNTE_KREUZUNGEN.items() ]) def test_bekannte_kreuzung_ist_behoben(modell, kante, element): """Prüft je per Issue erfasster Kreuzung, ob sie noch besteht; als xfail(strict=True) markiert, damit ein Fix auffällt und der Eintrag in BEKANNTE_KREUZUNGEN entfernt wird.""" assert (kante, element) not in kreuzungen(baue(SKILL / "beispiele" / f"{modell}.md")) def _box(el): b = el.find(f"{{{m.NS['dc']}}}Bounds") return tuple(float(b.get(a)) for a in ("x", "y", "width", "height")) def _ueberlappen(a, b, rand=0.5): """True, wenn sich die Rechtecke a und b (x, y, w, h) echt überschneiden (Berührung zählt nicht).""" return (a[0] + rand < b[0] + b[2] and b[0] + rand < a[0] + a[2] and a[1] + rand < b[1] + b[3] and b[1] + rand < a[1] + a[3]) def beschriftungskonflikte(defs): """Alle Konflikte einer Gateway-Beschriftung mit Kanten, Kantenbeschriftungen oder anderen Elementen.""" art = {el.get("id"): el.tag.rpartition("}")[2] for el in defs.iter() if el.get("id")} gw_labels, formen = {}, {} for shape in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape"): eid = shape.get("bpmnElement") if art.get(eid) in HINDERNISSE or art.get(eid) == "textAnnotation": formen[eid] = _box(shape) label = shape.find(f"{{{m.NS['bpmndi']}}}BPMNLabel") if art.get(eid) in GATEWAYS and label is not None: gw_labels[eid] = _box(label) konflikte = [] for edge in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNEdge"): kante = edge.get("bpmnElement") punkte = [(float(wp.get("x")), float(wp.get("y"))) for wp in edge.findall(f"{{{m.NS['di']}}}waypoint")] label = edge.find(f"{{{m.NS['bpmndi']}}}BPMNLabel") for gid, (x, y, w, h) in gw_labels.items(): if label is not None and _ueberlappen(_box(label), (x, y, w, h)): konflikte.append((gid, f"Beschriftung von {kante}")) for (x1, y1), (x2, y2) in zip(punkte, punkte[1:]): schritte = max(8, int(math.hypot(x2 - x1, y2 - y1) / 4)) if any(x < x1 + (x2 - x1) * s / schritte < x + w and y < y1 + (y2 - y1) * s / schritte < y + h for s in range(1, schritte)): konflikte.append((gid, kante)) break for gid, box in gw_labels.items(): konflikte += [(gid, eid) for eid, form in formen.items() if eid != gid and _ueberlappen(box, form)] return sorted(konflikte) @pytest.mark.parametrize("md", MODELLE, ids=lambda p: p.stem) def test_gateway_beschriftung_wird_von_nichts_geschnitten(md): """Prüft, dass jede Gateway-Beschriftung eine feste Position hat und weder von einer Kante noch von einer Kantenbeschriftung oder einem anderen Element geschnitten wird (Issue #14; PICTURE übernimmt die Position).""" defs = baue(md) art = {el.get("id"): el.tag.rpartition("}")[2] for el in defs.iter() if el.get("id")} ohne = [s.get("bpmnElement") for s in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape") if art.get(s.get("bpmnElement")) in GATEWAYS and s.find(f"{{{m.NS['bpmndi']}}}BPMNLabel") is None and next(el for el in defs.iter() if el.get("id") == s.get("bpmnElement")).get("name")] assert ohne == [], "Gateway mit Namen, aber ohne Beschriftungsposition: " + ", ".join(ohne) konflikte = beschriftungskonflikte(defs) assert konflikte == [], "Gateway-Beschriftung geschnitten: " + ", ".join(f"{g} ← {k}" for g, k in konflikte) @pytest.mark.parametrize("md", [md for md in MODELLE if md.stem != "beispiel_prozess"], ids=lambda p: p.stem) def test_gateway_beschriftung_bleibt_in_der_eigenen_lane(md): """Prüft, dass jede Gateway-Beschriftung vollständig in der Lane ihres Gateways liegt und nicht in den Lane-Kopf mit den Lane-Namen oder in eine Nachbar-Lane ragt (Issue #14).""" defs = baue(md) lanes = [_box(s) for s in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape") if s.get("bpmnElement").startswith("Lane_")] assert lanes, "Modell ohne Lanes" for shape in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape"): label = shape.find(f"{{{m.NS['bpmndi']}}}BPMNLabel") if label is None: continue x, y, w, h = _box(shape) lx, ly, lw, lh = _box(label) lane = next(b for b in lanes if b[1] <= y + h / 2 <= b[1] + b[3]) assert lane[1] <= ly and ly + lh <= lane[1] + lane[3], f"{shape.get('bpmnElement')}: ragt aus der Lane" assert lx >= lane[0] + m.LANE_HEADER_WIDTH, f"{shape.get('bpmnElement')}: ragt in den Lane-Kopf" MIT_ANMERKUNGEN = [md for md in MODELLE if " Hinweis:" in md.read_text(encoding="utf-8")] def test_es_gibt_ein_modell_mit_anmerkungen(): """Prüft, dass die Anmerkungs-Tests tatsächlich an einem Modell laufen.""" assert MIT_ANMERKUNGEN @pytest.mark.parametrize("md", MIT_ANMERKUNGEN, ids=lambda p: p.stem) def test_anmerkungen_ueberlappen_nichts_und_stehen_neben_dem_pool(md): """Prüft, dass Anmerkungen (Hinweise) sich nicht gegenseitig überlappen, kein anderes Element überdecken und – wenn es einen Prozess-Pool gibt – rechts daneben stehen, also keine Lane-Grenze kreuzen (Issue #14).""" defs = baue(md) art = {el.get("id"): el.tag.rpartition("}")[2] for el in defs.iter() if el.get("id")} formen = {s.get("bpmnElement"): _box(s) for s in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape")} anmerkungen = {eid: box for eid, box in formen.items() if art.get(eid) == "textAnnotation"} ids = sorted(anmerkungen) for i, a in enumerate(ids): for b in ids[i + 1:]: assert not _ueberlappen(anmerkungen[a], anmerkungen[b]), f"{a} überlappt {b}" for eid, form in formen.items(): if art.get(eid) in HINDERNISSE: assert not _ueberlappen(anmerkungen[a], form), f"{a} überdeckt {eid}" if "Participant_1" in formen: px, _, pw, _ = formen["Participant_1"] for eid, (x, _, _, _) in anmerkungen.items(): assert x >= px + pw, f"{eid} steht im Pool (x {x:g} < {px + pw:g})" @pytest.mark.parametrize("md", MIT_ANMERKUNGEN, ids=lambda p: p.stem) def test_anmerkung_ist_gross_genug_fuer_ihren_text(md): """Prüft, dass jedes Anmerkungs-Kästchen seinen Text fassen kann (grob: 7 Einheiten je Zeichen, 16 je Zeile). Sonst vergrößert PICTURE das Kästchen beim Import, und es überlappt Nachbarn und Lane-Grenzen (Issue #14).""" defs = baue(md) formen = {s.get("bpmnElement"): _box(s) for s in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape")} for anm in defs.iter(f"{{{m.NS['bpmn']}}}textAnnotation"): text = anm.find(f"{{{m.NS['bpmn']}}}text").text or "" _, _, w, h = formen[anm.get("id")] zeilen = math.ceil(len(text) * 7 / (w - 10)) assert h >= zeilen * 16, f"{anm.get('id')}: {len(text)} Zeichen brauchen etwa {zeilen} Zeilen, Höhe nur {h:g}" RECHTECKE = HINDERNISSE[len(GATEWAYS + EREIGNISSE):] # Aufgaben, Teilprozesse, Datenelemente def schraege_anschluesse(defs): """Alle Kantenenden an Rechteck-Elementen, deren erstes bzw. letztes Stück nicht rechtwinklig zu der Elementkante verläuft, auf der der Anschlusspunkt liegt (Ost/West verlangt waagrecht, Nord/Süd senkrecht).""" art = {el.get("id"): el.tag.rpartition("}")[2] for el in defs.iter() if el.get("id")} enden = {el.get("id"): (el.get("sourceRef"), el.get("targetRef")) for el in defs.iter() if art.get(el.get("id")) in KANTEN} formen = {s.get("bpmnElement"): _box(s) for s in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape")} funde = [] for edge in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNEdge"): kante = edge.get("bpmnElement") punkte = [(float(wp.get("x")), float(wp.get("y"))) for wp in edge.findall(f"{{{m.NS['di']}}}waypoint")] if kante not in enden or len(punkte) < 2: continue for element, (a, b) in ((enden[kante][0], punkte[:2]), (enden[kante][1], punkte[:-3:-1])): if art.get(element) not in RECHTECKE: continue x, y, w, h = formen[element] if abs(a[0] - x) < 0.5 or abs(a[0] - x - w) < 0.5: rechtwinklig = abs(a[1] - b[1]) < 1e-6 # Ost/West: waagrecht elif abs(a[1] - y) < 0.5 or abs(a[1] - y - h) < 0.5: rechtwinklig = abs(a[0] - b[0]) < 1e-6 # Nord/Süd: senkrecht else: rechtwinklig = False # Anschluss nicht auf dem Rand if not rechtwinklig: funde.append((kante, element)) return sorted(funde) @pytest.mark.parametrize("md", MODELLE, ids=lambda p: p.stem) def test_kanten_setzen_an_aufgaben_rechtwinklig_an(md): """Prüft, dass jede Kante an einer Aufgabe (bzw. einem anderen Rechteck-Element) rechtwinklig zu der Kante ansetzt, auf der ihr Anschlusspunkt liegt, statt an ihr entlangzulaufen (Issue #23).""" funde = schraege_anschluesse(baue(md)) assert funde == [], "nicht rechtwinklig (Kante, Element): " + ", ".join(f"{k} → {e}" for k, e in funde) # Belastungstest (#12, #22): kleine Modelle, in die je eine zusätzliche Kante "zurück -> " eingefügt # wird – für alle Paare, ohne Lanes, mit einer Lane und mit zwei abwechselnden Lanes. Querverbindungen zwischen # Geschwisterzweigen verschieben dabei das Layout (eine Aufgabe rutscht eine Ebene tiefer). BELASTUNGSMODELLE = { "drei": ["S1 Start: A", " -> S2", "S2 Gateway XOR: W?", " a -> S3", " b -> S4", " c -> S5", "S3 Task: Links", " -> S6", "S4 Task: Mitte", " -> S6", "S5 Task: Rechts", " -> S6", "S6 Gateway XOR: Z", " -> S7", "S7 Task: Danach", " -> S8", "S8 Ende: E"], "parallel": ["S1 Start: A", " -> S2", "S2 Task: Erfassen", " -> S3", "S3 Gateway AND: T", " -> S4", " -> S5", "S4 Task: L", " -> S6", "S5 Task: R", " -> S6", "S6 Gateway AND: Z", " -> S7", "S7 Gateway XOR: Ok?", " Ja -> S8", "S8 Ende: E"], } def _mit_lanes(zeilen, lane_von): return [z.replace(": ", f": {lane_von(z.split()[0])}: ", 1) if z[:1] == "S" else z for z in zeilen] def _belastungsfaelle(): faelle = [] for name, zeilen in BELASTUNGSMODELLE.items(): ids = [z.split()[0] for z in zeilen if z[:1] == "S"] for quelle in ids[:-1]: for ziel in ids[1:ids.index(quelle) + 1]: mit = [] for z in zeilen: mit.append(z) if z.startswith(quelle + " "): mit.append(f" zurück -> {ziel}") for variante, zz in (("ohne", mit), ("1lane", _mit_lanes(mit, lambda s: "A")), ("2lanes", _mit_lanes(mit, lambda s: "A" if int(s[1:]) % 2 else "B"))): faelle.append(pytest.param("\n".join(zz), id=f"{name}-{quelle}-{ziel}-{variante}")) return faelle @pytest.mark.parametrize("text", _belastungsfaelle()) def test_belastung_keine_kante_durch_ein_element_und_rechtwinklig(text): """Prüft an allen Varianten des Belastungstests, dass keine Kante durch ein Element läuft und jede Kante an Aufgaben rechtwinklig ansetzt – auch wenn eine Querverbindung zwischen Zweigen das Layout verschiebt (Issue #22).""" defs = m.build_bpmn(*m.parse_markdown(text)) assert kreuzungen(defs) == [] assert schraege_anschluesse(defs) == [] def test_beispiel_aus_issue_22_ohne_kreuzung(): """Prüft das Beispiel aus Issue #22: Sprung von hinten in den rechten Zweig; die Kante vom Gateway zu diesem Zweig läuft nicht durch die linke Aufgabe.""" zeilen = list(BELASTUNGSMODELLE["drei"]) zeilen.insert(zeilen.index("S7 Task: Danach") + 1, " zurück -> S5") text = "\n".join(zeilen) assert kreuzungen(m.build_bpmn(*m.parse_markdown(text))) == [] def test_kreuzungspruefung_findet_eine_kante_durch_eine_aufgabe(): """Prüft die Prüfung selbst: eine absichtlich durch eine Aufgabe gelegte Kante wird gefunden, eine daneben nicht.""" defs = baue(SKILL / "beispiele" / "beispiel_prozess.md") task_shape = next(s for s in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape") if s.get("bpmnElement").startswith("Task_")) b = task_shape.find(f"{{{m.NS['dc']}}}Bounds") x, y, w, h = (float(b.get(a)) for a in ("x", "y", "width", "height")) edge = next(e for e in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNEdge") if e.get("bpmnElement").startswith("Flow_")) wp = edge.findall(f"{{{m.NS['di']}}}waypoint") wp[0].set("x", str(x - 20)); wp[0].set("y", str(y + h / 2)) wp[-1].set("x", str(x + w + 20)); wp[-1].set("y", str(y + h / 2)) for mitte in wp[1:-1]: edge.remove(mitte) assert (edge.get("bpmnElement"), task_shape.get("bpmnElement")) in kreuzungen(defs) wp[0].set("y", str(y - 30)); wp[-1].set("y", str(y - 30)) assert (edge.get("bpmnElement"), task_shape.get("bpmnElement")) not in kreuzungen(defs)