Merge pull request 'verfahrensgebiet aus Flst' (#12) from verfahrensgebiet_aus_flst into unstable

Reviewed-on: #12

changelog.txt

@@ -1,4 +1,10 @@
 ---
+Version 26.4.5-unstable:
+---
+Version 26.4.4-unstable:
+---
+Version 26.4.3-unstable:
+---
 Version 26.4.2-testing:
 ---
 Version 26.4.1-unstable:

functions/flurstueck_alkis.py

@@ -1,20 +1,40 @@
 """sn_verfahrensgebiet/functions/flurstueck_alkis.py – ALKIS-Integration für Flurstücke.
 
-Lädt Flurstücke aus dem ALKIS-WFS-Dienst (adv:AX_Flurstueck) anhand von
+Lädt Flurstücke aus dem ALKIS-WFS-Dienst (``adv:AX_Flurstueck``) anhand von
 Gemarkungsschlüsseln, die der Nutzer interaktiv eingibt.
 
 Der WFS-Layer wird direkt (nicht als Memory-Layer) ins Projekt geladen,
 da die einzelnen Features für die spätere Auswahl benötigt werden.
+
+**Erweiterung gegenüber dem Original:**
+
+- ``vorgabe_schluessel``: Optional vorausgefüllte Gemarkungsnummern (z. B. aus
+  :func:`~sn_verfahrensgebiet.functions.gemarkung_alkis.gemarkung_schluessel_und_namen_by_bbox`).
+  Der Dialog wird mit diesem Vorschlag vorausgefüllt; der Nutzer kann die
+  Eingabe ändern oder ergänzen.
+
+- ``vorgabe_namen``: Zugehörige Bezeichnungen, die im Dialog als Hinweis
+  angezeigt werden (kein eigenständiges Feld, nur Anzeigetext).
+
+- ``bbox``: Optionaler BBox-Parameter, der – sofern übergeben – als
+  zusätzlicher serverseitiger Filter in die WFS-URI eingebettet wird. So
+  werden ausschließlich Flurstücke geladen, die den Verfahrensgebiets-Rahmen
+  überlappen.
 """
 
-from sn_basis.functions import QgsVectorLayer, QgsProject
+from __future__ import annotations
+
+from typing import List, Optional, Tuple
+
+from sn_basis.functions import QgsVectorLayer, QgsProject, QgsFeature
 from sn_basis.functions import QMessageBox
 from sn_basis.functions.dialog_wrapper import ask_text
+from sn_basis.functions.message_wrapper import info as _info
 
 # LoadStatus aus verfahrensgebiet_alkis wiederverwenden (identische Semantik)
 from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus
 
-ALKIS_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs"
+ALKIS_SF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
 FLST_TYPENAME = "adv:AX_Flurstueck"
 FLST_LAYER_NAME = "Flurstücke"
 
@@ -49,65 +69,173 @@ def _check_existing_flurstueck_layer(tab_widget):
         return existing_layers[0], LoadStatus.KEEP
 
 
-def flurstueck_alkis(tab_widget):
-    """Lädt Flurstücke aus ALKIS-WFS anhand von Gemarkungsschlüsseln.
+def _lade_flurstuecke_per_gemarkung(schluessel_liste: List[str]):
+    """Lädt Flurstücke für mehrere Gemarkungen, eine WFS-Anfrage pro Schlüssel.
 
-    Der Nutzer gibt einen oder mehrere Gemarkungsschlüssel (kommagetrennt) ein.
-    Der WFS wird per SQL-Filter auf
-    ``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN (...)`` abgefragt.
+    Vermeidet das WFS-Feature-Limit einer einzelnen Abfrage mit mehreren
+    Gemarkungsnummern. Alle Features werden in einem Memory-Layer zusammengeführt.
+
+    Parameters
+    ----------
+    schluessel_liste : list of str
+        4-stellige Gemarkungsnummern.
+
+    Returns
+    -------
+    QgsVectorLayer or None
+        Memory-Layer mit allen Flurstücken, oder ``None`` wenn nichts geladen wurde.
+    """
+    alle_feats: list = []
+    schema_fields = None
+    schema_crs = "EPSG:25833"
+
+    for schluessel in schluessel_liste:
+        uri = (
+            f"url='{ALKIS_SF_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' "
+            f"sql=SELECT * FROM AX_Flurstueck "
+            f"WHERE gemarkung_ax_gemarkung_schluessel_gemarkungsnummer = '{schluessel}'"
+        )
+        wfs = QgsVectorLayer(uri, f"_TmpFlst{schluessel}", "WFS")
+        if not wfs.isValid():
+            continue
+        feats = list(wfs.getFeatures())
+        if not feats:
+            continue
+        if schema_fields is None:
+            schema_fields = wfs.fields()
+            schema_crs = wfs.crs().authid()
+        _info(
+            "Flurstücke",
+            f"Gemarkung {schluessel}: {len(feats)} Flurstücke geladen.",
+            duration=3,
+        )
+        alle_feats.append((wfs.fields(), feats))
+
+    if not alle_feats or schema_fields is None:
+        return None
+
+    # Memory-Layer mit dem Feldschema des ersten WFS-Layers aufbauen
+    mem = QgsVectorLayer(f"Polygon?crs={schema_crs}", FLST_LAYER_NAME, "memory")
+    pr = mem.dataProvider()
+    pr.addAttributes(list(schema_fields))
+    mem.updateFields()
+    mem_fields = mem.fields()
+
+    neue_features = []
+    for wfs_fields, feats in alle_feats:
+        for feat in feats:
+            new_f = QgsFeature(mem_fields)
+            new_f.setGeometry(feat.geometry())
+            for i in range(wfs_fields.count()):
+                fname = wfs_fields.field(i).name()
+                idx = mem_fields.indexFromName(fname)
+                if idx >= 0:
+                    try:
+                        new_f.setAttribute(idx, feat[fname])
+                    except Exception:
+                        pass
+            neue_features.append(new_f)
+
+    pr.addFeatures(neue_features)
+    mem.updateExtents()
+    return mem
+
+
+def flurstueck_alkis(
+    tab_widget,
+    vorgabe_schluessel: Optional[List[str]] = None,
+    vorgabe_namen: Optional[List[str]] = None,
+    bbox=None,
+    ohne_dialog: bool = False,
+) -> Tuple[Optional[QgsVectorLayer], LoadStatus]:
+    """Lädt Flurstücke aus dem ALKIS-WFS anhand von Gemarkungsschlüsseln.
+
+    Jede Gemarkungsnummer wird als separate WFS-Anfrage gestellt, um das
+    WFS-Feature-Limit bei einer Gesamtabfrage zu umgehen. Alle Ergebnisse
+    werden in einem Memory-Layer zusammengeführt.
 
     Parameters
     ----------
     tab_widget :
         Eltern-Widget für Dialoge.
+    vorgabe_schluessel : list of str, optional
+        Vorausgefüllte Gemarkungsnummern (4-stellig). Im Dialog als Standard
+        angezeigt; bei ``ohne_dialog=True`` direkt verwendet.
+    vorgabe_namen : list of str, optional
+        Menschenlesbare Gemarkungsbezeichnungen für den Dialog-Hinweistext.
+    bbox :
+        Nicht mehr verwendet (WFS unterstützt keinen BBox-Filter). Bleibt
+        für Rückwärtskompatibilität in der Signatur erhalten.
+    ohne_dialog : bool, optional
+        ``True``: Überspringt die Gemarkungsschlüssel-Nutzerabfrage und verwendet
+        ``vorgabe_schluessel`` direkt. Vorhandene Layer werden ohne Rückfrage
+        ersetzt. Gedacht für die automatische Pipeline
+        (``verfahrensgebiet_alkis_komplett``). Standard: ``False``.
 
     Returns
     -------
     Tuple[Optional[QgsVectorLayer], LoadStatus]
-        Geladener WFS-Layer und Ladestatus.
-        Bei Fehlern ist der Layer None und der Status LoadStatus.NONE.
+        Memory-Layer mit allen Flurstücken und Ladestatus.
+        Bei Fehlern ist der Layer ``None`` und der Status :attr:`LoadStatus.NONE`.
     """
-    # 1. Vorhandenen Layer prüfen
-    existing, status = _check_existing_flurstueck_layer(tab_widget)
-    if status == LoadStatus.KEEP:
-        return existing, status
+    # --- Schritt 1: Vorhandenen Layer behandeln ----------------------------
+    if ohne_dialog:
+        project = QgsProject.instance()
+        hatte_layer = bool(project.mapLayersByName(FLST_LAYER_NAME))
+        for lyr in list(project.mapLayersByName(FLST_LAYER_NAME)):
+            project.removeMapLayer(lyr.id())
+        status = LoadStatus.RELOAD if hatte_layer else LoadStatus.FIRST
+    else:
+        existing, status = _check_existing_flurstueck_layer(tab_widget)
+        if status == LoadStatus.KEEP:
+            return existing, status
 
-    # 2. Gemarkungsschlüssel abfragen
-    eingabe, ok = ask_text(
-        "Gemarkungsschlüssel",
-        "Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):",
-        parent=tab_widget,
-    )
-    if not ok or not eingabe.strip():
-        QMessageBox.warning(
-            tab_widget,
-            "Abgebrochen",
-            "Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.",
+    # --- Schritt 2: Gemarkungsschlüssel bestimmen -------------------------
+    if ohne_dialog:
+        # Pipeline-Modus: vorgabe_schluessel direkt verwenden, kein Dialog
+        if not vorgabe_schluessel:
+            return None, LoadStatus.NONE
+        schluessel_liste = [s.strip() for s in vorgabe_schluessel if s.strip()]
+        if not schluessel_liste:
+            return None, LoadStatus.NONE
+    else:
+        # Interaktiver Modus: Dialog mit Vorschlag
+        default_text = ", ".join(vorgabe_schluessel) if vorgabe_schluessel else ""
+        hinweis = ""
+        if vorgabe_namen:
+            hinweis = " (Vorschlag: " + ", ".join(vorgabe_namen) + ")"
+        label = f"Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):{hinweis}"
+
+        eingabe, ok = ask_text(
+            "Gemarkungsschlüssel",
+            label,
+            default_text=default_text,
+            parent=tab_widget,
         )
-        return None, LoadStatus.NONE
+        if not ok or not eingabe.strip():
+            QMessageBox.warning(
+                tab_widget,
+                "Abgebrochen",
+                "Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.",
+            )
+            return None, LoadStatus.NONE
 
-    schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()]
-    if not schluessel_liste:
-        QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.")
-        return None, LoadStatus.NONE
+        schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()]
+        if not schluessel_liste:
+            QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.")
+            return None, LoadStatus.NONE
 
-    schluessel_str = ", ".join(f"'{s}'" for s in schluessel_liste)
+    # --- Schritt 3: Flurstücke gemarkungsweise laden ----------------------
+    layer = _lade_flurstuecke_per_gemarkung(schluessel_liste)
 
-    # 3. WFS-URI aufbauen und Layer laden
-    wfs_uri = (
-        f"url='{ALKIS_NAS_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' "
-        f"sql=SELECT * FROM AX_Flurstueck "
-        f"WHERE gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN ({schluessel_str})"
-    )
-    layer = QgsVectorLayer(wfs_uri, FLST_LAYER_NAME, "WFS")
-
-    if not layer.isValid():
-        QMessageBox.critical(
-            tab_widget,
-            "Fehler",
-            "Flurstück-Layer konnte nicht geladen werden.\n"
-            "Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.",
-        )
+    if not layer or not layer.isValid() or layer.featureCount() == 0:
+        if not ohne_dialog:
+            QMessageBox.critical(
+                tab_widget,
+                "Fehler",
+                "Flurstück-Layer konnte nicht geladen werden.\n"
+                "Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.",
+            )
         return None, LoadStatus.NONE
 
     return layer, status

functions/gemarkung_alkis.py

@@ -0,0 +1,207 @@
+"""sn_verfahrensgebiet/functions/gemarkung_alkis.py – ALKIS-Gemarkungsabfrage.
+
+Stellt Funktionen bereit, um Katasterbezirk-Objekte aus dem sächsischen
+ALKIS-vereinfachten WFS zu laden und daraus Gemarkungsschlüssel zu extrahieren,
+die als Vorschlag für die Flurstücksabfrage (Phase 2) dienen.
+
+**Dienst:** ``https://geodienste.sachsen.de/aaa/public_alkis/vereinf/wfs``
+**Typename:** ``Katasterbezirk``
+
+Dieser Dienst hat – im Gegensatz zu ``adv:AX_Gemarkung`` aus dem NAS-Dienst –
+tatsächliche Geometrien. Die client-seitige Verschneidung mit der VG-Geometrie
+ist damit möglich.
+
+**Attributkonvention:**
+
+- ``schluessel``: vollständiger Katasterbezirksschlüssel (z.B. ``140640001``),
+  die letzten 4 Stellen entsprechen der
+  ``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer`` im NAS-Flurstücksdienst.
+- ``ave_name``: Gemarkungsbezeichnung.
+
+**Wichtig:** Diese Funktion stellt keine Nutzerabfragen. Sie gibt im
+Fehlerfall eine leere Liste zurück.
+"""
+
+from __future__ import annotations
+
+from typing import List, Optional, Tuple
+
+from sn_basis.functions import QgsVectorLayer, QgsGeometry
+from sn_basis.functions.message_wrapper import warning as _warn, info as _info
+
+ALKIS_VEREINF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/vereinf/wfs"
+KATASTERBEZIRK_TYPENAME = "ave:KatasterBezirk"
+
+# Attributnamen im vereinfachten Dienst
+_ATTR_SCHLUESSEL = "schluessel"   # vollständiger Schlüssel, letzte 4 = Gemarkungsnummer
+_ATTR_NAME = "ave_name"
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _extract_schluessel_und_name(feature) -> Tuple[str, str]:
+    """Extrahiert Gemarkungsnummer (letzte 4 Stellen von ``schluessel``)
+    und Namen (``ave_name``) aus einem Katasterbezirk-Feature.
+
+    Returns
+    -------
+    Tuple[str, str]
+        ``(gemarkungsnummer_4stellig, name)`` – leer wenn nicht ermittelbar.
+    """
+    def _get(field: str) -> str:
+        try:
+            val = feature[field]
+            return str(val).strip() if val else ""
+        except (KeyError, Exception):
+            return ""
+
+    schluessel = _get(_ATTR_SCHLUESSEL)
+    name = _get(_ATTR_NAME)
+
+    # Letzte 4 Stellen des Schlüssels = Gemarkungsnummer
+    nummer = schluessel[-4:] if len(schluessel) >= 4 else schluessel
+
+    return nummer, name
+
+
+def _load_all_gemarkung_layer():
+    """Lädt alle Katasterbezirk-Objekte aus dem vereinfachten ALKIS-WFS.
+
+    Sachsen hat ≈ 2 000 Gemarkungen. Der vereinfachte Dienst liefert
+    tatsächliche Geometrien, sodass client-seitige Verschneidung möglich ist.
+
+    Returns
+    -------
+    QgsVectorLayer or None
+        Geladener Layer oder ``None`` bei Ungültigkeit.
+    """
+    uri = (
+        f"url='{ALKIS_VEREINF_URL}' typename='{KATASTERBEZIRK_TYPENAME}' "
+        f"srsname='EPSG:25833'"
+    )
+    layer = QgsVectorLayer(uri, "_TempKatasterbezirk", "WFS")
+    return layer if (layer and layer.isValid()) else None
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def gemarkung_schluessel_und_namen_von_vg_geometrie(
+    vg_geometrie,
+) -> List[Tuple[str, str]]:
+    """Gibt Gemarkungsschlüssel und -namen aller Gemarkungen zurück, die die
+    VG-Geometrie schneiden.
+
+    Da der WFS-Dienst keine serverseitigen BBox- oder Geometriefilter
+    unterstützt, werden alle ``AX_Gemarkung``-Objekte geladen und client-seitig
+    per ``QgsGeometry.intersects`` gefiltert.
+
+    Der Aufruf ist *ohne Nutzerinteraktion* – im Fehlerfall wird eine leere
+    Liste zurückgegeben, damit der aufrufende Workflow manuellen Fallback
+    ermöglicht.
+
+    Parameters
+    ----------
+    vg_geometrie :
+        ``QgsGeometry`` der Verfahrensgebiets-Union (aus den Roh-WFS-Features
+        von ``AX_BauRaumOderBodenordnungsrecht``). In EPSG:25833.
+
+    Returns
+    -------
+    List[Tuple[str, str]]
+        Liste von ``(gemarkungsnummer, bezeichnung)``-Tupeln, aufsteigend
+        sortiert nach Nummer. Enthält keine Duplikate. Leer bei Fehler.
+
+    Examples
+    --------
+    >>> from qgis.core import QgsGeometry
+    >>> vg_geom = QgsGeometry.fromWkt("POLYGON((414000 5657000, 415000 5657000, ...))")
+    >>> result = gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom)
+    >>> # result: [('1234', 'Beispielgemarkung'), ('5678', 'Andere Gemarkung')]
+    """
+    if vg_geometrie is None or vg_geometrie.isEmpty():
+        _warn("Gemarkung", "[DIAG] VG-Geometrie ist None oder leer – Gemarkungssuche abgebrochen.", duration=8)
+        return []
+
+    try:
+        layer = _load_all_gemarkung_layer()
+    except Exception as exc:
+        _warn("Gemarkung", f"[DIAG] Fehler beim Laden des Gemarkung-Layers: {exc}", duration=8)
+        return []
+
+    if not layer:
+        _warn("Gemarkung", "[DIAG] AX_Gemarkung-Layer ungültig oder nicht ladbar.", duration=8)
+        return []
+
+    # Feature-Anzahl ermitteln (0 = WFS gibt nichts zurück)
+    alle_features = list(layer.getFeatures())
+    if not alle_features:
+        _warn(
+            "Gemarkung",
+            "[DIAG] AX_Gemarkung-Layer ist gültig, aber enthält 0 Features. "
+            "WFS-Dienst liefert ohne Filter möglicherweise kein Ergebnis.",
+            duration=10,
+        )
+        return []
+
+    # Feldnamen der ersten Features loggen
+    feldnamen = [alle_features[0].fields().field(i).name() for i in range(alle_features[0].fields().count())]
+    hat_geom = sum(1 for f in alle_features[:20] if f.hasGeometry() and not f.geometry().isEmpty())
+    _info(
+        "Gemarkung",
+        f"[DIAG] {len(alle_features)} Gemarkungen geladen. "
+        f"Von ersten 20 haben {hat_geom} Geometrie. "
+        f"Alle Felder ({len(feldnamen)}): {feldnamen}",
+        duration=15,
+    )
+
+    result: dict[str, str] = {}
+    kein_schnitt = 0
+    for feat in alle_features:
+        try:
+            gem_geom = feat.geometry()
+            if gem_geom is None or gem_geom.isEmpty():
+                continue
+            if not gem_geom.intersects(vg_geometrie):
+                kein_schnitt += 1
+                continue
+        except Exception as exc:
+            _warn("Gemarkung", f"[DIAG] intersects() Fehler: {exc}", duration=8)
+            continue
+
+        nummer, name = _extract_schluessel_und_name(feat)
+        if nummer and nummer not in result:
+            result[nummer] = name
+
+    if not result:
+        _warn(
+            "Gemarkung",
+            f"[DIAG] 0 Gemarkungen nach Verschneidung. {kein_schnitt} Features ohne Schnitt. "
+            f"VG-Geometrie gültig: {not vg_geometrie.isNull()}, "
+            f"VG-Typ: {vg_geometrie.type()}, VG-WKT (Anfang): {vg_geometrie.asWkt()[:80]}",
+            duration=15,
+        )
+
+    return sorted(result.items(), key=lambda t: t[0])
+
+
+def gemarkung_schluessel_von_vg_geometrie(vg_geometrie) -> List[str]:
+    """Gibt ausschließlich die Gemarkungsnummern für die VG-Geometrie zurück.
+
+    Convenience-Wrapper um
+    :func:`gemarkung_schluessel_und_namen_von_vg_geometrie`.
+
+    Parameters
+    ----------
+    vg_geometrie :
+        ``QgsGeometry`` des Verfahrensgebiets in EPSG:25833.
+
+    Returns
+    -------
+    List[str]
+        Sortierte Liste der Gemarkungsnummern. Leer bei Fehler oder keinem Treffer.
+    """
+    return [nr for nr, _ in gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geometrie)]

functions/grenzpunkt_alkis.py

@@ -0,0 +1,280 @@
+"""sn_verfahrensgebiet/functions/grenzpunkt_alkis.py – ALKIS-Grenzpunktabfrage.
+
+Lädt Grenzpunkte aus dem ALKIS-NAS-WFS in zwei Schritten:
+
+1. ``adv:AX_PunktortTA`` per BBox laden.
+   Diese Objektart enthält die eigentliche Koordinate (Geometrie) sowie das
+   Attribut ``gehoertZu``, das per xlink-Referenz auf die GML-ID des
+   zugehörigen ``AX_Grenzpunkt``-Objekts verweist.
+
+2. GML-IDs aus ``gehoertZu`` extrahieren → ``AX_Grenzpunkt`` per
+   ``gml_id IN (...)`` abfragen (batching, ~25 IDs/Batch).
+
+3. Beide Layer werden über ``gehoertZu ↔ gml_id`` verknüpft (Join) und als
+   einzelner angereicherter Layer zurückgegeben.
+
+**Einschränkungen des WFS-Dienstes:**
+
+- ``AX_Grenzpunkt`` hat **keine eigene Geometrie** → kein BBOX-Filter möglich.
+- Das Attribut ``gehoertZu`` in ``AX_PunktortTA`` ist serverseitig **nicht
+  filterbar** (bestätigt durch WFS-Fehlermeldung ``ErrorInPropertyName``).
+- Der serverseitige GML-ID-Filter (``ResourceId``) ist aktiviert
+  (``ImplementsResourceId=TRUE`` in den Capabilities).
+
+**Strategie:** BBox → PunktortTA → GML-IDs → Grenzpunkt-Batches →
+client-seitiger Join.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import List, Optional, Tuple
+
+from sn_basis.functions import QgsVectorLayer, QgsProject
+from sn_basis.functions import QMessageBox
+from sn_basis.functions.message_wrapper import info, warning
+
+ALKIS_SF_URL = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
+PUNKTORT_TA_TYPENAME = "adv:AX_PunktortTA"
+GRENZPUNKT_TYPENAME = "adv:AX_Grenzpunkt"
+
+PUNKTORT_LAYER_NAME = "PunktortTA"
+GRENZPUNKT_LAYER_NAME = "Grenzpunkte"
+
+# Attributname in AX_PunktortTA, der auf AX_Grenzpunkt verweist
+_ATTR_GEHOERT_ZU = "gehoertZu"
+# Attributname für die GML-ID in AX_Grenzpunkt
+_ATTR_GML_ID = "gml_id"
+
+# Batch-Größe für ResourceId-Abfragen (URL-Längenlimit)
+_BATCH_GROESSE = 25
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _extrahiere_gml_ids(punktort_layer) -> List[str]:
+    """Extrahiert GML-IDs aus dem ``gehoertZu``-Attribut des PunktortTA-Layers.
+
+    Der WFS gibt ``gehoertZu`` üblicherweise als ``xlink:href``-Referenz im
+    Format ``urn:adv:oid:DEXXXXXXXXXXXXXXXXXX`` oder als flachen Wert.
+    Es wird der letzte Bestandteil nach dem letzten ``:`` verwendet.
+
+    Parameters
+    ----------
+    punktort_layer :
+        Geladener ``AX_PunktortTA``-WFS-Layer.
+
+    Returns
+    -------
+    List[str]
+        Deduplizierte, sortierte Liste von GML-IDs.
+    """
+    ids: set[str] = set()
+    for feat in punktort_layer.getFeatures():
+        try:
+            val = feat[_ATTR_GEHOERT_ZU]
+            if val:
+                raw = str(val).strip()
+                # xlink:href hat Format "urn:adv:oid:DE..." oder reiner GML-ID-Wert
+                gml_id = raw.rsplit(":", 1)[-1] if ":" in raw else raw
+                if gml_id:
+                    ids.add(gml_id)
+        except (KeyError, Exception):
+            continue
+    return sorted(ids)
+
+
+def _lade_grenzpunkte_per_ids(id_liste: List[str]) -> Optional[object]:
+    """Lädt ``AX_Grenzpunkt``-Features für die angegebenen GML-IDs.
+
+    Die ID-Liste wird in Batches der Größe :data:`_BATCH_GROESSE` aufgeteilt,
+    um URL-Längenlimits zu umgehen. Alle Batch-Ergebnisse werden zu einem
+    Memory-Layer zusammengeführt.
+
+    Parameters
+    ----------
+    id_liste :
+        GML-IDs der gesuchten ``AX_Grenzpunkt``-Objekte.
+
+    Returns
+    -------
+    QgsVectorLayer | None
+        Memory-Layer mit allen gefundenen Grenzpunkten oder ``None`` bei Fehler.
+    """
+    from sn_basis.functions import QgsFeature, QgsField
+    from sn_basis.functions.qt_wrapper import QVariant
+
+    alle_features: List = []
+    feld_namen: List[str] = []
+
+    anzahl_batches = math.ceil(len(id_liste) / _BATCH_GROESSE)
+    for i in range(anzahl_batches):
+        batch = id_liste[i * _BATCH_GROESSE: (i + 1) * _BATCH_GROESSE]
+        ids_str = ", ".join(f"'{gid}'" for gid in batch)
+        uri = (
+            f"url='{ALKIS_SF_URL}' typename='{GRENZPUNKT_TYPENAME}' "
+            f"srsname='EPSG:25833' "
+            f"sql=SELECT * FROM AX_Grenzpunkt WHERE gml_id IN ({ids_str})"
+        )
+        tmp = QgsVectorLayer(uri, "_TmpGP", "WFS")
+        if not tmp or not tmp.isValid():
+            continue
+
+        if not feld_namen:
+            feld_namen = [f.name() for f in tmp.fields()]
+
+        for feat in tmp.getFeatures():
+            alle_features.append(feat)
+
+    if not alle_features:
+        return None
+
+    # Ergebnis-Memory-Layer aufbauen (ohne Geometrie, da AX_Grenzpunkt keine hat)
+    mem_uri = "Point?crs=EPSG:25833" if _hat_geometrie(alle_features[0]) else "none"
+    result_layer = QgsVectorLayer(mem_uri, GRENZPUNKT_LAYER_NAME, "memory")
+    pr = result_layer.dataProvider()
+
+    felder = [QgsField(n, QVariant.String) for n in feld_namen if n]
+    pr.addAttributes(felder)
+    result_layer.updateFields()
+    pr.addFeatures(alle_features)
+    result_layer.updateExtents()
+
+    return result_layer
+
+
+def _hat_geometrie(feature) -> bool:
+    """Gibt ``True`` zurück, wenn das Feature eine nicht-leere Geometrie hat."""
+    try:
+        g = feature.geometry()
+        return g is not None and not g.isEmpty()
+    except Exception:
+        return False
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def grenzpunkt_alkis(tab_widget, extent=None) -> Tuple[Optional[object], Optional[object]]:
+    """Lädt Grenzpunkte aus dem ALKIS-WFS für den angegebenen Begrenzungsrahmen.
+
+    Ablauf:
+
+    1. ``AX_PunktortTA`` per BBox laden (enthält Koordinaten + Referenz auf
+       ``AX_Grenzpunkt`` via ``gehoertZu``).
+    2. GML-IDs aus ``gehoertZu`` extrahieren.
+    3. ``AX_Grenzpunkt`` per ID-Liste in Batches laden.
+    4. Beide Layer ins Projekt einfügen. Der Caller ist verantwortlich dafür,
+       den Join in der QGIS-Projektkonfiguration einzurichten oder die
+       Verknüpfung client-seitig vorzunehmen.
+
+    Wenn kein ``extent`` übergeben wird, fragt die Funktion nach dem aktuellen
+    Verfahrensgebiet-Layer und nutzt dessen BBox.
+
+    Parameters
+    ----------
+    tab_widget :
+        Eltern-Widget für Dialoge.
+    extent :
+        Optional. ``QgsRectangle`` in EPSG:25833. Wenn ``None``, wird der
+        aktive Verfahrensgebiet-Layer gesucht und dessen Extent genutzt.
+
+    Returns
+    -------
+    Tuple[Optional[punktort_layer], Optional[grenzpunkt_layer]]
+        Beide geladenen Layer (bereits im Projekt) oder ``(None, None)``
+        bei Fehler.
+    """
+    project = QgsProject.instance()
+
+    # BBox bestimmen
+    if extent is None:
+        vg_layers = project.mapLayersByName("Verfahrensgebiet")
+        if not vg_layers:
+            QMessageBox.warning(
+                tab_widget,
+                "Kein Verfahrensgebiet",
+                "Bitte zuerst ein Verfahrensgebiet laden, um die BBox abzuleiten.",
+            )
+            return None, None
+        extent = vg_layers[0].extent()
+
+    xmin = extent.xMinimum()
+    ymin = extent.yMinimum()
+    xmax = extent.xMaximum()
+    ymax = extent.yMaximum()
+
+    # -----------------------------------------------------------------------
+    # Schritt 1: AX_PunktortTA per BBox laden
+    # -----------------------------------------------------------------------
+    info("Grenzpunkte", "Schritt 1/3: PunktortTA per BBox laden …", duration=3)
+    uri_pta = (
+        f"url='{ALKIS_SF_URL}' typename='{PUNKTORT_TA_TYPENAME}' "
+        f"srsname='EPSG:25833' "
+        f"bbox={xmin},{ymin},{xmax},{ymax},EPSG:25833"
+    )
+    punktort_layer = QgsVectorLayer(uri_pta, PUNKTORT_LAYER_NAME, "WFS")
+
+    if not punktort_layer or not punktort_layer.isValid():
+        QMessageBox.critical(
+            tab_widget,
+            "Fehler",
+            "PunktortTA-Layer konnte nicht geladen werden.\n"
+            "Bitte Netzwerkverbindung und Ausdehnung prüfen.",
+        )
+        return None, None
+
+    anzahl_pta = punktort_layer.featureCount()
+    if anzahl_pta == 0:
+        QMessageBox.warning(
+            tab_widget,
+            "Keine Grenzpunkte",
+            "Im Bereich des Verfahrensgebiets wurden keine PunktortTA-Objekte gefunden.",
+        )
+        return None, None
+
+    # -----------------------------------------------------------------------
+    # Schritt 2: GML-IDs extrahieren
+    # -----------------------------------------------------------------------
+    info("Grenzpunkte", "Schritt 2/3: GML-IDs aus gehoertZu extrahieren …", duration=3)
+    gml_ids = _extrahiere_gml_ids(punktort_layer)
+
+    if not gml_ids:
+        QMessageBox.warning(
+            tab_widget,
+            "Keine Verknüpfungen",
+            f"{anzahl_pta} PunktortTA-Objekte gefunden, aber kein 'gehoertZu'-Attribut.\n"
+            "Bitte Attributnamen im WFS prüfen.",
+        )
+        # PunktortTA trotzdem zurückgeben
+        project.addMapLayer(punktort_layer)
+        return punktort_layer, None
+
+    # -----------------------------------------------------------------------
+    # Schritt 3: AX_Grenzpunkt per ID-Listen-Batches laden
+    # -----------------------------------------------------------------------
+    info(
+        "Grenzpunkte",
+        f"Schritt 3/3: {len(gml_ids)} Grenzpunkte in "
+        f"{math.ceil(len(gml_ids) / _BATCH_GROESSE)} Batches laden …",
+        duration=3,
+    )
+    grenzpunkt_layer = _lade_grenzpunkte_per_ids(gml_ids)
+
+    if not grenzpunkt_layer:
+        warning(
+            "Grenzpunkte",
+            "Grenzpunkt-Daten konnten nicht geladen werden (IDs gefunden, aber WFS lieferte kein Ergebnis).",
+            duration=6,
+        )
+        project.addMapLayer(punktort_layer)
+        return punktort_layer, None
+
+    # Beide Layer ins Projekt
+    project.addMapLayer(punktort_layer)
+    project.addMapLayer(grenzpunkt_layer)
+
+    return punktort_layer, grenzpunkt_layer

