Projektwissen/.agents/skills/sn-bpmn/test_md_to_bpmn.py

# -*- coding: utf-8 -*-
"""Tests für md_to_bpmn.py - reines Python/pytest, ohne Netzwerk."""
import xml.etree.ElementTree as ET

import pytest

import md_to_bpmn as m


def test_bestehende_syntax_bleibt_kompatibel():
    """Prüft, dass Start/Task/Gateway-XOR/Ende mit Verzweigung wie bisher geparst werden."""
    text = """
    S1 Start: Beginn
      -> S2
    S2 Task: Antrag pruefen
    S3 Gateway XOR: Vollständig?
      Ja -> S4
      Nein -> S2
    S4 Ende: Fertig
    """
    meta, steps, order, externals = m.parse_markdown(text)
    assert externals == []
    assert steps['S1'].type == 'StartEvent'
    assert steps['S3'].type == 'XorGateway'
    assert ('Ja', 'S4') in steps['S3'].edges
    defs = m.build_bpmn(meta, steps, order, externals)
    assert defs.find(f".//{{{m.NS['bpmn']}}}exclusiveGateway") is not None


def test_lane_erkennung_bleibt_unveraendert():
    """Prüft, dass 'Verantwortlicher: Rest' weiterhin als Swimlane erkannt wird."""
    text = """
    S1 Start: Beginn
      -> V1
    V1 Task: Flurneuordnung: Gewannenkonzept festlegen
    O3 Task: ÖbVI: Waldgrenzen verhandeln
    S_end Ende: Fertig
    """
    meta, steps, order, externals = m.parse_markdown(text)
    lanes = {s.lane for s in steps.values() if s.lane}
    assert lanes == {'Flurneuordnung', 'ÖbVI'}


@pytest.mark.parametrize('typewort, erwarteter_typ', [
    ('Manuelle Aufgabe', 'ManualTask'),
    ('User Aufgabe', 'UserTask'),
    ('Service Aufgabe', 'ServiceTask'),
    ('Script Aufgabe', 'ScriptTask'),
    ('Geschäftsregel Aufgabe', 'BusinessRuleTask'),
    ('Sendeaufgabe', 'SendTask'),
    ('Empfangsaufgabe', 'ReceiveTask'),
    ('Teilprozess', 'SubProcess'),
    ('Aufruf', 'CallActivity'),
    ('Gateway Ereignisbasiert', 'EventBasedGateway'),
    ('Datenobjekt', 'DataObject'),
    ('Datenspeicher', 'DataStore'),
    ('Gruppe', 'Group'),
])
def test_neue_elementtypen_werden_aufgeloest(typewort, erwarteter_typ):
    """Prüft, dass alle neuen Aufgaben-/Gateway-/Artefakttypen korrekt zugeordnet werden."""
    typ, kind, direction = m.resolve_type(typewort, 'Testzeile')
    assert typ == erwarteter_typ


