verfahrensgebiet aus Flst

functions/flurstueck_alkis.py

@@ -1,15 +1,35 @@
 """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
@@ -49,65 +69,173 @@ def _check_existing_flurstueck_layer(tab_widget):
         return existing_layers[0], LoadStatus.KEEP
 
 
-def flurstueck_alkis(tab_widget):
+def _lade_flurstuecke_per_gemarkung(schluessel_liste: List[str]):
-    """Lädt Flurstücke aus ALKIS-WFS anhand von Gemarkungsschlüsseln.
+    """Lädt Flurstücke für mehrere Gemarkungen, eine WFS-Anfrage pro Schlüssel.
 
-    Der Nutzer gibt einen oder mehrere Gemarkungsschlüssel (kommagetrennt) ein.
+    Vermeidet das WFS-Feature-Limit einer einzelnen Abfrage mit mehreren
-    Der WFS wird per SQL-Filter auf
+    Gemarkungsnummern. Alle Features werden in einem Memory-Layer zusammengeführt.
-    ``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN (...)`` abgefragt.
+
+    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_NAS_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.
+        Memory-Layer mit allen Flurstücken und Ladestatus.
-        Bei Fehlern ist der Layer None und der Status LoadStatus.NONE.
+        Bei Fehlern ist der Layer ``None`` und der Status :attr:`LoadStatus.NONE`.
     """
-    # 1. Vorhandenen Layer prüfen
+    # --- Schritt 1: Vorhandenen Layer behandeln ----------------------------
-    existing, status = _check_existing_flurstueck_layer(tab_widget)
+    if ohne_dialog:
-    if status == LoadStatus.KEEP:
+        project = QgsProject.instance()
-        return existing, status
+        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
+    # --- Schritt 2: Gemarkungsschlüssel bestimmen -------------------------
-    eingabe, ok = ask_text(
+    if ohne_dialog:
-        "Gemarkungsschlüssel",
+        # Pipeline-Modus: vorgabe_schluessel direkt verwenden, kein Dialog
-        "Gemarkungsschlüssel eingeben (mehrere mit Komma trennen):",
+        if not vorgabe_schluessel:
-        parent=tab_widget,
+            return None, LoadStatus.NONE
-    )
+        schluessel_liste = [s.strip() for s in vorgabe_schluessel if s.strip()]
-    if not ok or not eingabe.strip():
+        if not schluessel_liste:
-        QMessageBox.warning(
+            return None, LoadStatus.NONE
-            tab_widget,
+    else:
-            "Abgebrochen",
+        # Interaktiver Modus: Dialog mit Vorschlag
-            "Bitte mindestens einen gültigen Gemarkungsschlüssel eingeben.",
+        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()]
+        schluessel_liste = [s.strip() for s in eingabe.split(",") if s.strip()]
-    if not schluessel_liste:
+        if not schluessel_liste:
-        QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.")
+            QMessageBox.warning(tab_widget, "Fehler", "Keine gültigen Schlüssel gefunden.")
-        return None, LoadStatus.NONE
+            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
+    if not layer or not layer.isValid() or layer.featureCount() == 0:
-    wfs_uri = (
+        if not ohne_dialog:
-        f"url='{ALKIS_NAS_URL}' typename='{FLST_TYPENAME}' srsname='EPSG:25833' "
+            QMessageBox.critical(
-        f"sql=SELECT * FROM AX_Flurstueck "
+                tab_widget,
-        f"WHERE gemarkung_ax_gemarkung_schluessel_gemarkungsnummer IN ({schluessel_str})"
+                "Fehler",
-    )
+                "Flurstück-Layer konnte nicht geladen werden.\n"
-    layer = QgsVectorLayer(wfs_uri, FLST_LAYER_NAME, "WFS")
+                "Bitte Gemarkungsschlüssel und Netzwerkverbindung prüfen.",
-
+            )
-    if not layer.isValid():
-        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_NAS_URL = "https://geodienste.sachsen.de/aaa/public_alkis/nas/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_NAS_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_NAS_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/verfahrensgebiet_alkis.py