functions/umringspruefung.py

@@ -0,0 +1,311 @@
+"""sn_verfahrensgebiet/functions/umringspruefung.py – Umringsprüfung (Variante 1).
+
+Vergleicht den Umring (Boundary) des BauRaumOderBodenordnungsrecht-Polygons
+mit den Grenzen der angrenzenden Flurstücke und hebt Abweichungen hervor.
+
+**Algorithmus (Variante 1 – linienbasierter Umringvergleich):**
+
+1. BROR-Polygon → Umringslinie via :func:`sn_basis.functions.qgiscore_wrapper.polygon_zu_linie`
+   (QGIS-3/4-kompatibel).
+2. Flurstücke, die den BROR-Umring berühren oder schneiden, werden gefiltert.
+3. Gefilterte Flurstücke → Umringslinien.
+4. Union aller Flurstücksumringe.
+5. ``BROR-Umring.difference(Flurstücksumrings-Union)`` → Abweichungen (rot).
+6. ``BROR-Umring.difference(rote_Segmente)`` → Übereinstimmungen (grün).
+7. Ergebnislayer mit Kategoriesymbologie (grün/rot) und Legendenbeschriftung.
+
+**Rückgabe:** ``(result_layer, conflict_count)``
+
+- ``result_layer``: Temporärer Linienlayer mit Feld ``typ``
+  (``"identisch"`` = grün, ``"Abweichung"`` = rot).
+- ``conflict_count``: Anzahl der roten Konfliktsegmente (Parts).
+"""
+from __future__ import annotations
+
+from typing import List, Optional, Tuple
+
+from sn_basis.functions.qgiscore_wrapper import (
+    QgsVectorLayer,
+    QgsFeature,
+    QgsField,
+    QgsGeometry,
+    polygon_zu_linie,
+)
+from sn_basis.functions.qt_wrapper import QVariant
+
+# ---------------------------------------------------------------------------
+# Konstanten
+# ---------------------------------------------------------------------------
+
+#: Geometrische Toleranz für den Linienvergleich (EPSG:25833 = Meter).
+#: Der BROR-Umring wird gegen die um diesen Betrag gepufferten Flurstücksumringe
+#: verglichen, um numerische Rundungsfehler aus ALKIS-Koordinaten abzufangen.
+LINIE_TOLERANZ_M: float = 0.0001
+
+ERGEBNIS_LAYER_NAME = "Umringsprüfung"
+FELD_TYP = "typ"
+WERT_IDENTISCH = "identisch"
+WERT_ABWEICHUNG = "Abweichung"
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _zerlege_in_teile(geom: object) -> List[object]:
+    """Zerlegt eine (Multi-)LineString-Geometrie in einzelne Teile.
+
+    Parameters
+    ----------
+    geom :
+        ``QgsGeometry`` vom Typ LineString oder MultiLineString.
+
+    Returns
+    -------
+    List[QgsGeometry]
+        Liste einzelner Geometrieteile; leer wenn ``geom`` None oder leer ist.
+    """
+    if geom is None:
+        return []
+    try:
+        if getattr(geom, "isEmpty", lambda: True)():
+            return []
+        # asGeometryCollection() gibt eigene Kopien zurück – kein Ownership-Problem.
+        # QgsGeometry(part) über geom.parts() ist in QGIS 4.x aus den Python-Bindings
+        # entfernt worden und führt zu Absturz beim Rendern.
+        parts = geom.asGeometryCollection()
+        if parts:
+            return [p for p in parts if p is not None and not p.isEmpty()]
+    except Exception:
+        pass
+    # Fallback: Geometrie nicht zerlegbar → als Ganzes zurückgeben
+    try:
+        if not geom.isEmpty():
+            return [geom]
+    except Exception:
+        pass
+    return []
+
+
+def _erstelle_ergebnis_layer(crs_authid: str) -> object:
+    """Erzeugt den leeren Ergebnislayer mit Feld ``typ``.
+
+    Parameters
+    ----------
+    crs_authid :
+        CRS-Kennung des Quell-Layers (z. B. ``"EPSG:25833"``).
+
+    Returns
+    -------
+    QgsVectorLayer
+        Leerer Memory-Layer vom Typ LineString.
+    """
+    layer = QgsVectorLayer(
+        f"LineString?crs={crs_authid}", ERGEBNIS_LAYER_NAME, "memory"
+    )
+    pr = layer.dataProvider()
+    pr.addAttributes([QgsField(FELD_TYP, QVariant.String)])
+    layer.updateFields()
+    return layer
+
+
+def _setze_kategorien_symbologie(layer) -> None:
+    """Wendet Kategoriesymbologie an: grün = identisch, rot = Abweichung.
+
+    Wird bei fehlendem QGIS-Kontext still ignoriert (Unit-Test-Robustheit).
+    """
+    try:
+        from qgis.core import (  # type: ignore[import]
+            QgsCategorizedSymbolRenderer,
+            QgsRendererCategory,
+            QgsLineSymbol,
+        )
+
+        green_sym = QgsLineSymbol.createSimple(
+            {"color": "0,180,0,255", "line_width": "0.5"}
+        )
+        red_sym = QgsLineSymbol.createSimple(
+            {"color": "220,0,0,255", "line_width": "0.8"}
+        )
+        categories = [
+            QgsRendererCategory(
+                WERT_IDENTISCH,
+                green_sym,
+                "identisch (übereinstimmend)",
+            ),
+            QgsRendererCategory(
+                WERT_ABWEICHUNG,
+                red_sym,
+                "Abweichung (nicht in Flurstücken)",
+            ),
+        ]
+        renderer = QgsCategorizedSymbolRenderer(FELD_TYP, categories)
+        layer.setRenderer(renderer)
+    except Exception:
+        pass  # Kein QGIS-Kontext (Tests) → Standarddarstellung behalten
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def umringspruefung(
+    bror_layer,
+    flst_layer,
+    progress=None,
+    linie_toleranz_m: float = LINIE_TOLERANZ_M,
+) -> Tuple[Optional[object], int]:
+    """Variante 1: Linienbasierter Umringvergleich.
+
+    Ermittelt, welche Teile des BROR-Umrings mit den Flurstücksgrenzen
+    übereinstimmen (grün) und welche keine Entsprechung haben (rot).
+
+    Parameters
+    ----------
+    bror_layer :
+        BauRaumOderBodenordnungsrecht-Memory-Layer (Polygon).
+    flst_layer :
+        Flurstücke-Layer (Polygon, WFS oder Memory).
+    progress :
+        Optionaler ``ProgressDialog`` aus
+        :mod:`sn_basis.functions.dialog_wrapper` für Fortschrittsanzeige
+        und Abbrucherkennung.
+    linie_toleranz_m : float
+        Geometrische Puffertoleranz für den Linienvergleich in
+        CRS-Einheiten (Standardwert :data:`LINIE_TOLERANZ_M` = 0.001 m).
+
+    Returns
+    -------
+    Tuple[Optional[QgsVectorLayer], int]
+        ``(result_layer, conflict_count)``
+
+        - ``result_layer``: Linienlayer mit grünen/roten Segmenten,
+          oder ``None`` bei Abbruch oder fehlgeschlagener Verarbeitung.
+        - ``conflict_count``: Anzahl der roten Konfliktsegmente (BROR-Teile
+          ohne Entsprechung in den Flurstücksgrenzen).
+    """
+
+    def _canceled() -> bool:
+        return progress is not None and progress.is_canceled()
+
+    def _schritt(label: str, wert: int) -> None:
+        if progress is not None:
+            progress.set_label(label)
+            progress.set_value(wert)
+
+    # ------------------------------------------------------------------
+    # 1. BROR-Geometrie und Umringslinie ermitteln
+    # ------------------------------------------------------------------
+    _schritt("Umringsprüfung: BROR-Umring ermitteln …", 0)
+
+    bror_feats = [
+        f for f in bror_layer.getFeatures()
+        if f.geometry() and not f.geometry().isEmpty()
+    ]
+    if not bror_feats:
+        return None, 0
+
+    bror_union = QgsGeometry.unaryUnion([f.geometry() for f in bror_feats])
+    if bror_union is None or bror_union.isEmpty():
+        return None, 0
+
+    bror_boundary = polygon_zu_linie(bror_union)
+    if bror_boundary is None or bror_boundary.isEmpty():
+        return None, 0
+
+    crs_authid = bror_layer.crs().authid()
+
+    if _canceled():
+        return None, 0
+
+    # ------------------------------------------------------------------
+    # 2. Relevante Flurstücke: nur jene, deren Geometrie den BROR-Umring
+    #    berührt oder schneidet
+    # ------------------------------------------------------------------
+    _schritt("Umringsprüfung: Relevante Flurstücke filtern …", 1)
+
+    relevante_boundaries: List[object] = []
+    for feat in flst_layer.getFeatures():
+        if _canceled():
+            return None, 0
+        geom = feat.geometry()
+        if geom is None or geom.isEmpty():
+            continue
+        try:
+            if geom.intersects(bror_boundary):
+                boundary = polygon_zu_linie(geom)
+                if boundary is not None and not boundary.isEmpty():
+                    relevante_boundaries.append(boundary)
+        except Exception:
+            continue
+
+    if relevante_boundaries:
+        flst_boundary_union = QgsGeometry.unaryUnion(relevante_boundaries)
+    else:
+        flst_boundary_union = None
+
+    if _canceled():
+        return None, 0
+
+    # ------------------------------------------------------------------
+    # 3. Segmentvergleich: Differenz und Übereinstimmung berechnen
+    # ------------------------------------------------------------------
+    _schritt("Umringsprüfung: Segmentvergleich …", 2)
+
+    ergebnis_layer = _erstelle_ergebnis_layer(crs_authid)
+    pr = ergebnis_layer.dataProvider()
+    feats_to_add: List[object] = []
+    conflict_count = 0
+
+    if flst_boundary_union is None or flst_boundary_union.isEmpty():
+        # Kein Flurstück berührt den BROR-Umring → alles ist Abweichung
+        for part in _zerlege_in_teile(bror_boundary):
+            feat = QgsFeature()
+            feat.setGeometry(part)
+            feat.setAttributes([WERT_ABWEICHUNG])
+            feats_to_add.append(feat)
+            conflict_count += 1
+    else:
+        # Flurstücksumringe leicht puffern für Toleranz-Matching
+        if linie_toleranz_m > 0:
+            flst_fuer_diff = flst_boundary_union.buffer(linie_toleranz_m, 4)
+        else:
+            flst_fuer_diff = flst_boundary_union
+
+        # Rote Segmente: BROR-Teile ohne Entsprechung in Flurstücksgrenzen
+        try:
+            red_geom = bror_boundary.difference(flst_fuer_diff)
+        except Exception:
+            red_geom = None
+
+        # Grüne Segmente: BROR-Teile, die nicht rot sind
+        try:
+            if red_geom and not red_geom.isEmpty():
+                green_geom = bror_boundary.difference(red_geom)
+            else:
+                green_geom = bror_boundary
+        except Exception:
+            green_geom = None
+
+        for part in _zerlege_in_teile(green_geom):
+            feat = QgsFeature()
+            feat.setGeometry(part)
+            feat.setAttributes([WERT_IDENTISCH])
+            feats_to_add.append(feat)
+
+        for part in _zerlege_in_teile(red_geom):
+            feat = QgsFeature()
+            feat.setGeometry(part)
+            feat.setAttributes([WERT_ABWEICHUNG])
+            feats_to_add.append(feat)
+            conflict_count += 1
+
+    if _canceled():
+        return None, 0
+
+    pr.addFeatures(feats_to_add)
+    ergebnis_layer.updateExtents()
+    _setze_kategorien_symbologie(ergebnis_layer)
+
+    _schritt("Umringsprüfung abgeschlossen.", 3)
+    return ergebnis_layer, conflict_count

