API_Karte_QGISDemo/tools/headless_dienste_test.py

# -*- coding: utf-8 -*-
"""Die Linkliste im Plugin: holen, Layer bauen, Stil anwenden — gegen die echte Schnittstelle.

Aufruf (macOS, QGIS 4.2 im Bundle; VLN_TOKEN = API-Key aus dem Login):

    APP=/Applications/QGIS-final-4_2_0.app; LIB="$APP/Contents/Resources/python3.12"
    QT_QPA_PLATFORM=offscreen PROJ_DATA="$APP/Contents/Resources/qgis/proj" \\
    PYTHONPATH="$LIB:$LIB/lib-dynload:$LIB/site-packages" \\
    "$APP/Contents/MacOS/python3.12" tools/headless_dienste_test.py
"""
import os, sys, tempfile, time
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from qgis.core import QgsApplication, QgsNetworkAccessManager, QgsProject

app = QgsApplication([], True)
QgsApplication.setPrefixPath("/Applications/QGIS-final-4_2_0.app/Contents/MacOS", True)
QgsApplication.setPluginPath("/Applications/QGIS-final-4_2_0.app/Contents/PlugIns/qgis")
app.initQgis()
from qgis.PyQt.QtNetwork import QNetworkProxyFactory, QNetworkDiskCache
QNetworkProxyFactory.setUseSystemConfiguration(False)
c = QNetworkDiskCache(); c.setCacheDirectory(tempfile.mkdtemp())
QgsNetworkAccessManager.instance().setCache(c)

from vln_karten import linkliste as LL
from vln_karten.api_client import KartenApiClient

BASE, KEY, VKZ = "https://api.flurneuordnung-sachsen.de/v2", os.environ["VLN_TOKEN"], "27027"
FEHLER = []


def pruefe(b, t):
    print(("  OK   " if b else "  FEHL ") + t)
    if not b:
        FEHLER.append(t)


client = KartenApiClient(base_url=BASE, api_key=KEY)
print("== 1) Liste holen ==")
t = time.time()
dienste = LL.dienste_laden(client, VKZ)
pruefe(len(dienste) > 0, "%d Dienste in %.2f s" % (len(dienste), time.time() - t))
pruefe(all({"name", "provider", "url", "herkunft"} <= set(d) for d in dienste),
       "alle Felder vorhanden, die der Layerbau braucht")

print("== 2) Layer bauen, je Anbieter einer ==")
gebaut = {}
for d in dienste:
    p = d["provider"]
    if p in gebaut:
        continue
    t = time.time()
    lyr = LL.layer_bauen(d)
    ok = lyr is not None and lyr.isValid()
    gebaut[p] = ok
    pruefe(ok, "%-5s %-34s %s (%.1f s)" % (p, d["name"][:34],
                                           "gueltig" if ok else (lyr.error().message()[:60] if lyr else "kein Layer"),
                                           time.time() - t))
    if ok and hasattr(lyr, "featureCount"):
        print("        Features:", lyr.featureCount())

print("== 3) Stil aus der Datenbank anwenden ==")
mit_stil = [d for d in dienste if d.get("stil")]
if mit_stil:
    d = mit_stil[0]
    lyr = LL.layer_bauen(d)
    if lyr is not None and lyr.isValid():
        vorher = lyr.renderer().type() if hasattr(lyr, "renderer") and lyr.renderer() else "-"
        ok, meldung = LL.stil_anwenden(lyr, BASE, d["id"], KEY)
        nachher = lyr.renderer().type() if hasattr(lyr, "renderer") and lyr.renderer() else "-"
        pruefe(ok, "Stil fuer '%s': %s (Renderer %s -> %s)" % (d["name"], meldung or "angewendet",
                                                               vorher, nachher))
    else:
        print("   (Layer nicht gueltig, Stil nicht pruefbar)")
else:
    print("   (kein Eintrag mit Stil)")

print("== 4) ohne Stil meldet sauber ==")
ohne = [d for d in dienste if not d.get("stil")]
if ohne:
    lyr = LL.layer_bauen(ohne[0])
    if lyr is not None and lyr.isValid():
        ok, meldung = LL.stil_anwenden(lyr, BASE, ohne[0]["id"], KEY)
        pruefe(not ok and "kein Stil" in meldung, "Meldung statt Fehler: %s" % meldung)

# ---------------------------------------------------------------------------------------
# Ab hier wird geschrieben: eigene Eintraege anlegen, die Vererbung pruefen, wieder aufraeumen.
# Das Plugin selbst schreibt nichts in die Linkliste — deshalb laufen die Pflegeaufrufe hier
# ueber urllib und nicht ueber den KartenApiClient (der kennt nur GET/POST/PUT).
# ---------------------------------------------------------------------------------------
import json as _json
import urllib.error
import urllib.request