def test_typisiertes_startereignis_erzeugt_eventdefinition():
    """Prüft, dass ein typisiertes Startereignis das passende BPMN-eventDefinition-Kind erhaelt."""
    text = """
    S1 Start Nachricht: Antrag eingegangen
      -> S2
    S2 Ende: Fertig
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    assert defs.find(f".//{{{m.NS['bpmn']}}}messageEventDefinition") is not None


def test_zwischenereignis_sendend_wird_throw_event():
    """Prüft, dass 'sendend' bei einem Zwischenereignis ein intermediateThrowEvent erzeugt."""
    text = """
    S1 Start: Beginn
      -> S2
    S2 Zwischenereignis Nachricht sendend: Status melden
      -> S3
    S3 Ende: Fertig
    """
    meta, steps, order, externals = m.parse_markdown(text)
    assert m.bpmn_tag_for(steps['S2']) == 'intermediateThrowEvent'


def test_terminierung_nur_als_endereignis_erlaubt():
    """Prüft, dass 'Terminierung' bei einem Startereignis einen Fehler ausloest."""
    text = """
    S1 Start Terminierung: X
      -> S2
    S2 Ende: Y
    """
    with pytest.raises(ValueError, match='Terminierung'):
        m.parse_markdown(text)


def test_link_nur_als_zwischenereignis_erlaubt():
    """Prüft, dass 'Link' bei einem Endereignis einen Fehler ausloest."""
    text = """
    S1 Start: X
      -> S2
    S2 Ende Link: Y
    """
    with pytest.raises(ValueError, match='Link'):
        m.parse_markdown(text)


def test_ereignisbasiertes_gateway_verbietet_zusammenfuehrung():
    """Prüft, dass ein ereignisbasiertes Gateway mit mehreren eingehenden Kanten abgelehnt wird."""
    text = """
    S1 Start: X
      -> S2
      -> S3
    S2 Task: A
      -> S4
    S3 Task: B
      -> S4
    S4 Gateway Ereignisbasiert: C
      -> S5
    S5 Ende: Y
    """
    with pytest.raises(ValueError, match='Zusammenführung'):
        m.parse_markdown(text)


def test_datenobjekt_bekommt_keinen_sequenzfluss():
    """Prüft, dass ein Sequenzfluss zu einem Datenobjekt abgelehnt wird."""
    text = """
    S1 Start: X
      -> D1
    D1 Datenobjekt: Akte
    S2 Ende: Y
    """
    with pytest.raises(ValueError, match='keine Ablaufelemente'):
        m.parse_markdown(text)


def test_datenobjekt_wird_nicht_automatisch_verkettet():
    """Prüft, dass die automatische Verkettung über unverbundene Artefakte hinwegspringt."""
    text = """
    S1 Start: X
    D1 Datenobjekt: Akte
    S2 Ende: Y
    """
    meta, steps, order, externals = m.parse_markdown(text)
    assert steps['S1'].edges == [(None, 'S2')]
    assert steps['D1'].edges == []


def test_datenobjekt_und_gruppe_erzeugen_erwartete_elemente():
    """Prüft, dass Datenobjekt und Gruppe als eigenstaendige, unverbundene Artefakte gerendert werden."""
    text = """
    S1 Start: X
      -> S2
    S2 Ende: Y
    D1 Datenobjekt: Akte
    G1 Gruppe: Fachliche Pruefung
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    assert defs.find(f".//{{{m.NS['bpmn']}}}dataObjectReference") is not None
    assert defs.find(f".//{{{m.NS['bpmn']}}}group") is not None
    assert defs.find(f".//{{{m.NS['bpmn']}}}categoryValue") is not None


def test_externer_beteiligter_erzeugt_messageflow():
    """Prüft, dass ein deklarierter externer Beteiligter mit Nachrichtenfluss korrekt verknuepft wird."""
    text = """
    Extern: Bürger

    S1 Start: X
      -> S2
    S2 SendTask: Bescheid senden
      ~> Bürger
      -> S3
    S3 Ende: Y
    """
    meta, steps, order, externals = m.parse_markdown(text)
    assert externals == ['Bürger']
    defs = m.build_bpmn(meta, steps, order, externals)
    msg_flow = defs.find(f".//{{{m.NS['bpmn']}}}messageFlow")
    assert msg_flow is not None
    participants = defs.findall(f".//{{{m.NS['bpmn']}}}participant")
    assert any(p.get('name') == 'Bürger' and p.get('processRef') is None for p in participants)


def test_unbekannter_externer_beteiligter_wird_abgelehnt():
    """Prüft, dass ein Nachrichtenfluss zu einem nicht deklarierten externen Beteiligten scheitert."""
    text = """
    S1 Start: X
      ~> Unbekannt
    S2 Ende: Y
    """
    with pytest.raises(ValueError, match='unbekannter externer Beteiligter'):
        m.parse_markdown(text)


def test_verteilte_anschlusspunkte_liegen_auf_der_rautenkontur():
    """Prüft, dass verteilte Anschlusspunkte an einem Gateway auf der Diamant-Kontur liegen, nicht auf der Bounding-Box."""
    x, y, w, h = 590, 2290, 50, 50
    cx, cy = x + w / 2, y + h / 2
    for idx in range(4):
        px, py = m.spread_point(x, y, w, h, 'S', idx, 4, 'diamond')
        # Manhattan-Abstand zum Mittelpunkt, skaliert auf die Diamant-Halbachsen, muss 1 ergeben.
        assert abs(abs(px - cx) / (w / 2) + abs(py - cy) / (h / 2) - 1) < 1e-9