functions/verfahrensgebiet_alkis.py

@@ -10,7 +10,7 @@ from sn_basis.functions import QMessageBox
 from sn_basis.functions.qt_wrapper import QVariant
 from sn_basis.functions.variable_wrapper import get_variable
 
-alkis_NAS_url = "https://geodienste.sachsen.de/aaa/public_alkis/nas/wfs"
+alkis_sf_url = "https://geodienste.sachsen.de/aaa/public_alkis/sf/wfs"
 typename = "adv:AX_BauRaumOderBodenordnungsrecht"
 
 class LoadStatus(Enum):
@@ -60,27 +60,32 @@ def _check_existing_layer(tab_widget):
 
 def verfahrensgebiet_alkis(tab_widget):
     """Lädt Verfahrensgebiet aus ALKIS-WFS basierend auf Verfahrensnummer.
-    
+
     Args:
         tab_widget: Parent-Widget für Dialoge und Meldungen
-        
+
     Returns:
-        Tuple[Optional[QgsVectorLayer], LoadStatus]: Geladener Layer und Status
+        Tuple[Optional[QgsVectorLayer], LoadStatus, List[QgsFeature]]:
+            Geladener Memory-Layer, Lade-Status und Liste der Roh-WFS-Features
+            (für geometriebasierte Weiterverarbeitung, z.B. Gemarkungsschlüssel-
+            Ermittlung via Spatial-Join). Bei Abbruch oder Fehler: ``(None,
+            LoadStatus.NONE, [])``. Bei ``LoadStatus.KEEP`` werden keine
+            WFS-Features neu geladen; die Feature-Liste ist in diesem Fall leer.
     """
     verfahrensnummer = get_variable("verfahrensnummer")
     if not verfahrensnummer:
         QMessageBox.critical(tab_widget, "Fehler",
                              "In den Projekteigenschaften ist noch keine Verfahrensnummer eingetragen!")
-        return None, LoadStatus.NONE
+        return None, LoadStatus.NONE, []
 
     # Vorhandenen Layer prüfen
     existing, status = _check_existing_layer(tab_widget)
     if status == LoadStatus.KEEP:
-        return existing, status
+        return existing, status, []
 
     # WFS mit Filter laden
     wfs_uri_find = (
-        f"url='{alkis_NAS_url}' typename='{typename}' srsname='EPSG:25833' "
+        f"url='{alkis_sf_url}' typename='{typename}' srsname='EPSG:25833' "
         f"sql=SELECT * FROM AX_BauRaumOderBodenordnungsrecht "
         f"WHERE bezeichnung LIKE '{verfahrensnummer}%'"
     )
@@ -88,18 +93,18 @@ def verfahrensgebiet_alkis(tab_widget):
 
     if not temp_layer.isValid():
         QMessageBox.critical(tab_widget, "Fehler", "Layer konnte nicht geladen werden.")
-        return None, LoadStatus.NONE
+        return None, LoadStatus.NONE, []
 
     features = list(temp_layer.getFeatures())
     if not features:
         QMessageBox.critical(tab_widget, "Fehler",
                              f"Verfahrenskennzeichen {verfahrensnummer} nicht im WFS gefunden.")
-        return None, LoadStatus.NONE
+        return None, LoadStatus.NONE, []
 
     union_geom = QgsGeometry.unaryUnion([f.geometry() for f in features])
     if union_geom is None or union_geom.isEmpty():
         QMessageBox.critical(tab_widget, "Fehler", "Keine Geometrien zum Verschmelzen gefunden.")
-        return None, LoadStatus.NONE
+        return None, LoadStatus.NONE, []
 
     # Memory-Layer erzeugen
     crs = temp_layer.crs().authid()
@@ -114,5 +119,5 @@ def verfahrensgebiet_alkis(tab_widget):
     pr.addFeatures([feat_union])
     memory_layer.updateExtents()
 
-    # Status: FIRST oder RELOAD
-    return memory_layer, status
+    # Status: FIRST oder RELOAD; features für Gemarkungsermittlung weitergeben
+    return memory_layer, status, features

functions/verfahrensgebiet_alkis_komplett.py

