tools/headless_ogc_test.py: PyQGIS-Headless-Test gegen die OGC API (Laden, Anlegen, Aendern, Loeschen)
Mit QGIS 4.2.0 am 07.09.2026 gemessen: 0 Fehler. README nennt die Server-Anforderungen (service-desc/OpenAPI, Part-5-Schema, OPTIONS Allow, Merge Patch). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Erik
@@ -169,6 +169,15 @@ CRS-URIs: http://www.opengis.net/def/crs/OGC/1.3/CRS84 (Vorgabe),
|
||||
http://www.opengis.net/def/crs/EPSG/0/25833
|
||||
```
|
||||
|
||||
### Verifiziert mit QGIS 4.2 (headless)
|
||||
|
||||
[tools/headless_ogc_test.py](tools/headless_ogc_test.py) treibt das installierte QGIS ohne
|
||||
Oberfläche (PyQGIS, `QT_QPA_PLATFORM=offscreen`) gegen die API: Layer laden, Feature anlegen,
|
||||
Attribut und Geometrie ändern, löschen — für einen Verfahrens-Layer und ein Maßnahmenblatt,
|
||||
jeder Schritt per API gegengeprüft. Stand 07.09.2026: **0 Fehler** mit QGIS 4.2.0. Was QGIS
|
||||
dafür vom Server braucht (und die API liefert): `service-desc`-Link auf ein OpenAPI-Dokument,
|
||||
Part-5-Schema für Felder/Geometrietyp, `OPTIONS …/items` mit `Allow`, PATCH als JSON Merge Patch.
|
||||
|
||||
## Das Beispiel-Plugin (klassischer GeoJSON-Weg)
|
||||
|
||||
Das Plugin macht den kompletten Arbeitsablauf in der QGIS-Oberfläche
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Headless-Test (PyQGIS, ohne GUI): QGIS gegen die OGC API Features des VLN Managers.
|
||||
|
||||
Laedt je eine Familien-Sammlung (Verfahrens-Layer `st-<vkz>-punkt`, Massnahmenblatt
|
||||
`massnahmeblatt-<vkz>-linie`), legt ein Feature an, aendert Attribut und Geometrie, loescht es —
|
||||
jeder Schritt per API gegengeprueft. Genau das, was QGIS in der Oberflaeche beim
|
||||
„Layer-Aenderungen speichern" tut (OGC API Features Part 4).
|
||||
|
||||
Aufruf (macOS, QGIS-Bundle-Python; Token = userauth aus POST /person/login oder ein vlnk_-Key):
|
||||
|
||||
APP=/Applications/QGIS-final-4_2_0.app; LIB="$APP/Contents/Resources/python3.12"
|
||||
VLN_TOKEN=… VLN_VKZ=27998 QT_QPA_PLATFORM=offscreen PROJ_DATA="$APP/Contents/Resources/qgis/proj" \
|
||||
QGIS_AUTH_DB_DIR_PATH="$(mktemp -d)" PYTHONPATH="$LIB:$LIB/lib-dynload:$LIB/site-packages" \
|
||||
"$APP/Contents/MacOS/python3.12" tools/headless_ogc_test.py
|
||||
|
||||
Nur gegen ein TEST-Verfahren laufen lassen: der Test schreibt echte Features (und raeumt sie
|
||||
wieder weg; ein neu angelegtes Massnahmenblatt bleibt als leere Huelle — UUID wird ausgegeben).
|
||||
Gemessen am 07.09.2026 mit QGIS 4.2.0: 0 Fehler.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
|
||||
from qgis.core import (QgsApplication, QgsAuthMethodConfig, QgsDataSourceUri, QgsFeature,
|
||||
QgsGeometry, QgsPointXY, QgsVectorDataProvider, QgsVectorLayer)
|
||||
|
||||
BASE = os.environ.get("VLN_OGC", "https://api.flurneuordnung-sachsen.de/v2/ogc")
|
||||
TOKEN = os.environ["VLN_TOKEN"]
|
||||
VKZ = os.environ.get("VLN_VKZ", "27998")
|
||||
FEHLER = []
|
||||
|
||||
|
||||
def api(pfad, crs=None):
|
||||
url = BASE + pfad + (("&" if "?" in pfad else "?") + "crs=" + crs if crs else "")
|
||||
req = urllib.request.Request(url, headers={"X-API-Key": TOKEN})
|
||||
with urllib.request.urlopen(req, timeout=30) as r:
|
||||
return json.loads(r.read().decode("utf-8"))
|
||||
|
||||
|
||||
def pruefe(bedingung, text):
|
||||
print((" OK " if bedingung else " FEHL ") + text)
|
||||
if not bedingung:
|
||||
FEHLER.append(text)
|
||||
|
||||
|
||||
def anzahl(cid):
|
||||
return api("/collections/%s/items?limit=1" % cid)["numberMatched"]
|
||||
|
||||
|
||||
def layer_fuer(cid, authcfg, srs=None):
|
||||
uri = QgsDataSourceUri()
|
||||
uri.setParam("url", BASE)
|
||||
uri.setParam("typename", cid)
|
||||
uri.setParam("pagingEnabled", "true")
|
||||
uri.setParam("pageSize", "1000")
|
||||
if srs:
|
||||
uri.setParam("srsname", srs)
|
||||
uri.setAuthConfigId(authcfg)
|
||||
return QgsVectorLayer(uri.uri(False), cid, "OAPIF")
|
||||
|
||||
|
||||
def commit(layer, schritt):
|
||||
ok = layer.commitChanges()
|
||||
pruefe(ok, "%s: commit (%s)" % (schritt, "; ".join(layer.commitErrors())[:200] if not ok else "ok"))
|
||||
layer.reload()
|
||||
return ok
|
||||
|
||||
|
||||
def bahn_layer(authcfg):
|
||||
cid = "st-" + VKZ + "-punkt"
|
||||
print("== %s ==" % cid)
|
||||
vorher = anzahl(cid)
|
||||
lyr = layer_fuer(cid, authcfg, "EPSG:25833")
|
||||
pruefe(lyr.isValid(), "Layer gueltig (%s)" % (lyr.error().message() if not lyr.isValid() else lyr.crs().authid()))
|
||||
if not lyr.isValid():
|
||||
return
|
||||
caps = lyr.dataProvider().capabilities()
|
||||
pruefe(bool(caps & QgsVectorDataProvider.AddFeatures), "Capability AddFeatures")
|
||||
pruefe(bool(caps & QgsVectorDataProvider.ChangeGeometries), "Capability ChangeGeometries")
|
||||
pruefe(bool(caps & QgsVectorDataProvider.ChangeAttributeValues), "Capability ChangeAttributeValues")
|
||||
pruefe(bool(caps & QgsVectorDataProvider.DeleteFeatures), "Capability DeleteFeatures")
|
||||
print(" Felder:", [f.name() for f in lyr.fields()][:10], "| CRS:", lyr.crs().authid(), "| Features:", lyr.featureCount())
|
||||
|
||||
lyr.startEditing()
|
||||
f = QgsFeature(lyr.fields())
|
||||
f.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(411500.0, 5671500.0)))
|
||||
for name, wert in (("typ", "qgis-test"), ("name", "QGIS headless")):
|
||||
idx = lyr.fields().indexOf(name)
|
||||
if idx >= 0:
|
||||
f.setAttribute(idx, wert)
|
||||
pruefe(lyr.addFeature(f), "addFeature")
|
||||
if not commit(lyr, "anlegen"):
|
||||
return
|
||||
pruefe(anzahl(cid) == vorher + 1, "API zaehlt +1 (%d -> %d)" % (vorher, anzahl(cid)))
|
||||
neu = [x for x in lyr.getFeatures() if x["typ"] == "qgis-test"] if lyr.fields().indexOf("typ") >= 0 else []
|
||||
pruefe(len(neu) == 1, "neues Feature nach reload sichtbar (%d)" % len(neu))
|
||||
if not neu:
|
||||
return
|
||||
fid = neu[0].id()
|
||||
feats = api("/collections/%s/items?limit=100" % cid, "http://www.opengis.net/def/crs/EPSG/0/25833")["features"]
|
||||
mein = [x for x in feats if x["properties"].get("typ") == "qgis-test"]
|
||||
pruefe(len(mein) == 1 and abs(mein[0]["geometry"]["coordinates"][0] - 411500.0) < 0.5,
|
||||
"API-Geometrie in 25833 stimmt (%s)" % (str(mein[0]["geometry"]["coordinates"])[:40] if mein else "-"))
|
||||
api_id = mein[0]["id"] if mein else None
|
||||
|
||||
lyr.startEditing()
|
||||
idx = lyr.fields().indexOf("name")
|
||||
if idx >= 0:
|
||||
pruefe(lyr.changeAttributeValue(fid, idx, "QGIS geaendert"), "changeAttributeValue")
|
||||
else:
|
||||
print(" (Feld name fehlt im Schema — Attributaenderung uebersprungen)")
|
||||
commit(lyr, "attribut")
|
||||
if api_id is not None:
|
||||
f_api = api("/collections/%s/items/%s" % (cid, api_id))
|
||||
pruefe(f_api["properties"].get("name") == "QGIS geaendert", "API: name geaendert (%s)" % f_api["properties"].get("name"))
|
||||
|
||||
lyr.startEditing()
|
||||
pruefe(lyr.changeGeometry(fid, QgsGeometry.fromPointXY(QgsPointXY(412000.0, 5672000.0))), "changeGeometry")
|
||||
commit(lyr, "geometrie")
|
||||
if api_id is not None:
|
||||
f_api = api("/collections/%s/items/%s" % (cid, api_id), "http://www.opengis.net/def/crs/EPSG/0/25833")
|
||||
pruefe(abs(f_api["geometry"]["coordinates"][0] - 412000.0) < 0.5, "API: Geometrie verschoben (%s)" % str(f_api["geometry"]["coordinates"])[:40])
|
||||
|
||||
lyr.startEditing()
|
||||
pruefe(lyr.deleteFeature(fid), "deleteFeature")
|
||||
commit(lyr, "loeschen")
|
||||
pruefe(anzahl(cid) == vorher, "API zaehlt wieder %d" % vorher)
|
||||
|
||||
|
||||
def bahn_blatt(authcfg):
|
||||
cid = "massnahmeblatt-" + VKZ + "-linie"
|
||||
print("== %s ==" % cid)
|
||||
vorher = anzahl(cid)
|
||||
lyr = layer_fuer(cid, authcfg)
|
||||
pruefe(lyr.isValid(), "Layer gueltig (%s)" % (lyr.error().message() if not lyr.isValid() else lyr.crs().authid()))
|
||||
if not lyr.isValid():
|
||||
return
|
||||
print(" Felder:", [f.name() for f in lyr.fields()][:12], "| CRS:", lyr.crs().authid())
|
||||
lyr.startEditing()
|
||||
f = QgsFeature(lyr.fields())
|
||||
f.setGeometry(QgsGeometry.fromPolylineXY([QgsPointXY(13.736, 51.185), QgsPointXY(13.738, 51.186)]))
|
||||
werte = {"blatt_typ": "wegebau", "blatt_name": "QGIS-Testblatt", "blatt_mkz": "116 98-1"}
|
||||
for name, wert in werte.items():
|
||||
idx = lyr.fields().indexOf(name)
|
||||
pruefe(idx >= 0, "Feld %s vorhanden" % name)
|
||||
if idx >= 0:
|
||||
f.setAttribute(idx, wert)
|
||||
pruefe(lyr.addFeature(f), "addFeature (neues Blatt)")
|
||||
if not commit(lyr, "blatt anlegen"):
|
||||
return
|
||||
pruefe(anzahl(cid) == vorher + 1, "API zaehlt +1 (%d -> %d)" % (vorher, anzahl(cid)))
|
||||
feats = api("/collections/%s/items?limit=100" % cid)["features"]
|
||||
mein = [x for x in feats if x["properties"].get("blatt_name") == "QGIS-Testblatt"]
|
||||
pruefe(len(mein) == 1, "Blatt per API sichtbar (%d)" % len(mein))
|
||||
if not mein:
|
||||
return
|
||||
api_id = mein[0]["id"]
|
||||
neu = [x for x in lyr.getFeatures() if x["blatt_name"] == "QGIS-Testblatt"]
|
||||
pruefe(len(neu) == 1, "Feature nach reload in QGIS (%d)" % len(neu))
|
||||
if neu:
|
||||
fid = neu[0].id()
|
||||
lyr.startEditing()
|
||||
pruefe(lyr.changeAttributeValue(fid, lyr.fields().indexOf("blatt_name"), "QGIS-Testblatt 2"), "changeAttributeValue blatt_name")
|
||||
commit(lyr, "blatt attribut")
|
||||
f_api = api("/collections/%s/items/%s" % (cid, api_id))
|
||||
pruefe(f_api["properties"].get("blatt_name") == "QGIS-Testblatt 2", "API: blatt_name geaendert (%s)" % f_api["properties"].get("blatt_name"))
|
||||
lyr.startEditing()
|
||||
pruefe(lyr.deleteFeature(fid), "deleteFeature (Lage-Geometrie)")
|
||||
commit(lyr, "blatt geometrie loeschen")
|
||||
pruefe(anzahl(cid) == vorher, "API zaehlt wieder %d" % vorher)
|
||||
# Das Blatt selbst (ohne Geometrien) bleibt in MASSNAHMEBLATT — der Aufrufer raeumt es weg.
|
||||
print(" Blatt-UUID (aufraeumen):", api_id.split(":")[0])
|
||||
with open(os.environ.get("BLATT_UUID_DATEI", "/tmp/qgis_blatt_uuid.txt"), "w") as fh:
|
||||
fh.write(api_id.split(":")[0])
|
||||
|
||||
|
||||
app = QgsApplication([], False)
|
||||
QgsApplication.setPrefixPath(os.environ.get("QGIS_APP", "/Applications/QGIS-final-4_2_0.app") + "/Contents/MacOS", True)
|
||||
QgsApplication.setPluginPath(os.environ.get("QGIS_APP", "/Applications/QGIS-final-4_2_0.app") + "/Contents/PlugIns/qgis")
|
||||
app.initQgis()
|
||||
# Headless: kein System-Proxy (die PAC-Datei des Macs bricht Qt-Requests ab), Provider-Meldungen sichtbar.
|
||||
from qgis.PyQt.QtNetwork import QNetworkProxyFactory
|
||||
QNetworkProxyFactory.setUseSystemConfiguration(False)
|
||||
QgsApplication.messageLog().messageReceived.connect(lambda msg, tag, level: print(" [log %s] %s" % (tag, str(msg)[:300])))
|
||||
am = QgsApplication.authManager()
|
||||
if not am.setMasterPassword("headless-test", True):
|
||||
print("FEHL Auth-Master"); sys.exit(2)
|
||||
cfg = QgsAuthMethodConfig("APIHeader")
|
||||
cfg.setName("vln-test")
|
||||
cfg.setConfigMap({"X-API-Key": TOKEN})
|
||||
ok = am.storeAuthenticationConfig(cfg, True)
|
||||
if not (ok[0] if isinstance(ok, tuple) else ok):
|
||||
print("FEHL Authcfg"); sys.exit(2)
|
||||
print("Authcfg:", cfg.id(), "| QGIS", QgsApplication.applicationVersion() if hasattr(QgsApplication, "applicationVersion") else "")
|
||||
try:
|
||||
bahn_layer(cfg.id())
|
||||
bahn_blatt(cfg.id())
|
||||
finally:
|
||||
app.exitQgis()
|
||||
print("== Ergebnis: %d Fehler ==" % len(FEHLER))
|
||||
for f in FEHLER:
|
||||
print(" -", f)
|
||||
sys.exit(1 if FEHLER else 0)
|
||||
Reference in New Issue
Block a user