@@ -65,18 +65,23 @@ def verfahrensgebiet_alkis(tab_widget):
         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 = (
@@ -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
+    # Status: FIRST oder RELOAD; features für Gemarkungsermittlung weitergeben
-    return memory_layer, status
+    return memory_layer, status, features

functions/verfahrensgebiet_alkis_komplett.py

@@ -0,0 +1,309 @@
+"""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:
+
+1. Verfahrensgebiet aus ``AX_BauRaumOderBodenordnungsrecht`` per VKZ laden.
+2. BBox des VG ableiten.
+3. Gemarkungsschlüssel per BBox aus ``AX_Gemarkung`` vorschlagen.
+4. Flurstücke per Dialog (mit Vorschlag) laden.
+5. Flurstücke räumlich auf das VG einschränken (``selectByLocation``).
+6. Aus der Auswahl den enriched Verfahrensgebiet-Memory-Layer erzeugen.
+7. ALKIS-VG-Hilfs-Layer entfernen, enriched Layer zurückgeben.
+
+Der enriched Layer enthält die Felder ``flurstueckskennzeichen`` und
+``gemarkung_ax_gemarkung_schluessel_gemarkungsnummer`` des
+:mod:`~sn_verfahrensgebiet.functions.verfahrensgebiet_edit_service`-Schemas
+und ist damit vollständig kompatibel mit ``ergaenze_verfahrensgebiet`` und
+``entferne_aus_verfahrensgebiet``.
+
+**Abbruchverhalten:** Jeder Schritt, der :attr:`LoadStatus.NONE` zurückgibt
+oder kein Ergebnis liefert, bricht den Workflow ab. ALKIS-Hilfs-Layer werden
+dabei gesäubert, damit das Projekt nicht in einem inkonsistenten Zustand
+verbleibt.
+"""
+
+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 info, warning
+from sn_basis.functions.variable_wrapper import get_variable
+
+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,
+    FELD_KENNZEICHEN,
+    VERF_LAYER_NAME,
+)
+
+# Feld für das Verfahrenskennzeichen (VKZ) im enriched Layer
+FELD_VKZ = "verfahrensnummer"
+
+
+# ---------------------------------------------------------------------------
+# Interne Hilfsfunktionen
+# ---------------------------------------------------------------------------
+
+def _entferne_hilfs_layer(project, layer_name: str) -> None:
+    """Entfernt alle Layer mit dem angegebenen Namen aus dem Projekt.
+
+    Wird verwendet, um den temporären ALKIS-VG-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 das Verfahrensgebiet überlappen oder darin liegen.
+
+    Verwendet ``processing.run("native:selectbylocation")`` mit den Prädikaten
+    ``overlaps`` (5) und ``within`` (6). Flurstücke, die das VG nur berühren
+    (gemeinsame Grenzlinie oder -punkt, aber kein gemeinsames Inneres), werden
+    damit ausgeschlossen.
+
+    Parameters
+    ----------
+    flst_layer :
+        Flurstücks-WFS-Layer (Quelle der Selektion).
+    vg_layer :
+        Verfahrensgebiet-Memory-Layer (Schnittmaske).
+
+    Returns
+    -------
+    int
+        Anzahl der selektierten Features.
+    """
+    try:
+        import processing  # type: ignore[import]
+
+        processing.run(
+            "native:selectbylocation",
+            {
+                "INPUT": flst_layer,
+                "PREDICATE": [5, 6],  # 5 = overlaps, 6 = within; schließt reines Berühren aus
+                "INTERSECT": vg_layer,
+                "METHOD": 0,          # 0 = neue Selektion
+            },
+        )
+        return flst_layer.selectedFeatureCount()
+
+    except Exception:
+        # Fallback: keine Selektion (z. B. im Unit-Test-Modus)
+        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.
+
+    Das Feld wird nur hinzugefügt, wenn ``vkz`` nicht leer ist, damit der
+    Layer ohne Projektvariable weiterhin funktioniert.
+
+    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
+
+
+# ---------------------------------------------------------------------------
+# Öffentliche API
+# ---------------------------------------------------------------------------
+
+def verfahrensgebiet_alkis_komplett(
+    tab_widget,
+) -> Tuple[Optional[object], Optional[object], LoadStatus]:
+    """Vollständiger VG-Lade-Workflow: ALKIS-VG → BBox → Gemarkungen →
+    Flurstücke → Verschneidung → enriched Memory-Layer.
+
+    Schrittfolge:
+
+    1. ``verfahrensgebiet_alkis(tab_widget)`` – Verfahrensgebiet temporär laden
+    2. ``vg_layer.extent()`` – BBox ableiten
+    3. ``gemarkung_schluessel_und_namen_by_bbox(extent)`` – Schlüssel vorschlagen
+    4. ``flurstueck_alkis(tab_widget, vorgabe_schluessel, vorgabe_namen, bbox)`` – Flurstücke laden
+    5. ``_select_by_location(flst_layer, vg_layer)`` – räumliche Selektion
+    6. ``erstelle_verfahrensgebiet_aus_auswahl(flst_layer)`` – enriched Layer erzeugen
+    7. Ggf. VKZ-Feld setzen, ALKIS-VG-Hilfs-Layer entfernen
+
+    Parameters
+    ----------
+    tab_widget :
+        Eltern-Widget für alle Dialoge (weitergereicht an Unteraufrufe).
+
+    Returns
+    -------
+    Tuple[Optional[enriched_layer], Optional[flst_layer], LoadStatus]
+        - ``enriched_layer``: Memory-Layer „Verfahrensgebiet" mit
+          ``flurstueckskennzeichen``- und ``gemarkung``-Feldern, oder ``None``
+          bei Abbruch/Fehler.
+        - ``flst_layer``: Geladener WFS-Flurstück-Layer (bleibt im Projekt),
+          oder ``None``.
+        - ``LoadStatus``: Status des VG-Lade-Vorgangs (``FIRST``, ``RELOAD``
+          oder ``NONE`` bei Abbruch).
+
+    Notes
+    -----
+    Der temporäre ALKIS-VG-Layer (aus ``AX_BauRaumOderBodenordnungsrecht``)
+    wird nach der Verschneidung aus dem Projekt entfernt. Im Projekt verbleiben
+    nur der enriched Memory-Layer und der Flurstücks-WFS-Layer.
+    """
+    project = QgsProject.instance()
+    alkis_vg_layer = None
+
+    try:
+        # ------------------------------------------------------------------
+        # Schritt 1: VG aus ALKIS laden (Zwischenobjekt)
+        # ------------------------------------------------------------------
+        info("VG laden", "Schritt 1/6: Verfahrensgebiet aus ALKIS laden …", duration=3)
+        alkis_vg_layer, vg_status, vg_features = verfahrensgebiet_alkis(tab_widget)
+
+        if vg_status == LoadStatus.NONE or not alkis_vg_layer:
+            return None, None, LoadStatus.NONE
+
+        # Temporär ins Projekt hinzufügen, damit selectByLocation ihn findet
+        if vg_status in (LoadStatus.FIRST, LoadStatus.RELOAD):
+            project.addMapLayer(alkis_vg_layer)
+
+        # ------------------------------------------------------------------
+        # Schritt 2 (entfernt): BBox-Ableitung war hier – WFS unterstützt kein
+        # BBox-Filter, daher wird der Extent nicht mehr benötigt.
+        # ------------------------------------------------------------------
+
+        # ------------------------------------------------------------------
+        # Schritt 3: Gemarkungsschlüssel per Geometrie-Spatial-Join ermitteln
+        # ------------------------------------------------------------------
+        info("VG laden", "Schritt 2/6: Gemarkungsschlüssel ermitteln \u2026", duration=3)
+        # VG-Geometrie aus den Roh-WFS-Features bilden (Union, damit
+        # alle Teilflächen berücksichtigt werden)
+        from sn_basis.functions import QgsGeometry as _QgsGeometry
+        if vg_features:
+            vg_geom = _QgsGeometry.unaryUnion([f.geometry() for f in vg_features])
+        else:
+            # Bei KEEP-Status keine WFS-Features vorhanden; Fallback auf Layergeometrie
+            vg_geom = next(alkis_vg_layer.getFeatures()).geometry()
+        schluessel_namen = gemarkung_schluessel_und_namen_von_vg_geometrie(vg_geom)
+        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,
+            )
+
+        # ------------------------------------------------------------------
+        # Schritt 4: Flurstücke laden (mit Nutzer-Dialog, Vorschlag befüllt)
+        # ------------------------------------------------------------------
+        info("VG laden", "Schritt 3/6: Flurstücke aus ALKIS laden …", duration=3)
+        flst_layer, flst_status = flurstueck_alkis(
+            tab_widget,
+            vorgabe_schluessel=vorgabe_schluessel,
+            vorgabe_namen=vorgabe_namen if vorgabe_namen else None,
+            # bbox wird nicht übergeben – WFS unterstützt keinen BBox-Filter
+            ohne_dialog=True,  # kein Nutzer-Dialog in der Pipeline
+        )
+
+        if flst_status == LoadStatus.NONE or not flst_layer:
+            return None, None, LoadStatus.NONE
+
+        # ------------------------------------------------------------------
+        # Schritt 5: Räumliche Selektion
+        # ------------------------------------------------------------------
+        info("VG laden", "Schritt 4/6: Flurstücke räumlich selektieren …", duration=3)
+        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, LoadStatus.NONE
+
+        # ------------------------------------------------------------------
+        # Schritt 6: Enriched Memory-Layer aus Auswahl erzeugen
+        # ------------------------------------------------------------------
+        info("VG laden", "Schritt 5/6: Verfahrensgebiet-Layer erzeugen …", duration=3)
+        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, LoadStatus.NONE
+
+        # ------------------------------------------------------------------
+        # Schritt 7: VKZ-Feld setzen (optional, wenn Projektvariable vorhanden)
+        # ------------------------------------------------------------------
+        vkz = get_variable("verfahrensnummer") or ""
+        _fuege_vkz_feld_hinzu(enriched_layer, vkz)
+
+        info(
+            "VG laden",
+            f"Schritt 6/6: Fertig. {erfolgreich} Flurstücke verarbeitet, "
+            f"{fehlgeschlagen} fehlgeschlagen.",
+            duration=5,
+        )
+
+        return enriched_layer, flst_layer, vg_status
+
+    finally:
+        # Hilfs-Layer (ALKIS-VG) stets aus dem Projekt entfernen
+        if alkis_vg_layer is not None:
+            _entferne_hilfs_layer(project, VERF_LAYER_NAME)
+            # Auch nach Layername des ALKIS-Temp-Layers bereinigen
+            _entferne_hilfs_layer(project, "TempVerfahrensgebiet")
+            try:
+                project.removeMapLayer(alkis_vg_layer.id())
+            except Exception:
+                pass

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,360 @@
+"""sn_verfahrensgebiet/tests/test_verfahrensgebiet_alkis_komplett.py
+
+Unit-Tests für :mod:`sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett`.
+
+Getestete Szenarien:
+
+Erfolgsfall:
+- Vollständiger Workflow → enriched Layer + flst_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)
+- flurstueck_alkis gibt LoadStatus.NONE → (None, None, NONE)
+- Keine Flurstücke selektiert → (None, flst_layer, NONE) + Warnung
+- erstelle_verfahrensgebiet_aus_auswahl liefert ungültigen Layer → (None, flst_layer, NONE)
+- Cleanup: ALKIS-VG-Hilfs-Layer wird in allen Fällen aus Projekt entfernt
+
+Hilfsfunktionen:
+- _select_by_location: processing-Fallback liefert 0 → keine Exception
+- _fuege_vkz_feld_hinzu: leeres VKZ wird ignoriert
+"""
+
+import sys
+import types
+import unittest
+from unittest.mock import patch, MagicMock, call
+
+# ---------------------------------------------------------------------------
+# 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,
+)
+
+
+# ---------------------------------------------------------------------------
+# 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):
+    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
+    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
+
+
+# ---------------------------------------------------------------------------
+# Tests: _select_by_location
+# ---------------------------------------------------------------------------
+
+class TestSelectByLocation(unittest.TestCase):
+
+    def test_processing_not_available_returns_zero(self):
+        """Wenn processing nicht importierbar ist → 0 ohne Exception."""
+        flst = MagicMock()
+        vg = 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):
+        """processing.run erfolgreich → selectedFeatureCount des Layers."""
+        flst = MagicMock()
+        flst.selectedFeatureCount.return_value = 7
+        vg = MagicMock()
+
+        mock_processing = MagicMock()
+        with patch.dict("sys.modules", {"processing": mock_processing}):
+            result = _select_by_location(flst, vg)
+
+        self.assertEqual(result, 7)
+
+    def test_processing_exception_returns_zero(self):
+        """Exception in processing.run → 0 ohne Propagation."""
+        flst = MagicMock()
+        vg = MagicMock()
+
+        mock_processing = MagicMock()
+        mock_processing.run.side_effect = RuntimeError("Kein QGIS")
+        with patch.dict("sys.modules", {"processing": mock_processing}):
+            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):
+        """Leeres VKZ → kein Zugriff auf Layer."""
+        layer = MagicMock()
+        _fuege_vkz_feld_hinzu(layer, "")
+        layer.dataProvider.assert_not_called()
+
+    def test_none_vkz_does_nothing(self):
+        """None-VKZ → kein Zugriff auf Layer."""
+        layer = MagicMock()
+        _fuege_vkz_feld_hinzu(layer, None)
+        layer.dataProvider.assert_not_called()
+
+    def test_vkz_field_added(self):
+        """Nicht-leeres VKZ → Provider.addAttributes wird aufgerufen."""
+        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: verfahrensgebiet_alkis_komplett (Hauptworkflow)
+# ---------------------------------------------------------------------------
+
+_PATCH_VG_ALKIS = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.verfahrensgebiet_alkis"
+_PATCH_GEM_BY_BBOX = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.gemarkung_schluessel_und_namen_by_bbox"
+_PATCH_FLST = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.flurstueck_alkis"
+_PATCH_SELECT = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett._select_by_location"
+_PATCH_ERSTELLE = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.erstelle_verfahrensgebiet_aus_auswahl"
+_PATCH_VKZ = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett._fuege_vkz_feld_hinzu"
+_PATCH_PROJECT = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.QgsProject"
+_PATCH_MSGBOX = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.QMessageBox"
+_PATCH_INFO = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.info"
+_PATCH_WARNING = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.warning"
+_PATCH_GET_VAR = "sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.get_variable"
+
+
+class TestVerfahrensgebietAlkisKomplett(unittest.TestCase):
+
+    def _make_project(self):
+        proj = _DummyProject()
+        mock = MagicMock()
+        mock.instance.return_value = proj
+        # Leere mapLayersByName für Cleanup
+        proj._layers_by_name = {"Verfahrensgebiet": [], "TempVerfahrensgebiet": []}
+        return mock, proj
+
+    # ------------------------------------------------------------------ #
+    # 1. VG-ALKIS liefert NONE → sofortiger Abbruch                       #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_VG_ALKIS)
+    def test_vg_alkis_none_aborts(self, mock_vg, mock_proj):
+        mock_proj.instance.return_value = _DummyProject()
+        mock_vg.return_value = (None, LoadStatus.NONE)
+
+        enriched, flst, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIsNone(flst)
+        self.assertEqual(status, LoadStatus.NONE)
+
+    # ------------------------------------------------------------------ #
+    # 2. flurstueck_alkis liefert NONE → Abbruch nach Gemarkungsabfrage   #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_GEM_BY_BBOX)
+    @patch(_PATCH_VG_ALKIS)
+    def test_flst_none_aborts(self, mock_vg, mock_gem, mock_flst, mock_proj):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer = _make_layer(name="Verfahrensgebiet")
+        mock_vg.return_value = (vg_layer, LoadStatus.FIRST)
+        mock_gem.return_value = [("1234", "Muster")]
+        mock_flst.return_value = (None, LoadStatus.NONE)
+
+        enriched, flst, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIsNone(flst)
+        self.assertEqual(status, LoadStatus.NONE)
+
+    # ------------------------------------------------------------------ #
+    # 3. Keine Selektion → Warnung + (None, flst_layer, NONE)             #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_GEM_BY_BBOX)
+    @patch(_PATCH_VG_ALKIS)
+    def test_no_selection_returns_none_enriched(
+        self, mock_vg, mock_gem, mock_flst, mock_proj, mock_select, mock_msgbox
+    ):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer = _make_layer(name="Verfahrensgebiet")
+        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  # keine Selektion
+
+        enriched, flst, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIs(flst, flst_layer)
+        self.assertEqual(status, LoadStatus.NONE)
+        mock_msgbox.warning.assert_called_once()
+
+    # ------------------------------------------------------------------ #
+    # 4. Vollständiger Erfolgsfall                                         #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_GEM_BY_BBOX)
+    @patch(_PATCH_VG_ALKIS)
+    def test_success_full_workflow(
+        self,
+        mock_vg, mock_gem, mock_flst, mock_proj,
+        mock_select, mock_erstelle, mock_get_var, mock_vkz_fn,
+    ):
+        """Vollständiger Workflow → enriched Layer + flst Layer + FIRST."""
+        proj_instance = _DummyProject()
+        mock_proj.instance.return_value = proj_instance
+
+        vg_layer = _make_layer(name="Verfahrensgebiet")
+        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"
+
+        result_enriched, result_flst, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIs(result_enriched, enriched)
+        self.assertIs(result_flst, flst_layer)
+        self.assertEqual(status, LoadStatus.FIRST)
+        mock_vkz_fn.assert_called_once_with(enriched, "VF-2024-001")
+
+    # ------------------------------------------------------------------ #
+    # 5. Kein VKZ → _fuege_vkz_feld_hinzu mit leerem String              #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_VKZ)
+    @patch(_PATCH_GET_VAR)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_GEM_BY_BBOX)
+    @patch(_PATCH_VG_ALKIS)
+    def test_no_vkz_variable_passes_empty_string(
+        self,
+        mock_vg, mock_gem, mock_flst, mock_proj,
+        mock_select, mock_erstelle, mock_get_var, mock_vkz_fn,
+    ):
+        proj_instance = _DummyProject()
+        mock_proj.instance.return_value = proj_instance
+
+        vg_layer = _make_layer(name="Verfahrensgebiet")
+        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 = ""  # keine Projektvariable
+
+        verfahrensgebiet_alkis_komplett(None)
+
+        mock_vkz_fn.assert_called_once_with(enriched, "")
+
+    # ------------------------------------------------------------------ #
+    # 6. erstelle_verfahrensgebiet liefert ungültigen Layer               #
+    # ------------------------------------------------------------------ #
+    @patch(_PATCH_MSGBOX)
+    @patch(_PATCH_ERSTELLE)
+    @patch(_PATCH_SELECT)
+    @patch(_PATCH_PROJECT)
+    @patch(_PATCH_FLST)
+    @patch(_PATCH_GEM_BY_BBOX)
+    @patch(_PATCH_VG_ALKIS)
+    def test_invalid_enriched_layer_returns_none(
+        self, mock_vg, mock_gem, mock_flst, mock_proj, mock_select, mock_erstelle, mock_msgbox
+    ):
+        mock_proj.instance.return_value = _DummyProject()
+        vg_layer = _make_layer()
+        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)
+
+        enriched, flst, status = verfahrensgebiet_alkis_komplett(None)
+
+        self.assertIsNone(enriched)
+        self.assertIs(flst, flst_layer)
+        self.assertEqual(status, LoadStatus.NONE)
+        mock_msgbox.critical.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_working_tab.py