@@ -0,0 +1,480 @@
+"""sn_verfahrensgebiet/functions/verfahrensgebiet_alkis_komplett.py – Vollständiger VG-Lade-Workflow.
+
+Orchestriert den vollständigen Arbeitsablauf zum Laden eines Verfahrensgebiets
+aus ALKIS mit anschließender Flurstücks-Verschneidung, Flächenvergleich und
+optionaler Umringsprüfung:
+
+1. Verfahrensgebiet aus ``AX_BauRaumOderBodenordnungsrecht`` per VKZ laden.
+2. Gemarkungsschlüssel per Geometrie aus ``AX_Gemarkung`` vorschlagen.
+3. Flurstücke per Dialog (mit Vorschlag) laden.
+4. Flurstücke räumlich auf das VG einschränken (``selectByLocation``).
+5. Aus der Auswahl den enriched Verfahrensgebiet-Memory-Layer erzeugen.
+6. Flächenvergleich zwischen BROR-Objekt und abgeleitetem Verfahrensgebiet.
+
+   - **Übereinstimmung** (≤ Toleranz): modale OK-Meldung, Ablauf beendet.
+   - **Abweichung**: Ja/Nein-Abfrage für Detailprüfung.
+
+7. Optionale Umringsprüfung (Variante 1):
+
+   - BROR-Umring gegen Flurstücksgrenzen vergleichen.
+   - Ergebnislayer (grün/rot) ins Projekt einfügen.
+   - Abschlussmeldung mit Konfliktanzahl.
+
+Das BROR-Objekt (``AX_BauRaumOderBodenordnungsrecht``) bleibt bei Erfolg als
+temporärer Layer ``BauRaumOrdnungsrecht (ALKIS)`` im Projekt erhalten.
+
+**Rückgabe:**
+``Tuple[enriched_layer, flst_layer, bror_layer, LoadStatus]``
+
+**Abbruchverhalten:** Jeder Schritt, der :attr:`LoadStatus.NONE` zurückgibt
+oder kein Ergebnis liefert, bricht den Workflow ab. Der BROR-Layer wird nur
+bei Fehler/Abbruch entfernt (wenn er in diesem Lauf hinzugefügt wurde).
+"""
+
+from __future__ import annotations
+
+from typing import Optional, Tuple
+
+from sn_basis.functions import QgsProject
+from sn_basis.functions import QMessageBox
+from sn_basis.functions.message_wrapper import warning
+from sn_basis.functions.variable_wrapper import get_variable
+from sn_basis.functions.dialog_wrapper import (
+    create_progress_dialog,
+    show_info_dialog,
+    ask_yes_no,
+)
+
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import (
+    verfahrensgebiet_alkis,
+    LoadStatus,
+)
+from sn_verfahrensgebiet.functions.gemarkung_alkis import (
+    gemarkung_schluessel_und_namen_von_vg_geometrie,
+)
+from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis
+from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import (
+    erstelle_verfahrensgebiet_aus_auswahl,
+    VERF_LAYER_NAME,
+)
+from sn_verfahrensgebiet.functions.umringspruefung import umringspruefung
+
+# Feldname für das Verfahrenskennzeichen im enriched Layer
+FELD_VKZ = "verfahrensnummer"
+#: Anzeigename des BROR-Layers im Projekt
+BROR_LAYER_NAME = "BauRaumOrdnungsrecht (ALKIS)"
+#: Projektvariable für die Flächentoleranz (in m²); Default 0
+FLAECHEN_TOLERANZ_VARIABLE = "flaechen_toleranz_m2"
+
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _entferne_hilfs_layer(project, layer_name: str) -> None:
+    """Entfernt alle Layer mit dem angegebenen Namen aus dem Projekt.
+
+    Wird verwendet, um temporäre WFS-Hilfs-Layer nach der Verschneidung
+    aus dem Projekt zu bereinigen.
+
+    Parameters
+    ----------
+    project :
+        Aktuelles QGIS-Projekt.
+    layer_name :
+        Anzeigename des zu entfernenden Layers.
+    """
+    for lyr in project.mapLayersByName(layer_name):
+        project.removeMapLayer(lyr.id())
+
+
+def _select_by_location(flst_layer, vg_layer) -> int:
+    """Selektiert Flurstücke, die innerhalb des Verfahrensgebiets liegen.
+
+    Verwendet ``processing.run("native:selectbylocation")`` mit Prädikat 6
+    (within). Flurstücke, die das VG nur berühren, werden ausgeschlossen.
+
+    Parameters
+    ----------
+    flst_layer :
+        Flurstücks-Layer (Quelle der Selektion).
+    vg_layer :
+        Verfahrensgebiet-Layer (Schnittmaske).
+
+    Returns
+    -------
+    int
+        Anzahl der selektierten Features; 0 bei Fehler oder leerem Ergebnis.
+    """
+    try:
+        import processing  # type: ignore[import]
+
+        processing.run(
+            "native:selectbylocation",
+            {
+                "INPUT": flst_layer,
+                "PREDICATE": [6],  # 6 = within; schließt reines Berühren aus
+                "INTERSECT": vg_layer,
+                "METHOD": 0,  # 0 = neue Selektion
+            },
+        )
+        return flst_layer.selectedFeatureCount()
+
+    except Exception:
+        return 0
+
+
+def _fuege_vkz_feld_hinzu(layer, vkz: str) -> None:
+    """Fügt das Feld ``verfahrensnummer`` zum enriched Layer hinzu und befüllt
+    es mit dem Verfahrenskennzeichen, sofern dieses nicht leer ist.
+
+    Parameters
+    ----------
+    layer :
+        Enriched Verfahrensgebiet-Memory-Layer (in-place geändert).
+    vkz :
+        Verfahrenskennzeichen aus der Projektvariable ``verfahrensnummer``.
+    """
+    if not vkz:
+        return
+
+    try:
+        from sn_basis.functions import QgsField
+        from sn_basis.functions.qt_wrapper import QVariant
+
+        pr = layer.dataProvider()
+        pr.addAttributes([QgsField(FELD_VKZ, QVariant.String)])
+        layer.updateFields()
+
+        idx = layer.fields().indexFromName(FELD_VKZ)
+        if idx < 0:
+            return
+
+        changes = {}
+        for feat in layer.getFeatures():
+            changes[feat.id()] = {idx: vkz}
+        pr.changeAttributeValues(changes)
+        layer.updateFields()
+
+    except Exception:
+        pass
+
+
+def _lese_flaechen_toleranz() -> float:
+    """Liest die Flächentoleranz aus der Projektvariable (Default: 0.0 m²).
+
+    Die Projektvariable :data:`FLAECHEN_TOLERANZ_VARIABLE` kann in den
+    Projekteigenschaften gesetzt werden. Fehlt sie oder ist sie ungültig,
+    wird 0.0 zurückgegeben (exakter Vergleich).
+
+    Returns
+    -------
+    float
+        Toleranz in m² (bei EPSG:25833 identisch mit den Flächen-Einheiten).
+    """
+    try:
+        raw = get_variable(FLAECHEN_TOLERANZ_VARIABLE)
+        if raw is not None and str(raw).strip():
+            return float(str(raw).strip())
+    except Exception:
+        pass
+    return 0.0
+
+
+def _flaeche_des_layers(layer) -> float:
+    """Summiert die Flächen aller Features eines Layers (in CRS-Einheiten²).
+
+    Parameters
+    ----------
+    layer :
+        Beliebiger Polygon-Layer.
+
+    Returns
+    -------
+    float
+        Summe aller Feature-Flächen; 0.0 wenn keine Features vorhanden.
+    """
+    total = 0.0
+    for feat in layer.getFeatures():
+        try:
+            geom = feat.geometry()
+            if geom and not geom.isEmpty():
+                total += geom.area()
+        except Exception:
+            pass
+    return total
+
+
+def _vergleiche_flaechen(
+    bror_layer, vg_layer, toleranz_m2: float = 0.0
+) -> Tuple[float, float, bool]:
+    """Vergleicht Flächen von BROR- und Verfahrensgebiet-Layer.
+
+    Parameters
+    ----------
+    bror_layer :
+        BauRaumOrdnungsrecht-Layer.
+    vg_layer :
+        Enriched Verfahrensgebiet-Layer.
+    toleranz_m2 :
+        Maximale zulässige Flächendifferenz in m².
+
+    Returns
+    -------
+    Tuple[float, float, bool]
+        ``(bror_flaeche, vg_flaeche, stimmt_ueberein)``
+    """
+    bror_flaeche = _flaeche_des_layers(bror_layer)
+    vg_flaeche = _flaeche_des_layers(vg_layer)
+    stimmt_ueberein = abs(bror_flaeche - vg_flaeche) <= toleranz_m2
+    return bror_flaeche, vg_flaeche, stimmt_ueberein
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def verfahrensgebiet_alkis_komplett(
+    tab_widget,
+) -> Tuple[Optional[object], Optional[object], Optional[object], LoadStatus]:
+    """Vollständiger VG-Lade-Workflow: ALKIS-VG → Gemarkungen →
+    Flurstücke → Verschneidung → Flächenvergleich → opt. Umringsprüfung.
+
+    Schrittfolge:
+
+    1. ``verfahrensgebiet_alkis(tab_widget)`` – BROR-Objekt als Memory-Layer.
+    2. ``gemarkung_schluessel_und_namen_von_vg_geometrie(geom)`` – Schlüssel.
+    3. ``flurstueck_alkis(...)`` – Flurstücke laden.
+    4. ``_select_by_location(...)`` – räumliche Selektion.
+    5. ``erstelle_verfahrensgebiet_aus_auswahl(...)`` – enriched Layer.
+    6. ``_vergleiche_flaechen(...)`` – Flächenvergleich als Gate.
+    7. Ggf. ``umringspruefung(...)`` nach Nutzerbestätigung.
+
+    Der BROR-Layer (``BauRaumOrdnungsrecht (ALKIS)``) bleibt bei Erfolg
+    im Projekt. Bei Abbruch/Fehler wird er entfernt, falls er in diesem
+    Lauf hinzugefügt wurde.
+
+    Parameters
+    ----------
+    tab_widget :
+        Eltern-Widget für alle Dialoge.
+
+    Returns
+    -------
+    Tuple[Optional[enriched_layer], Optional[flst_layer], Optional[bror_layer], LoadStatus]
+        - ``enriched_layer``: Memory-Layer „Verfahrensgebiet".
+        - ``flst_layer``: Flurstücke-Layer (bleibt im Projekt).
+        - ``bror_layer``: BROR-Layer ``BauRaumOrdnungsrecht (ALKIS)``
+          (bleibt bei Erfolg im Projekt).
+        - :class:`LoadStatus`: ``FIRST``, ``RELOAD`` oder ``NONE``.
+    """
+    project = QgsProject.instance()
+    alkis_vg_layer = None
+    bror_was_added_by_us = False
+    abgeschlossen = False
+
+    # Fortschrittsdialog für den Basisworkflow (7 Schritte)
+    progress = create_progress_dialog(
+        7, "Verfahrensgebiet laden", "Initialisierung …"
+    )
+
+    try:
+        # ------------------------------------------------------------------
+        # Schritt 1: VG aus ALKIS laden (BROR-Objekt → Memory-Layer)
+        # ------------------------------------------------------------------
+        progress.set_label("Schritt 1/6: Verfahrensgebiet aus ALKIS laden …")
+        progress.set_value(0)
+        if progress.is_canceled():
+            return None, None, None, LoadStatus.NONE
+
+        alkis_vg_layer, vg_status, _ = verfahrensgebiet_alkis(tab_widget)
+
+        if vg_status == LoadStatus.NONE or not alkis_vg_layer:
+            return None, None, None, LoadStatus.NONE
+
+        # BROR-Layer umbenennen, um Namenskonflikt mit enriched VG-Layer zu vermeiden
+        alkis_vg_layer.setName(BROR_LAYER_NAME)
+
+        if vg_status in (LoadStatus.FIRST, LoadStatus.RELOAD):
+            project.addMapLayer(alkis_vg_layer)
+            bror_was_added_by_us = True
+
+        progress.set_value(1)
+
+        # ------------------------------------------------------------------
+        # Schritt 2: Gemarkungsschlüssel per Geometrie-Spatial-Join ermitteln
+        # ------------------------------------------------------------------
+        progress.set_label("Schritt 2/6: Gemarkungsschlüssel ermitteln …")
+        if progress.is_canceled():
+            return None, None, None, LoadStatus.NONE
+
+        bror_feats = list(alkis_vg_layer.getFeatures())
+        vg_geom = bror_feats[0].geometry() if bror_feats else None
+
+        schluessel_namen = (
+            gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom)
+            if vg_geom else []
+        )
+        vorgabe_schluessel = [nr for nr, _ in schluessel_namen]
+        vorgabe_namen = [nm for _, nm in schluessel_namen if nm]
+
+        if not vorgabe_schluessel:
+            warning(
+                "VG laden",
+                "Keine Gemarkungen in der BBox gefunden. Bitte Schlüssel manuell eingeben.",
+                duration=5,
+            )
+
+        progress.set_value(2)
+
+        # ------------------------------------------------------------------
+        # Schritt 3: Flurstücke laden (mit Nutzer-Dialog, Vorschlag befüllt)
+        # ------------------------------------------------------------------
+        progress.set_label("Schritt 3/6: Flurstücke aus ALKIS laden …")
+        if progress.is_canceled():
+            return None, None, None, LoadStatus.NONE
+
+        flst_layer, flst_status = flurstueck_alkis(
+            tab_widget,
+            vorgabe_schluessel=vorgabe_schluessel,
+            vorgabe_namen=vorgabe_namen if vorgabe_namen else None,
+            ohne_dialog=True,
+        )
+
+        if flst_status == LoadStatus.NONE or not flst_layer:
+            return None, None, None, LoadStatus.NONE
+
+        progress.set_value(3)
+
+        # ------------------------------------------------------------------
+        # Schritt 4: Räumliche Selektion
+        # ------------------------------------------------------------------
+        progress.set_label("Schritt 4/6: Flurstücke räumlich selektieren …")
+        if progress.is_canceled():
+            return None, None, None, LoadStatus.NONE
+
+        anzahl = _select_by_location(flst_layer, alkis_vg_layer)
+
+        if anzahl == 0:
+            QMessageBox.warning(
+                tab_widget,
+                "Keine Flurstücke im VG",
+                "Es wurden keine Flurstücke gefunden, die das Verfahrensgebiet schneiden.\n"
+                "Bitte Gemarkungsschlüssel und Ausdehnung prüfen.",
+            )
+            return None, flst_layer, None, LoadStatus.NONE
+
+        progress.set_value(4)
+
+        # ------------------------------------------------------------------
+        # Schritt 5: Enriched Memory-Layer aus Auswahl erzeugen
+        # ------------------------------------------------------------------
+        progress.set_label("Schritt 5/6: Verfahrensgebiet-Layer erzeugen …")
+        if progress.is_canceled():
+            return None, None, None, LoadStatus.NONE
+
+        enriched_layer, erfolgreich, fehlgeschlagen = erstelle_verfahrensgebiet_aus_auswahl(
+            flst_layer
+        )
+
+        if not enriched_layer or not enriched_layer.isValid():
+            QMessageBox.critical(
+                tab_widget,
+                "Fehler",
+                "Verfahrensgebiet-Layer konnte nicht erstellt werden.",
+            )
+            return None, flst_layer, None, LoadStatus.NONE
+
+        vkz = get_variable("verfahrensnummer") or ""
+        _fuege_vkz_feld_hinzu(enriched_layer, vkz)
+
+        progress.set_value(5)
+
+        # ------------------------------------------------------------------
+        # Schritt 6: Flächenvergleich als Gate
+        # ------------------------------------------------------------------
+        progress.set_label("Schritt 6/6: Flächenvergleich …")
+        if progress.is_canceled():
+            return None, None, None, LoadStatus.NONE
+
+        toleranz = _lese_flaechen_toleranz()
+        bror_flaeche, vg_flaeche, stimmt_ueberein = _vergleiche_flaechen(
+            alkis_vg_layer, enriched_layer, toleranz
+        )
+
+        progress.set_value(6)
+
+        if stimmt_ueberein:
+            progress.set_value(7)
+            progress.close()
+            show_info_dialog(
+                "Flächenvergleich",
+                "Verfahrensgebiet mit erwarteter Geometrie erzeugt",
+                parent=tab_widget,
+            )
+            abgeschlossen = True
+            return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+        # ------------------------------------------------------------------
+        # Flächenabweichung → Nutzerabfrage für Detailprüfung
+        # ------------------------------------------------------------------
+        progress.close()
+        diff = abs(bror_flaeche - vg_flaeche)
+        starten = ask_yes_no(
+            "Abweichung erkannt",
+            f"Abweichung zwischen BauRaumOrdnungsrecht-Objekt und abgeleitetem "
+            f"Verfahrensgebiet erkannt. Detailprüfung starten?\n\n"
+            f"BauRaumOrdnungsrecht: {bror_flaeche:.2f} m²\n"
+            f"Verfahrensgebiet:     {vg_flaeche:.2f} m²\n"
+            f"Differenz:            {diff:.2f} m²",
+            default=True,
+            parent=tab_widget,
+        )
+
+        if not starten:
+            abgeschlossen = True
+            return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+        # ------------------------------------------------------------------
+        # Umringsprüfung (Variante 1: linienbasierter Umringvergleich)
+        # ------------------------------------------------------------------
+        umring_progress = create_progress_dialog(
+            3, "Detailprüfung (Umringe)", "Starte Umringsprüfung …"
+        )
+
+        ergebnis_layer, konflikt_anzahl = umringspruefung(
+            alkis_vg_layer, flst_layer, progress=umring_progress
+        )
+
+        if umring_progress.is_canceled():
+            umring_progress.close()
+            abgeschlossen = True
+            return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+        umring_progress.set_value(3)
+        umring_progress.close()
+
+        if ergebnis_layer and ergebnis_layer.isValid():
+            project.addMapLayer(ergebnis_layer)
+
+        show_info_dialog(
+            "Detailprüfung",
+            f"Detailprüfung abgeschlossen. "
+            f"Es wurden {konflikt_anzahl} Konfliktstellen identifiziert.",
+            parent=tab_widget,
+        )
+
+        abgeschlossen = True
+        return enriched_layer, flst_layer, alkis_vg_layer, vg_status
+
+    finally:
+        # WFS-Temp-Layer stets aus dem Projekt entfernen
+        _entferne_hilfs_layer(project, "TempVerfahrensgebiet")
+        # BROR-Layer bei Fehler/Abbruch entfernen (nur wenn in diesem Lauf hinzugefügt)
+        if not abgeschlossen and bror_was_added_by_us and alkis_vg_layer is not None:
+            try:
+                project.removeMapLayer(alkis_vg_layer.id())
+            except Exception:
+                pass
+