def test_verteilte_anschlusspunkte_liegen_auf_dem_ereigniskreis():
    """Prüft, dass verteilte Anschlusspunkte an einem Ereignis auf dem Kreisradius liegen, nicht auf der Bounding-Box."""
    x, y, w, h = 100, 100, 36, 36
    cx, cy = x + w / 2, y + h / 2
    r = w / 2
    for idx in range(3):
        px, py = m.spread_point(x, y, w, h, 'S', idx, 3, 'circle')
        dist = ((px - cx) ** 2 + (py - cy) ** 2) ** 0.5
        assert abs(dist - r) < 1e-9


def test_gateway_kanten_bekommen_eigene_kardinalpunkte():
    """Prüft, dass bis zu vier Kanten an einem Gateway je einen eigenen Eckpunkt (N/O/S/W) bekommen, wie PICTURE es rendert."""
    text = """
    S1 Start: X
      -> S2
    S2 Gateway XOR: Auswahl
      A -> S3
      B -> S4
      C -> S5
    S3 Ende: A
    S4 Ende: B
    S5 Ende: C
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    ns = m.NS
    gw = next(el for el in defs.iter(f"{{{ns['bpmn']}}}exclusiveGateway") if el.get('name') == 'Auswahl')
    shape = defs.find(f".//{{{ns['bpmndi']}}}BPMNShape[@bpmnElement='{gw.get('id')}']")
    x, y, w, h = (float(shape[0].get(a)) for a in ('x', 'y', 'width', 'height'))
    vertices = {(x + w / 2, y), (x + w / 2, y + h), (x, y + h / 2), (x + w, y + h / 2)}

    out_ids = {o.text for o in gw.findall(f"{{{ns['bpmn']}}}outgoing")}
    in_ids = {i.text for i in gw.findall(f"{{{ns['bpmn']}}}incoming")}
    used = set()
    for fid in out_ids | in_ids:
        edge = defs.find(f".//{{{ns['bpmndi']}}}BPMNEdge[@bpmnElement='{fid}']")
        wps = edge.findall(f"{{{ns['di']}}}waypoint")
        wp = wps[0] if fid in out_ids else wps[-1]
        point = (float(wp.get('x')), float(wp.get('y')))
        assert point in vertices
        used.add(point)
    assert len(used) == 4


def test_gateway_mit_mehr_als_vier_kanten_bleibt_gueltiges_xml():
    """Prüft, dass ein Gateway mit mehr als vier anliegenden Kanten (über PICTUREs Grenze hinaus) trotzdem gültiges XML ohne Absturz erzeugt."""
    text = """
    S1 Start: X
      -> S2
    S2 Gateway XOR: Fuenf-Wege
      A -> S3
      B -> S4
      C -> S5
      D -> S6
      E -> S7
    S3 Ende: A
    S4 Ende: B
    S5 Ende: C
    S6 Ende: D
    S7 Ende: E
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    ET.tostring(defs)  # loest bei kaputter Struktur eine Exception aus


