david-spacil/aiscr-qgis-amcr-viewer
1
0 Fork 0
mirror of https://github.com/ARUP-CAS/aiscr-qgis-amcr-viewer.git synced 2026-10-09 20:37:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

ci: denní kontrola API digiarchivu a AMČR OAI

Browse Source 
Plánovaný workflow api_monitor.yml (denně 05:17 UTC + ruční spuštění)
ověřuje kontrakt API, na kterém plugin závisí:

- tests/api_contract.py – stejné dotazy jako plugin, kontrola klíčů,
  typů a tvarů odpovědí (OK / DRIFT / FAIL / UNAVAILABLE)
- tests/api_plugin_live.py – vlastní funkce pluginu (fetch_set,
  load_amcr_data) proti živému API v qgis/qgis:ltr
- tests/api_monitor_report.py – jedno sledovací issue se štítkem
  api-monitor; čistý běh ho zavře, výpadek ho nemění

Neběží na PR, aby výpadek digiarchivu neshodil PR. Popis v AGENTS.md,
OpenSpec změna add-daily-api-monitor.

Připraveno s pomocí AI (Claude), ověřeno proti produkčnímu API.
This commit is contained in:
david-spacil committed 2026-10-02 22:17:01 +02:00
1 parent 7c0401c11b
commit dde76203b2
10 files changed
+2140
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api_plugin_live.py
+401
View File
@@ -0,0 +1,401 @@
# -*- coding: utf-8 -*-
"""
Live plugin test – calls the plugin's own functions against the live AMČR
API inside a real (headless) QGIS and checks they still produce non-empty,
well-formed results. Where the contract test says *what* changed, this test
says *whether users break*.
It is the API-sensitive counterpart of tests/smoke_test.py, which is
deliberately offline.
Run it from the repository root inside the qgis/qgis Docker image
(requests is bundled with QGIS):
docker run --rm -v "$PWD:/work:ro" -w /work \
--user "$(id -u):$(id -g)" -e HOME=/tmp \
-e AMCR_RESULTS_DIR=/tmp/results \
qgis/qgis:ltr python3 tests/api_plugin_live.py
The plugin package is imported as a package (amcr_viewer.amcr_tools), so
its relative imports work – a bare spec_from_file_location would make
load_amcr_data swallow the import error into "0 records".
Status model and outputs match tests/api_contract.py: OK / DRIFT / FAIL /
UNAVAILABLE per check, results-plugin_live.json + a Markdown table on
stdout and in $GITHUB_STEP_SUMMARY, exit 1 on any FAIL or DRIFT. A run
where everything is UNAVAILABLE is green but visible in the summary.
Test area (probe 2026-10-02, anonymous): the same Mikulov bbox as the
contract test, 48.8,16.6,48.9,16.75 – akce 185, lokalita 18,
samostatny_nalez 2, pian 294 records. The fake canvas extent uses it
directly in EPSG:4326, so no coordinate transformation is involved.
Thresholds (design decision 6):
* fetch_set per codelist set: >= 1 item and >= 50 % of that category's
row count in the bundled codelists/heslar.csv (a shrunken codelist is
the #67 symptom)
* load_amcr_data per data type: >= 1 layer with >= 1 feature, valid
geometry and the expected attribute fields
Env overrides (outage simulation; the plugin's own URLs are hard-coded,
so the overrides only steer the availability probe and the codelist
sets, whose URLs live in a module dict). An unreachable host fails fast:
after the full retry cycle of the first probe, later probes to the same
host do no network I/O:
AMCR_DA_URL digiarchiv base URL (default
https://digiarchiv.aiscr.cz)
AMCR_OAI_URL AMCR OAI base URL (default
https://api.aiscr.cz/2.2/oai)
AMCR_TIMEOUT per-request timeout in seconds (default 15)
"""
import csv
import json
import os
import sys
import traceback
# Offscreen, otherwise the widgets would need an X server
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
RESULTS_DIR = os.environ.get("AMCR_RESULTS_DIR", os.getcwd())
DA_URL = os.environ.get("AMCR_DA_URL", "https://digiarchiv.aiscr.cz")
OAI_URL = os.environ.get("AMCR_OAI_URL", "https://api.aiscr.cz/2.2/oai")
TIMEOUT = int(os.environ.get("AMCR_TIMEOUT", "15"))
TEST_BBOX = "48.8,16.6,48.9,16.75" # minLat,minLon,maxLat,maxLon (Mikulov)
BBOX_MIN_LAT, BBOX_MIN_LON, BBOX_MAX_LAT, BBOX_MAX_LON = (
float(x) for x in TEST_BBOX.split(",")
)
RESULTS = []
JOB = "plugin_live"
UNAVAILABLE_DA = False
UNAVAILABLE_OAI = False
def record(name, status, detail):
RESULTS.append({"name": name, "status": status, "detail": detail,
"url": ""})
print(f" {status:<12} {name} – {detail}")
import requests # noqa: E402
# ---------------------------------------------------------------- QGIS setup
from qgis.core import ( # noqa: E402
Qgis,
QgsApplication,
QgsCoordinateReferenceSystem,
QgsProject,
QgsRectangle,
)
print(f"QGIS {Qgis.QGIS_VERSION.split('-')[0]}")
QgsApplication.setPrefixPath(os.environ.get("QGIS_PREFIX_PATH", "/usr"),
True)
qgs = QgsApplication([], True)
qgs.initQgis()
import amcr_viewer.amcr_codelists as codelists # noqa: E402
import amcr_viewer.amcr_tools as tools # noqa: E402
# Point the codelist sets at the overridden base URLs (outage simulation).
# The data-download URL inside load_amcr_data is hard-coded and cannot be
# steered from here; when the probe says digiarchiv is unreachable, the
# download checks are reported UNAVAILABLE without calling the plugin.
if DA_URL != "https://digiarchiv.aiscr.cz" \
or OAI_URL != "https://api.aiscr.cz/2.2/oai":
_new = {}
for key, (base, api_set) in codelists.slovnicek.items():
if "digiarchiv" in base:
base = DA_URL + "/api/search/query"
else:
base = OAI_URL
_new[key] = (base, api_set)
codelists.slovnicek.clear()
codelists.slovnicek.update(_new)
# ------------------------------------------------------------------ fakes
class FakeMessageBar:
"""Collects messageBar() messages so failures can be diagnosed."""
def __init__(self):
self.messages = []
def pushMessage(self, title, text, level=Qgis.MessageLevel.Info):
self.messages.append((title, str(text), level))
class FakeIface:
def __init__(self):
self._bar = FakeMessageBar()
def messageBar(self):
return self._bar
def mapCanvas(self):
return fake_canvas
class FakeMapSettings:
def __init__(self, crs):
self._crs = crs
def destinationCrs(self):
return self._crs
class FakeCanvas:
"""Map canvas whose extent is the test bbox in EPSG:4326."""
def __init__(self):
self._extent = QgsRectangle(
BBOX_MIN_LON, BBOX_MIN_LAT, BBOX_MAX_LON, BBOX_MAX_LAT
)
self._settings = FakeMapSettings(
QgsCoordinateReferenceSystem("EPSG:4326"))
def extent(self):
return self._extent
def mapSettings(self):
return self._settings
fake_canvas = FakeCanvas()
fake_iface = FakeIface()
# amcr_tools does "from qgis.utils import iface", which binds None in a
# headless run – patch the module attribute, not qgis.utils
tools.iface = fake_iface
# ------------------------------------------------------------------ checks
# Circuit breaker (fast outage), the same as in tests/api_contract.py:
# once a host (netloc) is unreachable after full retries, later probes to
# it return False without network I/O, so an all-unreachable run finishes
# in seconds instead of tens of minutes.
DEAD_HOSTS = set()
def probe(url, params=None):
"""Availability probe with retries; True when the API answers.
Once a host is found unreachable after the full retry cycle, it is
added to DEAD_HOSTS and later probes to it fail immediately.
"""
import time
import urllib.parse
netloc = urllib.parse.urlparse(url).netloc
if netloc in DEAD_HOSTS:
return False
for attempt in range(3):
try:
resp = requests.get(url, params=params, timeout=TIMEOUT)
if resp.status_code < 500:
return True
except requests.exceptions.RequestException:
pass
if attempt < 2:
time.sleep((2, 8)[attempt])
DEAD_HOSTS.add(netloc)
return False
def bundled_counts():
"""Row count per category in the bundled codelists/heslar.csv."""
path = os.path.join(ROOT, "amcr_viewer", "codelists", "heslar.csv")
counts = {}
# utf-8-sig: the CSV carries a BOM on purpose (Excel)
with open(path, encoding="utf-8-sig", newline="") as f:
for row in csv.DictReader(f, delimiter=";"):
cat = (row.get("Kategorie") or "").strip()
if cat:
counts[cat] = counts.get(cat, 0) + 1
return counts
def check_translations():
if UNAVAILABLE_DA:
record("load_translations", "UNAVAILABLE",
"digiarchiv unreachable after retries")
return
tools.TRANSLATIONS.clear()
try:
tools.load_translations()
except Exception:
record("load_translations", "FAIL",
traceback.format_exc().rstrip().splitlines()[-1])
return
if tools.TRANSLATIONS:
record("load_translations", "OK",
f"{len(tools.TRANSLATIONS)} keys")
else:
record("load_translations", "FAIL",
"TRANSLATIONS stayed empty after load_translations()")
def check_fetch_set():
"""fetch_set per set in slovnicek against the live API."""
bundled = bundled_counts()
for name, (base_url, api_set) in codelists.slovnicek.items():
unavailable = (UNAVAILABLE_OAI if "digiarchiv" not in base_url
else UNAVAILABLE_DA)
if unavailable:
record(f"fetch_set {name}", "UNAVAILABLE",
"API unreachable after retries")
continue
try:
data = codelists.fetch_set(base_url, name, api_set)
except Exception:
record(f"fetch_set {name}", "FAIL",
traceback.format_exc().rstrip().splitlines()[-1])
continue
if data is None:
record(f"fetch_set {name}", "FAIL", "cancelled (task)")
continue
if not data:
record(f"fetch_set {name}", "FAIL",
f"set {api_set} returned 0 items – the #67 symptom")
continue
expected = bundled.get(name, 0)
if expected and len(data) < expected * 0.5:
record(f"fetch_set {name}", "FAIL",
f"{len(data)} items < 50 % of {expected} bundled rows "
"– a shrunken codelist is the #67 symptom")
else:
record(f"fetch_set {name}", "OK",
f"{len(data)} items"
+ (f" (bundled: {expected})" if expected else ""))
def _layer_specs(typ_dat):
"""Expected attribute fields per data type, from amcr_tools.py."""
common = ["pian", "presnost", "pian_typ", "dj", "typ_dj", typ_dat,
"definicni_body", "odkaz_do_digiarchivu", "okres", "katastr",
"dalsi_katastry", "pristupnost"]
if typ_dat == "akce":
common += ["akce_lokalizace", "vedouci", "organizace",
"specifikace_data", "zahajeni", "ukonceni",
"hlavni_typ", "vedlejsi_typ", "zjisteni",
"nahrazuje_NZ", "projekt"]
elif typ_dat == "lokalita":
common += ["nazev_lokality", "popis_lokality", "typ_lokality",
"druh_lokality", "zachovalost"]
elif typ_dat == "samostatny_nalez":
common = [typ_dat, "definicni_body", "odkaz_do_digiarchivu",
"okres", "katastr", "dalsi_katastry", "projekt",
"nalezce", "datum", "okolnosti", "hloubka_cm",
"lokalizace", "obdobi", "presna_datace", "nalez",
"material", "pocet", "poznamka", "pred_org",
"evidencni", "pristupnost"]
return common
def check_load_amcr_data():
"""load_amcr_data per data type on the test bbox (fake iface/canvas).
The call is synchronous in the main thread (load_amcr_data pumps the
event loop itself, it is not a QgsTask), so a plain call is enough.
"""
for typ_dat in ["akce", "lokalita", "samostatny_nalez"]:
if UNAVAILABLE_DA:
record(f"load_amcr_data {typ_dat}", "UNAVAILABLE",
"digiarchiv unreachable after retries")
continue
# Layers from a previous data type must not mix into the check
project = QgsProject.instance()
project.removeAllMapLayers()
try:
tools.load_amcr_data(fake_canvas, "true", None,
typ_dat=typ_dat, komponenty="false")
except Exception:
record(f"load_amcr_data {typ_dat}", "FAIL",
traceback.format_exc().rstrip().splitlines()[-1])
continue
layers = [lyr for lyr in project.mapLayers().values()
if "amcr_" in lyr.name().lower()]
if not layers:
record(f"load_amcr_data {typ_dat}", "FAIL",
"no AMCR layers were added to the project; messageBar: "
+ "; ".join(m[1] for m in fake_iface._bar.messages[-3:]))
continue
total_features = sum(lyr.featureCount() for lyr in layers)
if total_features < 1:
record(f"load_amcr_data {typ_dat}", "FAIL",
f"{len(layers)} layers but 0 features")
continue
# valid geometry + expected fields on the populated layers
problems = []
expected_fields = _layer_specs(typ_dat)
for layer in layers:
if layer.featureCount() == 0:
continue
fields = {f.name() for f in layer.fields()}
missing = [fl for fl in expected_fields if fl not in fields]
if missing:
problems.append(f"{layer.name()}: missing fields "
f"{missing}")
for feat in layer.getFeatures():
geom = feat.geometry()
if (geom is None or geom.isNull()
or not geom.isGeosValid()):
problems.append(f"{layer.name()}: invalid geometry")
break
if problems:
record(f"load_amcr_data {typ_dat}", "FAIL",
"; ".join(problems))
else:
record(f"load_amcr_data {typ_dat}", "OK",
f"{len(layers)} layers, {total_features} features, "
"fields and geometry valid")
def write_outputs():
os.makedirs(RESULTS_DIR, exist_ok=True)
path = os.path.join(RESULTS_DIR, f"results-{JOB}.json")
with open(path, "w", encoding="utf-8") as f:
json.dump(RESULTS, f, ensure_ascii=False, indent=2)
lines = ["| check | status | detail |", "|---|---|---|"]
for r in RESULTS:
detail = str(r["detail"]).replace("|", "\\|").replace("\n", " ")
lines.append(f"| {r['name']} | {r['status']} | {detail} |")
table = "\n".join(lines)
print()
print(table)
summary = os.environ.get("GITHUB_STEP_SUMMARY")
if summary:
with open(summary, "a", encoding="utf-8") as f:
f.write(table + "\n")
print()
print(f"written: {path}")
bad = [r for r in RESULTS if r["status"] in ("FAIL", "DRIFT")]
print(f"checks: {len(RESULTS)}, FAIL/DRIFT: {len(bad)}")
return 1 if bad else 0
def main():
global UNAVAILABLE_DA, UNAVAILABLE_OAI
print("live plugin test against the production AMČR API")
print(f"test bbox: {TEST_BBOX} (Mikulov)")
UNAVAILABLE_DA = not probe(
DA_URL + "/api/search/query", params={"entity": "akce", "rows": 0})
UNAVAILABLE_OAI = not probe(
OAI_URL, params={"verb": "Identify"})
if UNAVAILABLE_DA:
print("digiarchiv unreachable after retries")
if UNAVAILABLE_OAI:
print("AMČR OAI unreachable after retries")
check_translations()
check_fetch_set()
check_load_amcr_data()
qgs.exitQgis()
return write_outputs()
if __name__ == "__main__":
sys.exit(main())