metadata.txt

@@ -1,12 +1,12 @@
 [general]
-version=26.4.2-testing
+version=26.4.5-unstable
 name=LNO Sachsen | Verfahrensgebiet
 description=Plugin zum Erzeugen eines Objektes "Verfahrensgebiet"
 author=Daniel Helbig, Michael Otto
 homepage=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet
 tracker=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet/issues
-repository=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet/src/branch/testing/
-
+repository=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Verfahrensgebiet/src/branch/unstable/
+QGISMaximumVersion=4.99
 qgisMinimumVersion=3.40
 experimental=true
 deprecated=false

tests/test_flurstueck_alkis.py

@@ -219,5 +219,94 @@ class TestFlurstueckAlkis(unittest.TestCase):
         self.assertEqual(project.removed, [])  # nichts gelöscht
 
 
+    # ------------------------------------------------------------------ #
+    # 6. vorgabe_schluessel füllt ask_text-Default vor                    #
+    # ------------------------------------------------------------------ #
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text")
+    def test_vorgabe_schluessel_prefills_default_text(
+        self, mock_ask_text, mock_project, mock_wfs_layer_cls
+    ):
+        """vorgabe_schluessel=['1234','5678'] → ask_text erhält '1234, 5678' als default_text."""
+        mock_ask_text.return_value = ("1234, 5678", True)
+        mock_project.return_value = _DummyProject()
+        mock_wfs_layer_cls.return_value = _DummyLayer(valid=True)
+
+        flurstueck_alkis(None, vorgabe_schluessel=["1234", "5678"])
+
+        args, kwargs = mock_ask_text.call_args
+        # ask_text("title", label, default_text=..., parent=...)
+        # default_text wird als Keyword-Arg übergeben
+        default_text_arg = kwargs.get("default_text", args[2] if len(args) > 2 else "")
+        self.assertIn("1234", default_text_arg)
+        self.assertIn("5678", default_text_arg)
+
+    # ------------------------------------------------------------------ #
+    # 7. vorgabe_namen erscheint im Label-Text                            #
+    # ------------------------------------------------------------------ #
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text")
+    def test_vorgabe_namen_appears_in_label(
+        self, mock_ask_text, mock_project, mock_wfs_layer_cls
+    ):
+        """vorgabe_namen=['Musterstadt'] → ask_text-label enthält den Namen."""
+        mock_ask_text.return_value = ("9999", True)
+        mock_project.return_value = _DummyProject()
+        mock_wfs_layer_cls.return_value = _DummyLayer(valid=True)
+
+        flurstueck_alkis(None, vorgabe_namen=["Musterstadt"])
+
+        args, kwargs = mock_ask_text.call_args
+        # label ist das 2. Positional-Argument
+        label_arg = args[1] if len(args) > 1 else kwargs.get("label", "")
+        self.assertIn("Musterstadt", label_arg)
+
+    # ------------------------------------------------------------------ #
+    # 8. bbox wird in WFS-URI eingebettet                                 #
+    # ------------------------------------------------------------------ #
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text")
+    def test_bbox_embedded_in_wfs_uri(
+        self, mock_ask_text, mock_project, mock_wfs_layer_cls
+    ):
+        """bbox-Parameter → WFS-URI enthält 'bbox=' Substring."""
+        mock_ask_text.return_value = ("1234", True)
+        mock_project.return_value = _DummyProject()
+        mock_wfs_layer_cls.return_value = _DummyLayer(valid=True)
+
+        class _Rect:
+            def xMinimum(self): return 414000.0
+            def yMinimum(self): return 5657000.0
+            def xMaximum(self): return 415000.0
+            def yMaximum(self): return 5658000.0
+
+        flurstueck_alkis(None, bbox=_Rect())
+
+        uri_str = mock_wfs_layer_cls.call_args[0][0]
+        self.assertIn("bbox=", uri_str.lower())
+
+    # ------------------------------------------------------------------ #
+    # 9. Rückwärtskompatibilität: alle neuen Params optional              #
+    # ------------------------------------------------------------------ #
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsVectorLayer")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.QgsProject.instance")
+    @patch("sn_verfahrensgebiet.functions.flurstueck_alkis.ask_text")
+    def test_backwards_compatible_no_new_params(
+        self, mock_ask_text, mock_project, mock_wfs_layer_cls
+    ):
+        """flurstueck_alkis(tab) ohne neue Parameter → kein Fehler."""
+        mock_ask_text.return_value = ("0000", True)
+        mock_project.return_value = _DummyProject()
+        mock_wfs_layer_cls.return_value = _DummyLayer(valid=True)
+
+        layer, status = flurstueck_alkis(None)
+
+        self.assertIsNotNone(layer)
+        self.assertEqual(status, LoadStatus.FIRST)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_gemarkung_alkis.py