def test_mehrere_kanten_an_derselben_seite_bekommen_eigenen_anschlusspunkt():
    """Prüft, dass mehrere ausgehende Kanten an einem Gateway nicht im selben Punkt starten."""
    text = """
    S1 Start: X
      -> S2
    S2 Gateway XOR: Auswahl
      A -> S3
      B -> S4
      C -> S5
    S3 Ende: A
    S4 Ende: B
    S5 Ende: C
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    starts = []
    for edge in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNEdge"):
        wp = edge.find(f"{{{m.NS['di']}}}waypoint")
        starts.append((wp.get('x'), wp.get('y')))
    # 3 der 4 Kanten gehen vom selben Gateway aus - deren Startpunkte muessen sich unterscheiden.
    assert len(set(starts)) == len(starts)


def _liegt_in_rautenflaeche(px, py, x, y, w, h, sicherheitsabstand=0.5):
    """Manhattan-Abstand zum Mittelpunkt, skaliert auf die Diamant-Halbachsen: < 1 heisst innerhalb."""
    cx, cy = x + w / 2, y + h / 2
    wert = abs(px - cx) / (w / 2) + abs(py - cy) / (h / 2)
    return wert < 1 - (sicherheitsabstand / max(w, h))


def _kante_kreuzt_gateway(defs, ns):
    """Liefert (Kanten-Id, Gateway-Id) für die erste Kante, deren Streckenzug durchs Innere
    eines Gateways (statt nur an dessen Kontur entlang) laeuft - oder None, wenn keine kreuzt."""
    shapes = {}
    for shape in defs.iter(f"{{{ns['bpmndi']}}}BPMNShape"):
        b = shape.find(f"{{{ns['dc']}}}Bounds")
        if b is not None:
            shapes[shape.get('bpmnElement')] = tuple(float(b.get(a)) for a in ('x', 'y', 'width', 'height'))

    gateway_ids = set()
    for tag in ('exclusiveGateway', 'inclusiveGateway', 'parallelGateway', 'eventBasedGateway'):
        gateway_ids.update(el.get('id') for el in defs.iter(f"{{{ns['bpmn']}}}{tag}"))

    for edge in defs.iter(f"{{{ns['bpmndi']}}}BPMNEdge"):
        pts = [(float(wp.get('x')), float(wp.get('y'))) for wp in edge.findall(f"{{{ns['di']}}}waypoint")]
        for gid in gateway_ids:
            if gid not in shapes:
                continue
            x, y, w, h = shapes[gid]
            for (x1, y1), (x2, y2) in zip(pts, pts[1:]):
                for s in range(41):
                    t = s / 40
                    if _liegt_in_rautenflaeche(x1 + (x2 - x1) * t, y1 + (y2 - y1) * t, x, y, w, h):
                        return edge.get('bpmnElement'), gid
    return None


def test_kante_fuehrt_nicht_durchs_innere_eines_gateways():
    """Prüft, dass eine Kante von/zu einem Gateway, dessen Ziel/Quelle nicht auf der Achse des
    zugewiesenen Anschlusspunkts liegt, aussen herum und nicht durchs Innere der Raute fuehrt."""
    text = """
    S1 Start: X
      -> S5
    S5 Gateway XOR: Sofort?
      Ja -> S6
      Nein -> S6m
    S6 Task: Leitungsplaene einholen
      -> S6m
    S6m Gateway XOR: Zusammenfuehrung
      -> S7
    S7 Ende: Fertig
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    assert _kante_kreuzt_gateway(defs, m.NS) is None


def _streckenzug_kreuzt_raute(punkte, raute):
    """True, wenn ein Stück des Streckenzugs durchs Innere der Raute (x, y, w, h) läuft."""
    for (x1, y1), (x2, y2) in zip(punkte, punkte[1:]):
        for s in range(41):
            t = s / 40
            if _liegt_in_rautenflaeche(x1 + (x2 - x1) * t, y1 + (y2 - y1) * t, *raute):
                return True
    return False


def test_kante_zwischen_gateways_zum_nordpunkt_eines_hoeheren_ziels_umgeht_die_raute():
    """Prüft, dass eine Kante vom Südpunkt eines Gateways zum Nordpunkt eines höher liegenden Gateways
    seitlich an der Ziel-Raute vorbei nach oben führt statt senkrecht durch sie hindurch (Issue #8)."""
    quelle, ziel = (2200, 145, 50, 50), (2390, 60, 50, 50)      # Lage wie S12 -> S13 im Referenzmodell
    punkte = m.orthogonal_points(2225, 195, 'S', 2415, 60, 'N', src_is_gateway=True, tgt_is_gateway=True)
    assert punkte[0] == (2225, 195) and punkte[-1] == (2415, 60)
    assert not _streckenzug_kreuzt_raute(punkte, ziel)
    assert not _streckenzug_kreuzt_raute(punkte, quelle)
    assert all(a[0] == b[0] or a[1] == b[1] for a, b in zip(punkte, punkte[1:])), "nicht rechtwinklig"