def api(pfad, art="GET", rumpf=None):
    """Ein Pflegeaufruf gegen /v2/linkliste. Gibt (status, daten) zurueck."""
    req = urllib.request.Request(BASE + "/linkliste" + pfad, method=art,
                                 data=_json.dumps(rumpf).encode("utf-8") if rumpf is not None else None,
                                 headers={"X-API-Key": KEY, "Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=20) as a:
            roh = a.read()
            return a.status, (_json.loads(roh) if roh else {})
    except urllib.error.HTTPError as e:
        return e.code, {}


def aufraeumen(ids):
    for i in ids:
        api("/%s" % i, "DELETE")


print("== 5) WMS und WFS bauen (die Altdaten sind alle ArcGIS) ==")
PROBEN = [
    {"gruppe": "hintergrund", "name": "ZZ Probe WMS", "provider": "wms",
     "url": "https://geodienste.sachsen.de/wms_geosn_dop-rgb/guest?", "layer": "sn_dop_020"},
    {"gruppe": "sonstiges", "name": "ZZ Probe WFS", "provider": "wfs",
     "url": "https://sgx.geodatenzentrum.de/wfs_vg2500", "layer": "vg2500:vg2500_krs"},
]
angelegt = []
try:
    for probe in PROBEN:
        st, d = api("", "POST", dict(probe, stufe="land"))
        if st != 201:
            pruefe(False, "%s anlegen: HTTP %s" % (probe["name"], st))
            continue
        angelegt.append(d["data"]["id"])
    liste = LL.dienste_laden(client, VKZ)
    for probe in PROBEN:
        eintrag = next((x for x in liste if x["name"] == probe["name"]), None)
        if eintrag is None:
            pruefe(False, "%s steht nicht in der Liste" % probe["name"])
            continue
        t = time.time()
        lyr = LL.layer_bauen(eintrag)
        ok = lyr is not None and lyr.isValid()
        pruefe(ok, "%-4s %-16s %s (%.1f s)"
               % (probe["provider"], probe["name"][-12:],
                  "gueltig" if ok else (lyr.error().message()[:70] if lyr else "kein Layer"),
                  time.time() - t))
        if ok and probe["provider"] == "wfs" and hasattr(lyr, "featureCount"):
            print("        Features:", lyr.featureCount())

    print("== 6) Vererbung: Ausblenden wirkt, Zuruecknehmen holt zurueck ==")
    ziel = angelegt[0] if angelegt else None
    bereich = {"stufe": "tg", "vkz": int(VKZ), "team_id": None}
    if ziel:
        st, _ = api("/%s/abweichung" % ziel, "POST",
                    {"bereich": bereich, "art": "aus", "felder": {}})
        pruefe(st in (200, 201), "ausblenden fuer das Verfahren: HTTP %s" % st)
        namen = [x["name"] for x in LL.dienste_laden(client, VKZ)]
        pruefe("ZZ Probe WMS" not in namen,
               "ausgeblendeter Dienst kommt nicht mehr im Plugin an (%d Dienste)" % len(namen))
        st, _ = api("/%s/abweichung?stufe=tg&vkz=%s&team_id=" % (ziel, VKZ), "DELETE")
        pruefe(st == 204, "zuruecknehmen: HTTP %s" % st)
        namen = [x["name"] for x in LL.dienste_laden(client, VKZ)]
        pruefe("ZZ Probe WMS" in namen, "nach dem Zuruecknehmen wieder da (%d Dienste)" % len(namen))

    print("== 7) Stil setzen und wieder entfernen ==")
    if ziel:
        st, _ = api("/%s" % ziel, "PATCH",
                    dict(PROBEN[0], stufe="land",
                         qml="<qgis version=\"3.34\"><renderer-v2 type=\"singleSymbol\"/></qgis>"))
        pruefe(st == 200, "Stil ueber PATCH setzen: HTTP %s" % st)
        eintrag = next((x for x in LL.dienste_laden(client, VKZ) if x["name"] == "ZZ Probe WMS"), None)
        pruefe(bool(eintrag and eintrag.get("stil")), "die Liste meldet den Stil")

        # Aendern ohne QML darf den Stil NICHT loeschen — sonst verliert jede Korrektur den Stil.
        st, _ = api("/%s" % ziel, "PATCH", dict(PROBEN[0], stufe="land", beschreibung="ohne QML"))
        eintrag = next((x for x in LL.dienste_laden(client, VKZ) if x["name"] == "ZZ Probe WMS"), None)
        pruefe(bool(eintrag and eintrag.get("stil")), "Aendern ohne QML laesst den Stil stehen")

        st, _ = api("/%s/style.qml" % ziel, "DELETE")
        pruefe(st == 204, "Stil entfernen: HTTP %s" % st)
        lyr = LL.layer_bauen(eintrag) if eintrag else None
        if lyr is not None and lyr.isValid():
            ok, meldung = LL.stil_anwenden(lyr, BASE, ziel, KEY)
            pruefe(not ok and "kein Stil" in meldung,
                   "nach dem Entfernen meldet das Plugin sauber: %s" % meldung)
finally:
    aufraeumen(angelegt)
    rest = [x["name"] for x in LL.dienste_laden(client, VKZ) if x["name"].startswith("ZZ ")]
    pruefe(not rest, "aufgeraeumt, keine Testreste" + (" — offen: %s" % rest if rest else ""))

app.exitQgis()
print("== %d Fehler ==" % len(FEHLER))
for f in FEHLER:
    print("  -", f)
sys.exit(1 if FEHLER else 0)