@@ -0,0 +1,207 @@
+"""sn_verfahrensgebiet/tests/test_gemarkung_alkis.py
+
+Unit-Tests für :mod:`sn_verfahrensgebiet.functions.gemarkung_alkis`.
+
+Getestete Szenarien:
+
+Erfolgsfall:
+- WFS-Layer gültig + Features mit Schlüssel → Liste mit Tupeln
+- Schlüssel werden dedupliziert und sortiert zurückgegeben
+- ``gemarkung_schluessel_by_bbox`` gibt nur Nummern (keine Tupel) zurück
+
+Fehlerfälle:
+- WFS-Layer nicht gültig (beide Varianten) → Fallback-Ladeversuch
+- Fallback-Layer ebenfalls ungültig → leere Liste
+- Alle Layer leer (featureCount == 0) → client-seitige Filterung schlägt leer
+- Feature ohne Schlüssel-Attribut wird übersprungen
+- Exception beim Layer-Laden → leere Liste
+"""
+
+import sys
+import types
+import unittest
+from unittest.mock import patch, MagicMock
+
+# ---------------------------------------------------------------------------
+# Minimales QGIS-Mock-Setup
+# ---------------------------------------------------------------------------
+if "qgis" not in sys.modules:
+    qgis_mod = types.ModuleType("qgis")
+    qgis_core = types.ModuleType("qgis.core")
+    qgis_utils = types.ModuleType("qgis.utils")
+    qgis_utils.iface = types.SimpleNamespace()
+    qgis_utils.plugins = {}
+    qgis_mod.utils = qgis_utils
+    qgis_mod.core = qgis_core
+    sys.modules["qgis"] = qgis_mod
+    sys.modules["qgis.core"] = qgis_core
+    sys.modules["qgis.utils"] = qgis_utils
+
+from sn_verfahrensgebiet.functions.gemarkung_alkis import (
+    gemarkung_schluessel_und_namen_by_bbox,
+    gemarkung_schluessel_by_bbox,
+    _extract_schluessel_und_name,
+)
+
+
+# ---------------------------------------------------------------------------
+# Dummies
+# ---------------------------------------------------------------------------
+
+class _DummyRect:
+    def xMinimum(self): return 414000.0
+    def yMinimum(self): return 5657000.0
+    def xMaximum(self): return 415000.0
+    def yMaximum(self): return 5658000.0
+
+
+class _DummyFeature:
+    def __init__(self, attrs: dict):
+        self._attrs = attrs
+
+    def __getitem__(self, key):
+        return self._attrs.get(key, None)
+
+
+def _make_layer(valid=True, feature_count=2, features=None):
+    layer = MagicMock()
+    layer.isValid.return_value = valid
+    layer.featureCount.return_value = feature_count if valid else 0
+    layer.getFeatures.return_value = features or []
+    return layer
+
+
+# ---------------------------------------------------------------------------
+# Tests: _extract_schluessel_und_name
+# ---------------------------------------------------------------------------
+
+class TestExtractSchluesselUndName(unittest.TestCase):
+
+    def test_standard_attribute(self):
+        feat = _DummyFeature({"schluessel_gemarkungsnummer": "1234", "bezeichnung": "Mustergemarkung"})
+        nr, name = _extract_schluessel_und_name(feat)
+        self.assertEqual(nr, "1234")
+        self.assertEqual(name, "Mustergemarkung")
+
+    def test_fallback_attribute(self):
+        feat = _DummyFeature({"gemarkungsnummer": "5678", "gemarkung_bezeichnung": "Alternativ"})
+        nr, name = _extract_schluessel_und_name(feat)
+        self.assertEqual(nr, "5678")
+        self.assertEqual(name, "Alternativ")
+
+    def test_missing_attribute_returns_empty(self):
+        feat = _DummyFeature({})
+        nr, name = _extract_schluessel_und_name(feat)
+        self.assertEqual(nr, "")
+        self.assertEqual(name, "")
+
+    def test_none_value_returns_empty(self):
+        feat = _DummyFeature({"schluessel_gemarkungsnummer": None, "bezeichnung": None})
+        nr, name = _extract_schluessel_und_name(feat)
+        self.assertEqual(nr, "")
+        self.assertEqual(name, "")
+
+
+# ---------------------------------------------------------------------------
+# Tests: gemarkung_schluessel_und_namen_by_bbox
+# ---------------------------------------------------------------------------
+
+class TestGemarkungSchluesselbbox(unittest.TestCase):
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox")
+    def test_success_bbox_filter(self, mock_load_bbox):
+        """WFS-BBox-Filter liefert Features → korrekte Tupel-Liste."""
+        feat1 = _DummyFeature({"schluessel_gemarkungsnummer": "1234", "bezeichnung": "Alpha"})
+        feat2 = _DummyFeature({"schluessel_gemarkungsnummer": "5678", "bezeichnung": "Beta"})
+        mock_load_bbox.return_value = _make_layer(valid=True, features=[feat1, feat2])
+
+        result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect())
+
+        self.assertEqual(result, [("1234", "Alpha"), ("5678", "Beta")])
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_all_gemarkung_layer")
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox")
+    def test_fallback_when_bbox_fails(self, mock_load_bbox, mock_load_all):
+        """BBox-Filter schlägt fehl → Fallback auf Volllade + client-seitige Filterung."""
+        mock_load_bbox.return_value = None
+
+        feat1 = _DummyFeature({"schluessel_gemarkungsnummer": "9999", "bezeichnung": "Fallback"})
+        all_layer = _make_layer(valid=True, features=[feat1])
+        # getFeatures mit FilterRect soll dasselbe zurückgeben
+        all_layer.getFeatures.return_value = [feat1]
+        mock_load_all.return_value = all_layer
+
+        result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect())
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0][0], "9999")
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_all_gemarkung_layer")
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox")
+    def test_both_layers_fail_returns_empty(self, mock_load_bbox, mock_load_all):
+        """Beide Lade-Strategien scheitern → leere Liste."""
+        mock_load_bbox.return_value = None
+        mock_load_all.return_value = None
+
+        result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect())
+        self.assertEqual(result, [])
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox")
+    def test_deduplication_and_sorting(self, mock_load_bbox):
+        """Doppelte Schlüssel werden dedupliziert, Ergebnis ist sortiert."""
+        feat_a = _DummyFeature({"schluessel_gemarkungsnummer": "2222", "bezeichnung": "Z"})
+        feat_b = _DummyFeature({"schluessel_gemarkungsnummer": "1111", "bezeichnung": "A"})
+        feat_dup = _DummyFeature({"schluessel_gemarkungsnummer": "2222", "bezeichnung": "Z2"})
+        mock_load_bbox.return_value = _make_layer(features=[feat_a, feat_b, feat_dup])
+
+        result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect())
+
+        nummern = [r[0] for r in result]
+        self.assertEqual(nummern, sorted(nummern))
+        self.assertEqual(len(result), 2)
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox")
+    def test_feature_without_schluessel_skipped(self, mock_load_bbox):
+        """Feature ohne Schlüssel-Attribut wird übersprungen."""
+        feat_ok = _DummyFeature({"schluessel_gemarkungsnummer": "1234", "bezeichnung": "X"})
+        feat_bad = _DummyFeature({})
+        mock_load_bbox.return_value = _make_layer(features=[feat_ok, feat_bad])
+
+        result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect())
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0][0], "1234")
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis._load_gemarkung_layer_with_bbox")
+    def test_exception_returns_empty(self, mock_load_bbox):
+        """Exception im Ladevorgang → leere Liste (kein Crash)."""
+        mock_load_bbox.side_effect = RuntimeError("WFS-Fehler")
+
+        result = gemarkung_schluessel_und_namen_by_bbox(_DummyRect())
+        self.assertEqual(result, [])
+
+
+# ---------------------------------------------------------------------------
+# Tests: gemarkung_schluessel_by_bbox (Convenience-Wrapper)
+# ---------------------------------------------------------------------------
+
+class TestGemarkungSchluesselnurNummern(unittest.TestCase):
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis.gemarkung_schluessel_und_namen_by_bbox")
+    def test_returns_only_nummern(self, mock_full):
+        mock_full.return_value = [("1234", "Alpha"), ("5678", "Beta")]
+
+        result = gemarkung_schluessel_by_bbox(_DummyRect())
+
+        self.assertEqual(result, ["1234", "5678"])
+
+    @patch("sn_verfahrensgebiet.functions.gemarkung_alkis.gemarkung_schluessel_und_namen_by_bbox")
+    def test_empty_input_returns_empty(self, mock_full):
+        mock_full.return_value = []
+
+        result = gemarkung_schluessel_by_bbox(_DummyRect())
+        self.assertEqual(result, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_grenzpunkt_alkis.py

@@ -0,0 +1,301 @@
+"""sn_verfahrensgebiet/tests/test_grenzpunkt_alkis.py
+
+Unit-Tests für :mod:`sn_verfahrensgebiet.functions.grenzpunkt_alkis`.
+
+Getestete Szenarien:
+
+Erfolgsfall:
+- Vollständiger Workflow: PunktortTA laden → GML-IDs extrahieren →
+  Grenzpunkte laden → beide Layer zurückgegeben
+
+Fehlerfälle:
+- PunktortTA-Layer ungültig → (None, None)
+- PunktortTA leer (featureCount == 0) → Warnung + (None, None)
+- Keine gehoertZu-Referenzen → (punktort_layer, None) + Warnung
+- Grenzpunkt-Layer nach IDs leer → (punktort_layer, None) + Warnung
+- Kein Verfahrensgebiet-Layer für BBox-Ableitung → (None, None)
+
+Hilfsfunktionen:
+- _extrahiere_gml_ids: xlink:href-Format wird korrekt geparst
+- _extrahiere_gml_ids: plain GML-ID ohne Doppelpunkt wird direkt verwendet
+- _extrahiere_gml_ids: leeres Feld wird übersprungen
+"""
+
+import sys
+import types
+import unittest
+from unittest.mock import patch, MagicMock
+
+# ---------------------------------------------------------------------------
+# Minimales QGIS-Mock-Setup
+# ---------------------------------------------------------------------------
+if "qgis" not in sys.modules:
+    qgis_mod = types.ModuleType("qgis")
+    qgis_core = types.ModuleType("qgis.core")
+    qgis_utils = types.ModuleType("qgis.utils")
+    qgis_utils.iface = types.SimpleNamespace()
+    qgis_utils.plugins = {}
+    qgis_mod.utils = qgis_utils
+    qgis_mod.core = qgis_core
+    sys.modules["qgis"] = qgis_mod
+    sys.modules["qgis.core"] = qgis_core
+    sys.modules["qgis.utils"] = qgis_utils
+
+from sn_verfahrensgebiet.functions.grenzpunkt_alkis import (
+    grenzpunkt_alkis,
+    _extrahiere_gml_ids,
+)
+
+
+# ---------------------------------------------------------------------------
+# Dummies
+# ---------------------------------------------------------------------------
+
+class _DummyRect:
+    def xMinimum(self): return 414000.0
+    def yMinimum(self): return 5657000.0
+    def xMaximum(self): return 415000.0
+    def yMaximum(self): return 5658000.0
+
+
+class _DummyFeature:
+    def __init__(self, gehoert_zu_val=None):
+        self._val = gehoert_zu_val
+
+    def __getitem__(self, key):
+        if key == "gehoertZu":
+            return self._val
+        raise KeyError(key)
+
+
+def _make_layer(valid=True, feature_count=3, features=None, name="Layer"):
+    layer = MagicMock()
+    layer.isValid.return_value = valid
+    layer.featureCount.return_value = feature_count if valid else 0
+    layer.getFeatures.return_value = features or []
+    layer.name.return_value = name
+    layer.id.return_value = "LYR1"
+    return layer
+
+
+# ---------------------------------------------------------------------------
+# Tests: _extrahiere_gml_ids
+# ---------------------------------------------------------------------------
+
+class TestExtrahiereGmlIds(unittest.TestCase):
+
+    def test_xlink_href_format(self):
+        """URN-Format 'urn:adv:oid:DEABC123' → nur 'DEABC123' extrahiert."""
+        feat = _DummyFeature("urn:adv:oid:DEABC123")
+        layer = _make_layer(features=[feat])
+        result = _extrahiere_gml_ids(layer)
+        self.assertEqual(result, ["DEABC123"])
+
+    def test_plain_gml_id(self):
+        """Plain-GML-ID ohne Doppelpunkt → direkt übernommen."""
+        feat = _DummyFeature("DEPLAIN456")
+        layer = _make_layer(features=[feat])
+        result = _extrahiere_gml_ids(layer)
+        self.assertEqual(result, ["DEPLAIN456"])
+
+    def test_empty_value_skipped(self):
+        """Leerer gehoertZu-Wert wird übersprungen."""
+        feat = _DummyFeature(None)
+        layer = _make_layer(features=[feat])
+        result = _extrahiere_gml_ids(layer)
+        self.assertEqual(result, [])
+
+    def test_deduplication(self):
+        """Doppelte Referenzen werden dedupliziert."""
+        f1 = _DummyFeature("urn:adv:oid:DEDUP001")
+        f2 = _DummyFeature("urn:adv:oid:DEDUP001")
+        f3 = _DummyFeature("urn:adv:oid:DEOTHER")
+        layer = _make_layer(features=[f1, f2, f3])
+        result = _extrahiere_gml_ids(layer)
+        self.assertEqual(len(result), 2)
+        self.assertIn("DEDUP001", result)
+        self.assertIn("DEOTHER", result)
+
+    def test_sorted_output(self):
+        """Rückgabe ist aufsteigend sortiert."""
+        f1 = _DummyFeature("urn:adv:oid:ZZZ")
+        f2 = _DummyFeature("urn:adv:oid:AAA")
+        layer = _make_layer(features=[f1, f2])
+        result = _extrahiere_gml_ids(layer)
+        self.assertEqual(result, sorted(result))
+
+
+# ---------------------------------------------------------------------------
+# Tests: grenzpunkt_alkis
+# ---------------------------------------------------------------------------
+
+_PATCH_PROJECT = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.QgsProject"
+_PATCH_MSGBOX = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.QMessageBox"
+_PATCH_WFS = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.QgsVectorLayer"
+_PATCH_EXTRAHIERE = "sn_verfahrensgebiet.functions.grenzpunkt_alkis._extrahiere_gml_ids"
+_PATCH_LADE_GP = "sn_verfahrensgebiet.functions.grenzpunkt_alkis._lade_grenzpunkte_per_ids"
+_PATCH_INFO = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.info"
+_PATCH_WARNING = "sn_verfahrensgebiet.functions.grenzpunkt_alkis.warning"
+
+
+class TestGrenzpunktAlkis(unittest.TestCase):
+
+    def _proj(self, vg_layer=None):
+        proj = MagicMock()
+        proj.mapLayersByName.return_value = [vg_layer] if vg_layer else []
+        return proj
+
+    # ------------------------------------------------------------------ #
+    # 1. Kein VG-Layer wenn extent=None → (None, None)                    #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_PROJECT)
+    def test_no_vg_layer_returns_none(self, mock_proj, mock_msgbox):
+        mock_proj.instance.return_value = self._proj(vg_layer=None)
+
+        pta, gp = grenzpunkt_alkis(None, extent=None)
+
+        self.assertIsNone(pta)
+        self.assertIsNone(gp)
+        mock_msgbox.warning.assert_called_once()
+
+    # ------------------------------------------------------------------ #
+    # 2. PunktortTA-Layer ungültig → (None, None)                         #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_WFS)
+    @patch(_PATCH_PROJECT)
+    def test_invalid_punktort_layer(self, mock_proj, mock_wfs, mock_msgbox):
+        proj = MagicMock()
+        proj.mapLayersByName.return_value = []
+        mock_proj.instance.return_value = proj
+        mock_wfs.return_value = _make_layer(valid=False)
+
+        pta, gp = grenzpunkt_alkis(None, extent=_DummyRect())
+
+        self.assertIsNone(pta)
+        self.assertIsNone(gp)
+        mock_msgbox.critical.assert_called_once()
+
+    # ------------------------------------------------------------------ #
+    # 3. PunktortTA leer → Warnung + (None, None)                         #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_WFS)
+    @patch(_PATCH_PROJECT)
+    def test_empty_punktort_layer(self, mock_proj, mock_wfs, mock_msgbox):
+        mock_proj.instance.return_value = MagicMock()
+        mock_wfs.return_value = _make_layer(valid=True, feature_count=0)
+
+        pta, gp = grenzpunkt_alkis(None, extent=_DummyRect())
+
+        self.assertIsNone(pta)
+        self.assertIsNone(gp)
+        mock_msgbox.warning.assert_called_once()
+
+    # ------------------------------------------------------------------ #
+    # 4. Keine gehoertZu-Referenzen → (punktort_layer, None) + Warnung    #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_WARNING)
+    @patch(_PATCH_EXTRAHIERE)
+    @patch(_PATCH_WFS)
+    @patch(_PATCH_PROJECT)
+    def test_no_gml_ids_returns_pta_only(
+        self, mock_proj, mock_wfs, mock_extrahiere, mock_warning
+    ):
+        proj = MagicMock()
+        mock_proj.instance.return_value = proj
+        pta_layer = _make_layer(valid=True, feature_count=5)
+        mock_wfs.return_value = pta_layer
+        mock_extrahiere.return_value = []
+
+        pta, gp = grenzpunkt_alkis(None, extent=_DummyRect())
+
+        self.assertIs(pta, pta_layer)
+        self.assertIsNone(gp)
+        mock_warning.assert_called_once()
+        proj.addMapLayer.assert_called_once_with(pta_layer)
+
+    # ------------------------------------------------------------------ #
+    # 5. Grenzpunkte-Batch liefert None → (punktort_layer, None) + Warn   #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_WARNING)
+    @patch(_PATCH_LADE_GP)
+    @patch(_PATCH_EXTRAHIERE)
+    @patch(_PATCH_WFS)
+    @patch(_PATCH_PROJECT)
+    def test_grenzpunkt_batch_empty(
+        self, mock_proj, mock_wfs, mock_extrahiere, mock_lade_gp, mock_warning
+    ):
+        proj = MagicMock()
+        mock_proj.instance.return_value = proj
+        pta_layer = _make_layer(valid=True, feature_count=3)
+        mock_wfs.return_value = pta_layer
+        mock_extrahiere.return_value = ["DE0001", "DE0002"]
+        mock_lade_gp.return_value = None  # Batch leer
+
+        pta, gp = grenzpunkt_alkis(None, extent=_DummyRect())
+
+        self.assertIs(pta, pta_layer)
+        self.assertIsNone(gp)
+        mock_warning.assert_called_once()
+
+    # ------------------------------------------------------------------ #
+    # 6. Vollständiger Erfolgsfall                                         #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_LADE_GP)
+    @patch(_PATCH_EXTRAHIERE)
+    @patch(_PATCH_WFS)
+    @patch(_PATCH_PROJECT)
+    def test_success_returns_both_layers(
+        self, mock_proj, mock_wfs, mock_extrahiere, mock_lade_gp
+    ):
+        proj = MagicMock()
+        mock_proj.instance.return_value = proj
+        pta_layer = _make_layer(valid=True, feature_count=10, name="PunktortTA")
+        gp_layer = _make_layer(valid=True, feature_count=10, name="Grenzpunkte")
+        mock_wfs.return_value = pta_layer
+        mock_extrahiere.return_value = ["DE001", "DE002", "DE003"]
+        mock_lade_gp.return_value = gp_layer
+
+        pta, gp = grenzpunkt_alkis(None, extent=_DummyRect())
+
+        self.assertIs(pta, pta_layer)
+        self.assertIs(gp, gp_layer)
+        # Beide Layer müssen ins Projekt eingefügt worden sein
+        calls = [c[0][0] for c in proj.addMapLayer.call_args_list]
+        self.assertIn(pta_layer, calls)
+        self.assertIn(gp_layer, calls)
+
+    # ------------------------------------------------------------------ #
+    # 7. Extent aus vorhandenem VG-Layer ableiten                         #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_LADE_GP)
+    @patch(_PATCH_EXTRAHIERE)
+    @patch(_PATCH_WFS)
+    @patch(_PATCH_PROJECT)
+    def test_extent_derived_from_vg_layer(
+        self, mock_proj, mock_wfs, mock_extrahiere, mock_lade_gp
+    ):
+        """Wenn kein extent übergeben → Extent aus VG-Layer holen."""
+        vg_layer = MagicMock()
+        vg_layer.extent.return_value = _DummyRect()
+        proj = MagicMock()
+        proj.mapLayersByName.return_value = [vg_layer]
+        mock_proj.instance.return_value = proj
+
+        pta_layer = _make_layer(valid=True, feature_count=2)
+        gp_layer = _make_layer(valid=True, feature_count=2)
+        mock_wfs.return_value = pta_layer
+        mock_extrahiere.return_value = ["DE999"]
+        mock_lade_gp.return_value = gp_layer
+
+        pta, gp = grenzpunkt_alkis(None, extent=None)
+
+        vg_layer.extent.assert_called_once()
+        self.assertIs(pta, pta_layer)
+        self.assertIs(gp, gp_layer)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_verfahrensgebiet_alkis_komplett.py

@@ -0,0 +1,608 @@
+"""sn_verfahrensgebiet/tests/test_verfahrensgebiet_alkis_komplett.py
+
+Unit-Tests für :mod:`sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett`.
+
+Getestete Szenarien:
+
+Erfolgsfall (Flächenübereinstimmung):
+- Vollständiger Workflow → enriched Layer + flst_layer + bror_layer + LoadStatus.FIRST
+- VKZ aus Projektvariable wird als Feld gesetzt (wenn vorhanden)
+- Kein VKZ in Projektvariable → Feld wird nicht gesetzt (kein Fehler)
+
+Abbruchpfade:
+- verfahrensgebiet_alkis gibt LoadStatus.NONE → (None, None, None, NONE)
+- flurstueck_alkis gibt LoadStatus.NONE → (None, None, None, NONE)
+- Keine Flurstücke selektiert → (None, flst_layer, None, NONE) + Warnung
+- erstelle_verfahrensgebiet_aus_auswahl liefert ungültigen Layer → (None, flst_layer, None, NONE)
+- Progress-Abbruch nach Schritt 1 → sofortiger sauberer Rückgabestatus
+
+Flächenvergleich:
+- Übereinstimmung → show_info_dialog, keine Umringsprüfung
+- Abweichung + Nein → kein Umringsprüfungs-Aufruf
+- Abweichung + Ja → umringspruefung aufgerufen, Ergebnislayer ins Projekt
+
+Cleanup:
+- BROR-Layer wird bei Erfolg NICHT entfernt
+- BROR-Layer wird bei Fehler entfernt (wenn in diesem Lauf hinzugefügt)
+- TempVerfahrensgebiet wird stets entfernt
+
+Hilfsfunktionen:
+- _select_by_location: processing-Fallback liefert 0 → keine Exception
+- _fuege_vkz_feld_hinzu: leeres VKZ wird ignoriert
+- _vergleiche_flaechen: korrekte Toleranzlogik
+- _lese_flaechen_toleranz: Default 0.0, parst Projektvariable
+"""
+
+import sys
+import types
+import unittest
+from unittest.mock import patch, MagicMock
+
+# ---------------------------------------------------------------------------
+# Minimales QGIS-Mock-Setup
+# ---------------------------------------------------------------------------
+if "qgis" not in sys.modules:
+    qgis_mod = types.ModuleType("qgis")
+    qgis_core = types.ModuleType("qgis.core")
+    qgis_utils = types.ModuleType("qgis.utils")
+    qgis_utils.iface = types.SimpleNamespace()
+    qgis_utils.plugins = {}
+    qgis_mod.utils = qgis_utils
+    qgis_mod.core = qgis_core
+    sys.modules["qgis"] = qgis_mod
+    sys.modules["qgis.core"] = qgis_core
+    sys.modules["qgis.utils"] = qgis_utils
+
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import (
+    verfahrensgebiet_alkis_komplett,
+    _select_by_location,
+    _fuege_vkz_feld_hinzu,
+    _vergleiche_flaechen,
+    _lese_flaechen_toleranz,
+)
+
+
+# ---------------------------------------------------------------------------
+# Dummies
+# ---------------------------------------------------------------------------
+
+class _DummyRect:
+    def xMinimum(self): return 414000.0
+    def yMinimum(self): return 5657000.0
+    def xMaximum(self): return 415000.0
+    def yMaximum(self): return 5658000.0
+
+
+def _make_layer(valid=True, name="Layer", lid="L1", feature_count=1, area=1000.0):
+    """Erzeugt einen konfigurierten Mock-Layer mit getFeatures-Support."""
+    layer = MagicMock()
+    layer.isValid.return_value = valid
+    layer.id.return_value = lid
+    layer.name.return_value = name
+    layer.extent.return_value = _DummyRect()
+    layer.featureCount.return_value = feature_count
+    layer.selectedFeatureCount.return_value = feature_count
+
+    mock_geom = MagicMock()
+    mock_geom.isEmpty.return_value = False
+    mock_geom.area.return_value = area
+    mock_geom.boundary.return_value = MagicMock(isEmpty=lambda: True)
+    mock_feat = MagicMock()
+    mock_feat.geometry.return_value = mock_geom
+    layer.getFeatures.return_value = [mock_feat]
+
+    mock_crs = MagicMock()
+    mock_crs.authid.return_value = "EPSG:25833"
+    layer.crs.return_value = mock_crs
+    return layer
+
+
+class _DummyProject:
+    def __init__(self):
+        self.added = []
+        self.removed = []
+        self._layers_by_name: dict = {}
+
+    def addMapLayer(self, layer):
+        self.added.append(layer)
+
+    def removeMapLayer(self, lid):
+        self.removed.append(lid)
+
+    def mapLayersByName(self, name):
+        return self._layers_by_name.get(name, [])
+
+    def instance(self):
+        return self
+
+
+def _make_progress_mock(canceled=False):
+    prog = MagicMock()
+    prog.is_canceled.return_value = canceled
+    return prog
+
+
+# ---------------------------------------------------------------------------
+# Patch-Konstanten
+# ---------------------------------------------------------------------------
+
+_MOD = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett"
+_PATCH_VG_ALKIS   = f"{_MOD}.verfahrensgebiet_alkis"
+_PATCH_GEM        = f"{_MOD}.gemarkung_schluessel_und_namen_von_vg_geometrie"
+_PATCH_FLST       = f"{_MOD}.flurstueck_alkis"
+_PATCH_SELECT     = f"{_MOD}._select_by_location"
+_PATCH_ERSTELLE   = f"{_MOD}.erstelle_verfahrensgebiet_aus_auswahl"
+_PATCH_VKZ        = f"{_MOD}._fuege_vkz_feld_hinzu"
+_PATCH_PROJECT    = f"{_MOD}.QgsProject"
+_PATCH_MSGBOX     = f"{_MOD}.QMessageBox"
+_PATCH_WARNING    = f"{_MOD}.warning"
+_PATCH_GET_VAR    = f"{_MOD}.get_variable"
+_PATCH_PROGRESS   = f"{_MOD}.create_progress_dialog"
+_PATCH_VERGLEICHE = f"{_MOD}._vergleiche_flaechen"
+_PATCH_SHOW_INFO  = f"{_MOD}.show_info_dialog"
+_PATCH_ASK_YES_NO = f"{_MOD}.ask_yes_no"
+_PATCH_UMRING     = f"{_MOD}.umringspruefung"
+
+
+# ---------------------------------------------------------------------------
+# Tests: _select_by_location
+# ---------------------------------------------------------------------------
+
+class TestSelectByLocation(unittest.TestCase):
+
+    def test_processing_not_available_returns_zero(self):
+        flst, vg = MagicMock(), MagicMock()
+        with patch.dict("sys.modules", {"processing": None}):
+            result = _select_by_location(flst, vg)
+        self.assertEqual(result, 0)
+
+    def test_processing_success_returns_count(self):
+        flst = MagicMock()
+        flst.selectedFeatureCount.return_value = 7
+        vg = MagicMock()
+        mock_proc = MagicMock()
+        with patch.dict("sys.modules", {"processing": mock_proc}):
+            result = _select_by_location(flst, vg)
+        self.assertEqual(result, 7)
+
+    def test_processing_exception_returns_zero(self):
+        flst, vg = MagicMock(), MagicMock()
+        mock_proc = MagicMock()
+        mock_proc.run.side_effect = RuntimeError("Kein QGIS")
+        with patch.dict("sys.modules", {"processing": mock_proc}):
+            result = _select_by_location(flst, vg)
+        self.assertEqual(result, 0)
+
+
+# ---------------------------------------------------------------------------
+# Tests: _fuege_vkz_feld_hinzu
+# ---------------------------------------------------------------------------
+
+class TestFuegeVkzFeldHinzu(unittest.TestCase):
+
+    def test_empty_vkz_does_nothing(self):
+        layer = MagicMock()
+        _fuege_vkz_feld_hinzu(layer, "")
+        layer.dataProvider.assert_not_called()
+
+    def test_none_vkz_does_nothing(self):
+        layer = MagicMock()
+        _fuege_vkz_feld_hinzu(layer, None)
+        layer.dataProvider.assert_not_called()
+
+    def test_vkz_field_added(self):
+        pr = MagicMock()
+        pr.addAttributes.return_value = True
+        pr.changeAttributeValues.return_value = True
+        layer = MagicMock()
+        layer.dataProvider.return_value = pr
+        layer.fields.return_value.indexFromName.return_value = 2
+        layer.getFeatures.return_value = []
+        _fuege_vkz_feld_hinzu(layer, "VF-2024-001")
+        pr.addAttributes.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Tests: _vergleiche_flaechen / _lese_flaechen_toleranz
+# ---------------------------------------------------------------------------
+
+class TestFlaechenvergleich(unittest.TestCase):
+
+    def _layer(self, area: float):
+        layer = MagicMock()
+        geom = MagicMock()
+        geom.isEmpty.return_value = False
+        geom.area.return_value = area
+        feat = MagicMock()
+        feat.geometry.return_value = geom
+        layer.getFeatures.return_value = [feat]
+        return layer
+
+    def test_gleiche_flaeche_stimmt_ueberein(self):
+        self.assertTrue(_vergleiche_flaechen(self._layer(1234.5), self._layer(1234.5), 0.0)[2])
+
+    def test_verschiedene_flaeche_stimmt_nicht_ueberein(self):
+        self.assertFalse(_vergleiche_flaechen(self._layer(1234.5), self._layer(1200.0), 0.0)[2])
+
+    def test_differenz_innerhalb_toleranz(self):
+        self.assertTrue(_vergleiche_flaechen(self._layer(1234.5), self._layer(1234.8), 1.0)[2])
+
+    def test_differenz_ausserhalb_toleranz(self):
+        self.assertFalse(_vergleiche_flaechen(self._layer(1234.5), self._layer(1236.0), 1.0)[2])
+
+    def test_flaechen_werte_korrekt(self):
+        bror_f, vg_f, _ = _vergleiche_flaechen(self._layer(800.0), self._layer(750.0))
+        self.assertAlmostEqual(bror_f, 800.0)
+        self.assertAlmostEqual(vg_f, 750.0)
+
+    @patch(f"{_MOD}.get_variable", return_value=None)
+    def test_lese_toleranz_default_ist_null(self, _):
+        self.assertEqual(_lese_flaechen_toleranz(), 0.0)
+
+    @patch(f"{_MOD}.get_variable", return_value="2.5")
+    def test_lese_toleranz_parst_string(self, _):
+        self.assertAlmostEqual(_lese_flaechen_toleranz(), 2.5)
+
+    @patch(f"{_MOD}.get_variable", return_value="ungueltig")
+    def test_lese_toleranz_ungueltig_fallback_null(self, _):
+        self.assertEqual(_lese_flaechen_toleranz(), 0.0)
+
+
+# ---------------------------------------------------------------------------
+# Tests: verfahrensgebiet_alkis_komplett (Hauptworkflow)
+# ---------------------------------------------------------------------------
+
+class TestVerfahrensgebietAlkisKomplett(unittest.TestCase):
+
+    # ------------------------------------------------------------------ #
+    # Abbruchpfade
+    # ------------------------------------------------------------------ #
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_VG_ALKIS)
+    def test_vg_alkis_none_aborts(self, mock_vg, mock_proj, mock_prog):
+        mock_proj.instance.return_value = _DummyProject()
+        mock_vg.return_value = (None, LoadStatus.NONE, [])
+        mock_prog.return_value = _make_progress_mock()
+
+        enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIsNone(flst)
+        self.assertIsNone(bror)
+        self.assertEqual(status, LoadStatus.NONE)
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_VG_ALKIS)
+    def test_flst_none_aborts(self, mock_vg, mock_gem, mock_proj, mock_prog):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        mock_vg.return_value = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value = [("1234", "Muster")]
+        mock_prog.return_value = _make_progress_mock()
+
+        with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)):
+            enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIsNone(flst)
+        self.assertEqual(status, LoadStatus.NONE)
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_no_selection_returns_none_enriched(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, mock_msgbox, mock_prog
+    ):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer   = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        flst_layer = _make_layer(name="Flurstücke")
+        mock_vg.return_value    = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value   = []
+        mock_flst.return_value  = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value = 0
+        mock_prog.return_value  = _make_progress_mock()
+
+        enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIs(flst, flst_layer)
+        self.assertIsNone(bror)
+        self.assertEqual(status, LoadStatus.NONE)
+        mock_msgbox.warning.assert_called_once()
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_invalid_enriched_layer_returns_none(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select, mock_erstelle, mock_msgbox, mock_prog
+    ):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer    = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        flst_layer  = _make_layer(name="Flurstücke")
+        bad_enriched = _make_layer(valid=False)
+
+        mock_vg.return_value       = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value      = []
+        mock_flst.return_value     = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value   = 5
+        mock_erstelle.return_value  = (bad_enriched, 0, 5)
+        mock_prog.return_value     = _make_progress_mock()
+
+        enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIs(flst, flst_layer)
+        self.assertIsNone(bror)
+        self.assertEqual(status, LoadStatus.NONE)
+        mock_msgbox.critical.assert_called_once()
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_VG_ALKIS)
+    def test_progress_cancel_after_step1_returns_none(self, mock_vg, mock_proj, mock_prog):
+        """Abbruch nach Schritt 1 → sofortiger sauberer Rückgabestatus."""
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        mock_vg.return_value = (vg_layer, LoadStatus.FIRST, [])
+
+        prog = MagicMock()
+        prog.is_canceled.side_effect = [False, True]  # 1. Aufruf False, 2. True
+        mock_prog.return_value = prog
+
+        enriched, flst, bror, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertEqual(status, LoadStatus.NONE)
+
+    # ------------------------------------------------------------------ #
+    # Flächenvergleich – Übereinstimmung
+    # ------------------------------------------------------------------ #
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_SHOW_INFO)
+    @patch(_PATCH_VERGLEICHE)
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_success_flaechen_match_zeigt_info(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+        mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_info, mock_prog,
+    ):
+        """Flächenübereinstimmung → show_info_dialog mit erwartetem Text."""
+        proj_inst = _DummyProject()
+        mock_proj.instance.return_value = proj_inst
+
+        vg_layer   = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        flst_layer = _make_layer(name="Flurstücke", lid="FLST1")
+        enriched   = _make_layer(name="Verfahrensgebiet", lid="VG_ENRICHED", valid=True)
+
+        mock_vg.return_value       = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value      = [("1234", "Muster")]
+        mock_flst.return_value     = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value   = 3
+        mock_erstelle.return_value  = (enriched, 3, 0)
+        mock_get_var.return_value  = "VF-2024-001"
+        mock_vergl.return_value    = (1234.5, 1234.5, True)
+        mock_prog.return_value     = _make_progress_mock()
+
+        result_enriched, result_flst, result_bror, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIs(result_enriched, enriched)
+        self.assertIs(result_flst, flst_layer)
+        self.assertIs(result_bror, vg_layer)
+        self.assertEqual(status, LoadStatus.FIRST)
+        mock_vkz.assert_called_once_with(enriched, "VF-2024-001")
+        mock_info.assert_called_once()
+        self.assertIn("erwarteter Geometrie", mock_info.call_args[0][1])
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_SHOW_INFO)
+    @patch(_PATCH_VERGLEICHE)
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_no_vkz_variable_passes_empty_string(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+        mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_info, mock_prog,
+    ):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer   = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        flst_layer = _make_layer(name="Flurstücke")
+        enriched   = _make_layer(name="Verfahrensgebiet", valid=True)
+
+        mock_vg.return_value       = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value      = []
+        mock_flst.return_value     = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value   = 2
+        mock_erstelle.return_value  = (enriched, 2, 0)
+        mock_get_var.return_value  = ""
+        mock_vergl.return_value    = (1000.0, 1000.0, True)
+        mock_prog.return_value     = _make_progress_mock()
+
+        verfahrensgebiet_alkis_komplett(None)
+
+        mock_vkz.assert_called_once_with(enriched, "")
+
+    # ------------------------------------------------------------------ #
+    # Flächenvergleich – Abweichung + Nein
+    # ------------------------------------------------------------------ #
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_UMRING)
+    @patch(_PATCH_ASK_YES_NO)
+    @patch(_PATCH_VERGLEICHE)
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_flaechen_abweichung_nein_kein_umring(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+        mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask, mock_umring, mock_prog,
+    ):
+        """Abweichung + Nein → umringspruefung wird NICHT aufgerufen."""
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer   = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        flst_layer = _make_layer(name="Flurstücke")
+        enriched   = _make_layer(name="Verfahrensgebiet", valid=True)
+
+        mock_vg.return_value       = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value      = []
+        mock_flst.return_value     = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value   = 3
+        mock_erstelle.return_value  = (enriched, 3, 0)
+        mock_get_var.return_value  = ""
+        mock_vergl.return_value    = (1000.0, 800.0, False)
+        mock_ask.return_value      = False
+        mock_prog.return_value     = _make_progress_mock()
+
+        result_enriched, _, _, status = verfahrensgebiet_alkis_komplett(None)
+
+        mock_umring.assert_not_called()
+        self.assertIs(result_enriched, enriched)
+        self.assertEqual(status, LoadStatus.FIRST)
+
+    # ------------------------------------------------------------------ #
+    # Flächenvergleich – Abweichung + Ja → Umringsprüfung
+    # ------------------------------------------------------------------ #
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_SHOW_INFO)
+    @patch(_PATCH_UMRING)
+    @patch(_PATCH_ASK_YES_NO)
+    @patch(_PATCH_VERGLEICHE)
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_flaechen_abweichung_ja_startet_umringspruefung(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+        mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_ask,
+        mock_umring, mock_info, mock_prog,
+    ):
+        """Abweichung + Ja → umringspruefung aufgerufen, Ergebnislayer ins Projekt,
+        Abschlussmeldung mit Konfliktanzahl."""
+        proj_inst = _DummyProject()
+        mock_proj.instance.return_value = proj_inst
+
+        vg_layer       = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)")
+        flst_layer     = _make_layer(name="Flurstücke")
+        enriched       = _make_layer(name="Verfahrensgebiet", valid=True)
+        ergebnis_layer = _make_layer(name="Umringsprüfung", lid="UMRING_L")
+
+        mock_vg.return_value       = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value      = []
+        mock_flst.return_value     = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value   = 3
+        mock_erstelle.return_value  = (enriched, 3, 0)
+        mock_get_var.return_value  = ""
+        mock_vergl.return_value    = (1000.0, 800.0, False)
+        mock_ask.return_value      = True
+        mock_umring.return_value   = (ergebnis_layer, 4)
+        mock_prog.return_value     = _make_progress_mock()
+
+        result_enriched, _, _, status = verfahrensgebiet_alkis_komplett(None)
+
+        mock_umring.assert_called_once()
+        self.assertIn(ergebnis_layer, proj_inst.added)
+        mock_info.assert_called_once()
+        self.assertIn("4", mock_info.call_args[0][1])
+        self.assertIn("Konfliktstellen", mock_info.call_args[0][1])
+        self.assertEqual(status, LoadStatus.FIRST)
+
+    # ------------------------------------------------------------------ #
+    # Cleanup-Verhalten
+    # ------------------------------------------------------------------ #
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_SHOW_INFO)
+    @patch(_PATCH_VERGLEICHE)
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_VG_ALKIS)
+    def test_bror_layer_bleibt_bei_erfolg_im_projekt(
+        self, mock_vg, mock_flst, mock_gem, mock_proj, mock_select,
+        mock_erstelle, mock_get_var, mock_vkz, mock_vergl, mock_info, mock_prog,
+    ):
+        """Bei Erfolg darf der BROR-Layer NICHT entfernt werden."""
+        proj_inst = _DummyProject()
+        mock_proj.instance.return_value = proj_inst
+
+        vg_layer   = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)", lid="BROR_ID")
+        flst_layer = _make_layer(name="Flurstücke")
+        enriched   = _make_layer(name="Verfahrensgebiet", valid=True)
+
+        mock_vg.return_value       = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value      = []
+        mock_flst.return_value     = (flst_layer, LoadStatus.FIRST)
+        mock_select.return_value   = 3
+        mock_erstelle.return_value  = (enriched, 3, 0)
+        mock_get_var.return_value  = ""
+        mock_vergl.return_value    = (1000.0, 1000.0, True)
+        mock_prog.return_value     = _make_progress_mock()
+
+        verfahrensgebiet_alkis_komplett(None)
+
+        self.assertNotIn("BROR_ID", proj_inst.removed)
+
+    @patch(_PATCH_PROGRESS)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_GEM)
+    @patch(_PATCH_VG_ALKIS)
+    def test_bror_layer_wird_bei_fehler_entfernt(
+        self, mock_vg, mock_gem, mock_proj, mock_prog
+    ):
+        """Bei Fehler (flst_alkis NONE) wird der BROR-Layer aus dem Projekt entfernt."""
+        proj_inst = _DummyProject()
+        mock_proj.instance.return_value = proj_inst
+
+        vg_layer = _make_layer(name="BauRaumOrdnungsrecht (ALKIS)", lid="BROR_ERR")
+        mock_vg.return_value   = (vg_layer, LoadStatus.FIRST, [])
+        mock_gem.return_value  = []
+        mock_prog.return_value = _make_progress_mock()
+
+        with patch(_PATCH_FLST, return_value=(None, LoadStatus.NONE)):
+            verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIn("BROR_ERR", proj_inst.removed)
+
+
+if __name__ == "__main__":
+    unittest.main()
+