def test_ausweichweg_zwischen_gateways_meidet_die_spalte_dazwischen():
    """Prüft, dass der seitliche Umweg um die Ziel-Raute im freien Gang direkt vor der Ziel-Spalte liegt und
    nicht mitten zwischen Quelle und Ziel, wo in einer Spalte dazwischen eine Aufgabe stehen kann."""
    aufgabe, ziel = (2960, 215, 100, 80), (3150, 230, 50, 50)   # Lage wie S43 und S44 im Referenzmodell
    punkte = m.orthogonal_points(2795, 280, 'S', 3175, 230, 'N', src_is_gateway=True, tgt_is_gateway=True)
    assert not _streckenzug_kreuzt_raute(punkte, ziel)
    x, y, w, h = aufgabe
    for (x1, y1), (x2, y2) in zip(punkte, punkte[1:]):
        for s in range(41):
            px, py = x1 + (x2 - x1) * s / 40, y1 + (y2 - y1) * s / 40
            assert not (x < px < x + w and y < py < y + h), f"Kante läuft bei ({px:g}, {py:g}) durch die Aufgabe"


def test_kante_zwischen_gateways_zum_westpunkt_eines_linken_ziels_umgeht_die_raute():
    """Prüft den waagrechten Spiegelfall: vom Ostpunkt eines Gateways zum Westpunkt eines weiter links
    liegenden Gateways führt die Kante ober- oder unterhalb an der Ziel-Raute vorbei."""
    quelle, ziel = (400, 300, 50, 50), (200, 100, 50, 50)
    punkte = m.orthogonal_points(450, 325, 'E', 200, 125, 'W', src_is_gateway=True, tgt_is_gateway=True)
    assert punkte[0] == (450, 325) and punkte[-1] == (200, 125)
    assert not _streckenzug_kreuzt_raute(punkte, ziel)
    assert not _streckenzug_kreuzt_raute(punkte, quelle)
    assert all(a[0] == b[0] or a[1] == b[1] for a, b in zip(punkte, punkte[1:])), "nicht rechtwinklig"


def test_nur_xor_gateways_zeigen_ihr_x():
    """Prüft, dass die Diagrammform jedes exklusiven Gateways isMarkerVisible="true" trägt, damit das „X“ erscheint,
    und dass kein anderes Element das Attribut bekommt (Issue #15)."""
    text = """
    S1 Start: Beginn
      -> S2
    S2 Gateway XOR: Vollständig?
      Ja -> S3
      Nein -> S4
    S3 Gateway AND: Aufteilen
      -> S5
    S4 Gateway OR: Welche?
      -> S5
    S5 Task: Bearbeiten
      -> S6
    S6 Ende: Fertig
    """
    meta, steps, order, externals = m.parse_markdown(text)
    defs = m.build_bpmn(meta, steps, order, externals)
    xor_ids = {g.get('id') for g in defs.iter(f"{{{m.NS['bpmn']}}}exclusiveGateway")}
    assert xor_ids, "Testmodell ohne XOR-Gateway"
    for shape in defs.iter(f"{{{m.NS['bpmndi']}}}BPMNShape"):
        erwartet = 'true' if shape.get('bpmnElement') in xor_ids else None
        assert shape.get('isMarkerVisible') == erwartet, shape.get('bpmnElement')


def test_ausgabe_ist_wohlgeformtes_xml(tmp_path):
    """Prüft, dass die erzeugte .bpmn-Datei sich als valides XML parsen laesst."""
    src = tmp_path / 'prozess.md'
    src.write_text("""
    Extern: Bürger

    S1 Start Nachricht: Antrag geht ein
      -> S2
    S2 Gateway XOR: Vollständig?
      Ja -> S3
      Nein -> S2
    S3 SendTask: Bescheid senden
      ~> Bürger
      -> S4
    S4 Ende Terminierung: Fertig
    """, encoding='utf-8')
    out = tmp_path / 'prozess.bpmn'

    meta, steps, order, externals = m.parse_markdown(src.read_text(encoding='utf-8'))
    defs = m.build_bpmn(meta, steps, order, externals)
    out.write_bytes(m.prettify(defs))

    parsed = ET.parse(out)
    assert parsed.getroot().tag == f"{{{m.NS['bpmn']}}}definitions"