@@ -83,11 +83,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_verf = types.SimpleNamespace(setText=lambda _t: None, setStyleSheet=lambda _s: None)
-        tab.haken_flurst = object()
+        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 +128,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 → 3-Tupel (enriched, flst, FIRST)."""
         tab = self._make_tab()
-        layer = _DummyLayer(valid=True, layer_id="layer_ok")
+        enriched = _DummyLayer(valid=True, layer_id="enriched_ok")
-        mock_alkis.return_value = (layer, LoadStatus.FIRST)
+        flst = _DummyLayer(valid=True, layer_id="flst_ok")
-        mock_project_instance.return_value = _DummyProject()
+        mock_alkis_komplett.return_value = (enriched, flst, 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, 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,43 @@ 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).
+
+        Delegiert an :func:`~sn_verfahrensgebiet.functions.verfahrensgebiet_alkis_komplett.verfahrensgebiet_alkis_komplett`.
+        Nach Erfolg werden enriched VG-Layer und Flurstücks-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)
+            enriched_layer, flst_layer, status = verfahrensgebiet_alkis_komplett(self)
-            if not layer or not layer.isValid():
+
+            if status == LoadStatus.NONE or not enriched_layer:
                 set_verfahrensgebiet_layer_id("")
                 return
 
-            if status in (LoadStatus.FIRST, LoadStatus.RELOAD):
+            project = QgsProject.instance()
-                QgsProject.instance().addMapLayer(layer)
-                iface.mapCanvas().setExtent(layer.extent())
 
-                apply_style(layer, "verfahrensgebiet.qml")
+            # Flurstücks-Layer ins Projekt (sofern neu geladen)
-                iface.mapCanvas().refresh()
+            if flst_layer and flst_layer.isValid():
-                self.setze_haken(self.haken_verf, True)
+                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)
+            # Enriched VG-Layer ins Projekt
-                self._on_verf_layer_changed(layer)
+            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)
 
-                msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen"
+            self.verf_layer_combo.setLayer(enriched_layer)
-                success("Verfahrensgebiet", f"Layer wurde {msg}.", duration=5)
+            self._on_verf_layer_changed(enriched_layer)
 
-            elif status == LoadStatus.KEEP:
+            msg = "erstmalig geladen" if status == LoadStatus.FIRST else "neu geladen"
-                self.verf_layer_combo.setLayer(layer)
+            success("Verfahrensgebiet", f"Layer wurde {msg} (inkl. Flurstücksverschneidung).", duration=5)
-                self._on_verf_layer_changed(layer)
-                info("Verfahrensgebiet", "Vorhandener Layer wurde behalten.", duration=4)
-
-            elif status == LoadStatus.NONE:
-                self._on_verf_layer_changed(None)
-                warning("Verfahrensgebiet", "Layer konnte nicht geladen werden.", duration=None)
 
         finally:
             self._set_busy(self.btn_verf_alkis, False, "Aus ALKIS laden")
@@ -368,3 +394,39 @@ 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)