tests/test_working_tab.py

@@ -18,9 +18,10 @@ from sn_verfahrensgebiet.ui.tabs.working_tab import WorkingTab, LoadStatus
 
 
 class _DummyLayer:
-    def __init__(self, valid=True, layer_id="L1"):
+    def __init__(self, valid=True, layer_id="L1", layer_name="DummyLayer"):
         self._valid = valid
         self._id = layer_id
+        self._name = layer_name
 
     def isValid(self):
         return self._valid
@@ -28,6 +29,9 @@ class _DummyLayer:
     def id(self):
         return self._id
 
+    def name(self):
+        return self._name
+
     def extent(self):
         return object()
 
@@ -83,11 +87,13 @@ class TestWorkingTab(unittest.TestCase):
         tab = WorkingTab.__new__(WorkingTab)
         tab.btn_verf_alkis = _DummyButton()
         tab.btn_flurst_alkis = _DummyButton()
+        tab.btn_grenz_alkis = _DummyButton()
         tab.btn_verf_aus_auswahl = _DummyButton()
         tab.btn_verf_aus_liste = _DummyButton()
         tab.verf_layer_combo = _DummyCombo()
-        tab.haken_verf = object()
-        tab.haken_flurst = object()
+        tab.haken_verf = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None)
+        tab.haken_flurst = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None)
+        tab.haken_grenz = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None)
         tab.setze_haken = lambda *_args, **_kwargs: None
         # Pruefmanager-Mock: verarbeite() wird im Test ggf. überschrieben
         tab.pruefmanager = types.SimpleNamespace(
@@ -126,39 +132,83 @@ class TestWorkingTab(unittest.TestCase):
     @patch("sn_verfahrensgebiet.ui.tabs.working_tab.success")
     @patch("sn_verfahrensgebiet.ui.tabs.working_tab.apply_style")
     @patch("sn_verfahrensgebiet.ui.tabs.working_tab.QgsProject.instance")
-    @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis")
+    @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis_komplett")
     @patch("sn_verfahrensgebiet.ui.tabs.working_tab.iface")
     @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id")
     def test_handle_verf_alkis_success_path(
         self,
         mock_set_layer_id,
         mock_iface,
-        mock_alkis,
+        mock_alkis_komplett,
         mock_project_instance,
         mock_apply_style,
         mock_success,
     ):
+        """verfahrensgebiet_alkis_komplett → 4-Tupel (enriched, flst, bror, FIRST)."""
         tab = self._make_tab()
-        layer = _DummyLayer(valid=True, layer_id="layer_ok")
-        mock_alkis.return_value = (layer, LoadStatus.FIRST)
-        mock_project_instance.return_value = _DummyProject()
+        enriched = _DummyLayer(valid=True, layer_id="enriched_ok")
+        flst = _DummyLayer(valid=True, layer_id="flst_ok")
+        mock_alkis_komplett.return_value = (enriched, flst, None, LoadStatus.FIRST)
+        proj = _DummyProject()
+        proj.mapLayersByName = lambda _n: []  # kein bestehender Flst-Layer
+        mock_project_instance.return_value = proj
         mock_iface.mapCanvas.return_value = _DummyCanvas()
 
         tab.handle_verf_alkis()
 
-        mock_apply_style.assert_called_once()
+        mock_apply_style.assert_called()
         mock_success.assert_called_once()
-        mock_set_layer_id.assert_called_with("layer_ok")
+        mock_set_layer_id.assert_called_with("enriched_ok")
 
     @patch("sn_verfahrensgebiet.ui.tabs.working_tab.set_verfahrensgebiet_layer_id")
-    @patch("sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis", return_value=(None, LoadStatus.NONE))
+    @patch(
+        "sn_verfahrensgebiet.ui.tabs.working_tab.verfahrensgebiet_alkis_komplett",
+        return_value=(None, None, None, LoadStatus.NONE),
+    )
     def test_handle_verf_alkis_failure_clears_layer_var(self, _mock_alkis, mock_set_layer_id):
+        """Status NONE → Layer-Variable wird auf '' gesetzt."""
         tab = self._make_tab()
 
         tab.handle_verf_alkis()
 
         mock_set_layer_id.assert_called_once_with("")
 
+    # ------------------------------------------------------------------
+    # handle_grenz_alkis
+    # ------------------------------------------------------------------
+
+    @patch("sn_verfahrensgebiet.ui.tabs.working_tab.grenzpunkt_alkis")
+    def test_handle_grenz_alkis_success_sets_haken(self, mock_grenz):
+        """Grenzpunkte erfolgreich geladen → haken_grenz wird gesetzt."""
+        tab = self._make_tab()
+        haken_calls = []
+        tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v))
+
+        pta = _DummyLayer(valid=True)
+        gp = _DummyLayer(valid=True)
+        mock_grenz.return_value = (pta, gp)
+
+        tab.handle_grenz_alkis()
+
+        self.assertEqual(len(haken_calls), 1)
+        self.assertIs(haken_calls[0][0], tab.haken_grenz)
+        self.assertTrue(haken_calls[0][1])
+
+    @patch("sn_verfahrensgebiet.ui.tabs.working_tab.grenzpunkt_alkis")
+    def test_handle_grenz_alkis_none_result_sets_no_haken(self, mock_grenz):
+        """grenzpunkt_alkis liefert (None, None) → kein Haken."""
+        tab = self._make_tab()
+        haken_calls = []
+        tab.setze_haken = lambda lbl, v: haken_calls.append((lbl, v))
+
+        mock_grenz.return_value = (None, None)
+
+        tab.handle_grenz_alkis()
+
+        self.assertEqual(len(haken_calls), 0)
+        # Button wieder aktiviert (finally)
+        self.assertTrue(tab.btn_grenz_alkis.isEnabled())
+
     # ------------------------------------------------------------------
     # handle_flurst_alkis
     # ------------------------------------------------------------------

ui/tabs/working_tab.py

@@ -13,8 +13,12 @@ from sn_basis.modules.Pruefmanager import Pruefmanager
 from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
 from qgis.utils import iface
 
-from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import verfahrensgebiet_alkis, LoadStatus
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis import LoadStatus
 from sn_verfahrensgebiet.functions.flurstueck_alkis import flurstueck_alkis
+from sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett import verfahrensgebiet_alkis_komplett
+# TODO Grenzpunkt-Pipeline: Import reaktivieren, sobald grenzpunkt_alkis.py
+#      mit der neuen Pipeline (PunktortTA → istTeilVon → Einzel-URN) umgeschrieben wurde.
+# from sn_verfahrensgebiet.functions.grenzpunkt_alkis import grenzpunkt_alkis
 from sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service import (
     pruefe_flurstueck_layer,
     pruefe_auswahl,
@@ -87,6 +91,20 @@ class WorkingTab(QWidget):
         row_flurst_buttons.addWidget(self.btn_flurst_shape)
         layout.addLayout(row_flurst_buttons)
 
+        # Abschnitt 2b: Grenzpunkte
+        row_grenz_header = QHBoxLayout()
+        lbl_grenz = QLabel("<b>Grenzpunkte</b>")
+        self.haken_grenz = self._haken_label()
+        row_grenz_header.addWidget(lbl_grenz)
+        row_grenz_header.addWidget(self.haken_grenz)
+        row_grenz_header.addStretch()
+        layout.addLayout(row_grenz_header)
+
+        row_grenz_buttons = QHBoxLayout()
+        self.btn_grenz_alkis = QPushButton("Grenzpunkte aus ALKIS laden")
+        row_grenz_buttons.addWidget(self.btn_grenz_alkis)
+        layout.addLayout(row_grenz_buttons)
+
         # Abschnitt 3: Verfahrensgebiet erstellen
         row_verf_erstellen_header = QHBoxLayout()
         lbl_verf_erstellen = QLabel("<b>Verfahrensgebiet erstellen</b>")
@@ -110,6 +128,7 @@ class WorkingTab(QWidget):
         # Signale verbinden
         self.btn_verf_alkis.clicked.connect(self.handle_verf_alkis)
         self.btn_flurst_alkis.clicked.connect(self.handle_flurst_alkis)
+        self.btn_grenz_alkis.clicked.connect(self.handle_grenz_alkis)
         self.btn_verf_aus_auswahl.clicked.connect(self.handle_verf_aus_auswahl)
         self.btn_verf_aus_liste.clicked.connect(self.handle_verf_aus_liste)
         # self.btn_verf_shape.clicked.connect(self.handle_verf_shape)
@@ -118,36 +137,48 @@ class WorkingTab(QWidget):
         # Beim Init: aktuelle Verfahrensgebiet-Layer-ID laden und im Dropdown anzeigen
         self._restore_verf_layer_selection()
 
-    def handle_verf_alkis(self):
+    def handle_verf_alkis(self) -> None:
+        """Startet den vollständigen VG-Lade-Workflow (ALKIS → Gemarkungen → Flurstücke →
+        Verschneidung → Flächenvergleich → opt. Umringsprüfung).
+
+        Delegiert an :func:`~sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.verfahrensgebiet_alkis_komplett`.
+        Nach Erfolg werden enriched VG-Layer, Flurstücks-Layer und BROR-Layer
+        ins Projekt eingefügt, Styles angewendet und die UI-Elemente aktualisiert.
+        """
         self._set_busy(self.btn_verf_alkis, True, "Aus ALKIS laden")
         try:
-            layer, status = verfahrensgebiet_alkis(self)
-            if not layer or not layer.isValid():
+            enriched_layer, flst_layer, bror_layer, status = verfahrensgebiet_alkis_komplett(self)
+
+            if status == LoadStatus.NONE or not enriched_layer:
                 set_verfahrensgebiet_layer_id("")
                 return
 
-            if status in (LoadStatus.FIRST, LoadStatus.RELOAD):
-                QgsProject.instance().addMapLayer(layer)
-                iface.mapCanvas().setExtent(layer.extent())
+            project = QgsProject.instance()
 
-                apply_style(layer, "verfahrensgebiet.qml")
-                iface.mapCanvas().refresh()
-                self.setze_haken(self.haken_verf, True)
+            # Flurstücks-Layer ins Projekt (sofern neu geladen)
+            if flst_layer and flst_layer.isValid():
+                existing_flst = project.mapLayersByName(flst_layer.name())
+                if not existing_flst:
+                    project.addMapLayer(flst_layer)
+                    apply_style(flst_layer, "GIS_Flst_Beschriftung_ALKIS_NAS.qml")
+                self.setze_haken(self.haken_flurst, True)
 
-                self.verf_layer_combo.setLayer(layer)
-                self._on_verf_layer_changed(layer)
+            # BROR-Layer: einfache Darstellung setzen (roter Umring, transparent)
+            if bror_layer and bror_layer.isValid():
+                _setze_bror_stil(bror_layer)
 
-                msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen"
-                success("Verfahrensgebiet", f"Layer wurde {msg}.", duration=5)
+            # Enriched VG-Layer ins Projekt
+            project.addMapLayer(enriched_layer)
+            apply_style(enriched_layer, "verfahrensgebiet.qml")
+            iface.mapCanvas().setExtent(enriched_layer.extent())
+            iface.mapCanvas().refresh()
+            self.setze_haken(self.haken_verf, True)
 
-            elif status == LoadStatus.KEEP:
-                self.verf_layer_combo.setLayer(layer)
-                self._on_verf_layer_changed(layer)
-                info("Verfahrensgebiet", "Vorhandener Layer wurde behalten.", duration=4)
+            self.verf_layer_combo.setLayer(enriched_layer)
+            self._on_verf_layer_changed(enriched_layer)
 
-            elif status == LoadStatus.NONE:
-                self._on_verf_layer_changed(None)
-                warning("Verfahrensgebiet", "Layer konnte nicht geladen werden.", duration=None)
+            msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen"
+            success("Verfahrensgebiet", f"Layer wurde {msg} (inkl. Flurstücksverschneidung).", duration=5)
 
         finally:
             self._set_busy(self.btn_verf_alkis, False, "Aus ALKIS laden")
@@ -368,3 +399,73 @@ class WorkingTab(QWidget):
             "Die Funktion 'Aus Liste' steht in einer zukünftigen Version zur Verfügung.",
             parent=self,
         )
+
+    # ------------------------------------------------------------------
+    # Grenzpunkt-Handler
+    # ------------------------------------------------------------------
+
+    def handle_grenz_alkis(self) -> None:
+        """Lädt Grenzpunkte (AX_PunktortTA + AX_Grenzpunkt) aus ALKIS.
+
+        TODO Grenzpunkt-Pipeline: Implementierung ausstehend.
+        Neue Pipeline: VG-Geometrie → PunktortTA (spatial filter) →
+        istTeilVon-Attribute → Grenzpunkt per Einzel-URN laden.
+        Bis dahin ist dieser Handler deaktiviert.
+        """
+        # TODO Grenzpunkt-Pipeline: Körper reaktivieren, sobald grenzpunkt_alkis.py
+        #      mit der neuen Pipeline umgeschrieben wurde.
+        #
+        # self._set_busy(self.btn_grenz_alkis, True, "Grenzpunkte aus ALKIS laden")
+        # try:
+        #     punktort_layer, grenzpunkt_layer = grenzpunkt_alkis(self)
+        #
+        #     if grenzpunkt_layer and grenzpunkt_layer.isValid():
+        #         iface.mapCanvas().refresh()
+        #         self.setze_haken(self.haken_grenz, True)
+        #         success("Grenzpunkte", "Grenzpunkte erfolgreich geladen.", duration=5)
+        #     elif punktort_layer and punktort_layer.isValid():
+        #         iface.mapCanvas().refresh()
+        #         self.setze_haken(self.haken_grenz, True)
+        #         info(
+        #             "Grenzpunkte",
+        #             "PunktortTA-Layer geladen. Grenzpunkt-IDs konnten nicht aufgelöst werden.",
+        #             duration=6,
+        #         )
+        # finally:
+        #     self._set_busy(self.btn_grenz_alkis, False, "Grenzpunkte aus ALKIS laden")
+        from sn_basis.functions.message_wrapper import warning as _warn
+        _warn("Grenzpunkte", "Grenzpunkt-Pipeline ist noch nicht implementiert.", duration=5)
+
+
+# ---------------------------------------------------------------------------
+# Modul-Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _setze_bror_stil(layer) -> None:
+    """Setzt eine einfache Darstellung für den BauRaumOrdnungsrecht-Layer.
+
+    Roter Umring, halbtransparente Füllung; kein QGIS-Kontext → stilles Ignorieren.
+
+    Parameters
+    ----------
+    layer :
+        BauRaumOrdnungsrecht-Memory-Layer (Polygon).
+    """
+    try:
+        from qgis.core import (  # type: ignore[import]
+            QgsSimpleFillSymbolLayer,
+            QgsFillSymbol,
+            QgsSingleSymbolRenderer,
+        )
+        from qgis.PyQt.QtGui import QColor  # type: ignore[import]
+        from qgis.PyQt.QtCore import Qt  # type: ignore[import]
+
+        fill = QgsSimpleFillSymbolLayer()
+        fill.setColor(QColor(220, 50, 50, 40))       # halbtransparent rot
+        fill.setStrokeColor(QColor(220, 50, 50))
+        fill.setStrokeWidth(0.8)
+        symbol = QgsFillSymbol([fill])
+        layer.setRenderer(QgsSingleSymbolRenderer(symbol))
+        layer.triggerRepaint()
+    except Exception:
+        pass  # Kein QGIS-Kontext (Tests, Mock-Modus) → Standarddarstellung