diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5077ca6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +*.pyc +.claude/ +.codex/ +.pytest_cache/ +__pycache__/ +data/ diff --git a/DESIGN.md b/DESIGN.md index f851503..153a123 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -136,11 +136,20 @@ Question object (fields per the upstream README): | `picture_question` | Optional. Figure shown with the question stem. | | `picture_a`..`_d` | Optional. Per-choice figures. | -The `picture_*` fields, when present, contain a filename (e.g. -`AB404_q.svg`) that resolves into `svgs/`. Convention observed in the -archive: `_q.svg` for the question figure, `_a.svg` .. -`_d.svg` for per-answer figures. A small number of entries also -ship a `.png` next to the `.svg`. +The `picture_*` fields, when present, contain a **basename without +extension** (e.g. `AB404_q`, not `AB404_q.svg`). The consumer picks +the extension. Convention observed in the archive: + +- `_q` for the question figure (file `_q.svg`), +- `_a` .. `_d` for per-answer figures. + +Files live under `svgs/`. A small number of entries ship a `.png` +alongside the `.svg` (display-problem fallback). Consumers should +prefer `.svg` and fall back to `.png` by stem. + +This stem-only convention is why the soft picture-reference check +matches stems, not full filenames; matching the literal value +against directory listings would mark every reference missing. ### Important consumer-side conventions diff --git a/amateurfunk_fetch.py b/amateurfunk_fetch.py new file mode 100644 index 0000000..53abe00 --- /dev/null +++ b/amateurfunk_fetch.py @@ -0,0 +1,1088 @@ +#!/usr/bin/env python3 +"""Download the latest German amateur-radio exam question catalog from +the Bundesnetzagentur (BNetzA), validate it, and extract it into a +per-edition directory on disk. + +The full design and contract live in DESIGN.md. As an overview the +script does: + + 1. Ask the BNetzA server for the file's HTTP `Last-Modified` header. + 2. Skip everything if the previous run already extracted that exact + version (we recorded the header in manifest-latest.json). + 3. Otherwise stream the ZIP to a scratch file, computing its sha256 + as bytes arrive, and reject anything unsafe to unpack (zip-slip + paths, symlinks, oversized archives, etc.). + 4. Validate the JSON schema and that every picture reference resolves + to a file under svgs/. Missing pictures are a SOFT check: they get + recorded in the manifest but do not fail the extraction. + 5. Extract into a per-edition directory like + `data/2024-03-20-3-auflage/`, with a `manifest.json` next to the + data. + 6. Atomically update `data/manifest-latest.json` so a future run can + do step 1 and skip immediately. + +The script is intentionally a single file with stdlib only. Readability +beats cleverness here — most of the bytes below are docstrings and +comments. Performance is a non-goal (the catalog is ~3 MB and changes +maybe once a quarter). +""" + +import argparse +import dataclasses +import datetime as dt +import hashlib +import json +import os +import posixpath +import re +import shutil +import stat +import sys +import urllib.error +import urllib.request +import zipfile +from pathlib import Path + + +# ============================================================================ +# Constants +# ============================================================================ + +# Canonical download URL. BNetzA replaces the file in-place across +# editions, so this URL is stable; we rely on the HTTP Last-Modified +# header to notice updates. +DEFAULT_SOURCE_URL = ( + "https://www.bundesnetzagentur.de/SharedDocs/Downloads/DE/Sachgebiete/" + "Telekommunikation/Unternehmen_Institutionen/Frequenzen/Amateurfunk/" + "Fragenkatalog/PruefungsfragenZIP.zip?__blob=publicationFile" +) + +# Where extracted editions land by default. Created if it doesn't exist. +DEFAULT_OUT_DIR = Path("data") + +# Polite identifier for the BNetzA web server's access log. Not required +# by the server, but it's a friendly thing to send. +USER_AGENT = ( + "amateurfunk-fetch/0.1 " + "(+https://www.bnetza.de/amateurfunk-fragenkatalog)" +) + +# Compressed-size guard. The real archive is ~3 MB; 50 MB gives a lot +# of headroom while still rejecting a catastrophic upstream swap (such +# as an accidentally-published 10 GB file). +MAX_COMPRESSED_BYTES = 50 * 1024 * 1024 + +# Uncompressed-size guard. Defends against zip-bomb-style upstream +# regressions where a small ZIP expands to gigabytes. The real archive +# expands to ~8 MB. +MAX_UNCOMPRESSED_BYTES = 200 * 1024 * 1024 + +# Sanity floor for the figures directory. The real archive contains +# ~700 SVGs; an archive with fewer than 100 is almost certainly broken. +MIN_SVG_ENTRIES = 100 + +# Connection timeout in seconds for all HTTP requests. +NETWORK_TIMEOUT = 60 + +# Exit codes. See DESIGN.md for the contract behind each one. +EXIT_OK = 0 # Extracted, or already up to date. +EXIT_ERROR = 1 # Network or validation problem. +EXIT_BAD_STATE = 2 # Local state conflict the operator must resolve + # (stale .bak/.tmp dir, or different content + # already present without --force). + + +# ============================================================================ +# Exception types and data classes +# ============================================================================ + + +class ValidationError(Exception): + """Raised when the downloaded archive fails a structural check.""" + + +@dataclasses.dataclass +class ValidatedArchive: + """The result of a successful validation pass. + + Carries everything the caller needs to write a per-edition manifest + and to know what to expect from the extracted tree. + """ + + # Filename of the catalog JSON inside the archive. The name encodes + # the edition (e.g. `fragenkatalog3b.json` for "3rd edition, + # revision b"). + json_filename: str + + # The parsed JSON catalog itself. + json_data: dict + + # Picture references that point at files not present in the archive. + # Each entry has keys: question_number, field, file. + # Soft check: recorded for downstream consumers; doesn't fail the run. + missing_pictures: list + + # Raw text of README.txt from the archive (license + schema docs). + # Preserved verbatim into the extracted tree. + readme_text: str + + # The Quellenvermerk (attribution) sentence pulled out of the + # README. Empty string if the README is missing or has no quoted + # block. + attribution: str + + +# ============================================================================ +# HTTP +# ============================================================================ + + +def http_head_last_modified(url, timeout=NETWORK_TIMEOUT): + """Ask the server for the file's `Last-Modified` header via HEAD. + + Returns the header value as-is (a string like `"Wed, 20 May 2026 + 11:30:40 GMT"`) or `None` if the server didn't send one. + + This is the foundation of the idempotency check: on subsequent runs + we compare this value against what we recorded last time and skip + the actual download if they match. + """ + request = urllib.request.Request( + url, method="HEAD", headers={"User-Agent": USER_AGENT}, + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + return response.headers.get("Last-Modified") + + +def http_get_to_file(url, destination, max_bytes, timeout=NETWORK_TIMEOUT): + """Stream `url` to a local file, computing sha256 as bytes arrive. + + Returns a 3-tuple `(sha256_hex, last_modified, byte_count)`. The + sha256 is computed on the fly so we never need to re-read the file + to hash it. + + Raises `ValidationError` if the response exceeds `max_bytes`, and + `urllib.error.URLError` on connection problems. + """ + request = urllib.request.Request( + url, headers={"User-Agent": USER_AGENT}, + ) + hasher = hashlib.sha256() + bytes_written = 0 + + with urllib.request.urlopen(request, timeout=timeout) as response, \ + destination.open("wb") as out_file: + last_modified = response.headers.get("Last-Modified") + while True: + chunk = response.read(64 * 1024) + if not chunk: + break + bytes_written += len(chunk) + if bytes_written > max_bytes: + raise ValidationError( + f"upstream ZIP exceeds the " + f"{max_bytes}-byte compressed cap" + ) + hasher.update(chunk) + out_file.write(chunk) + + return hasher.hexdigest(), last_modified, bytes_written + + +# ============================================================================ +# ZIP safety +# ============================================================================ + + +def _normalize_zip_name(name): + """Normalize a ZIP member's path for comparison. + + Backslashes (from Windows-produced archives) become forward + slashes, and `posixpath.normpath` collapses redundant separators + and `.` segments. + """ + return posixpath.normpath(name.replace("\\", "/")) + + +def _is_symlink_entry(zi): + """Return True if a ZIP member encodes a POSIX symlink. + + ZIPs created on Unix store the file mode in the upper 16 bits of + `external_attr`. The S_IFLNK file-type bit identifies symlinks. + + Path-based checks (rejecting absolute paths and `..` segments) are + *not* enough on their own: a symlink with an innocuous name could + still point outside the destination root. + """ + upper_mode_bits = (zi.external_attr >> 16) & 0o170000 + return upper_mode_bits == stat.S_IFLNK + + +def _check_zip_member_safe(zi): + """Raise `ValidationError` if extracting this member could escape + the destination directory, or if it's a symlink. + + The whole archive is rejected on the first offence — we never + partially extract a suspicious one. + """ + name = zi.filename + + # Some ZIP producers (mainly on Windows) use backslash separators. + # Reject these outright to avoid ambiguity about what the path means. + if "\\" in name: + raise ValidationError(f"backslash in path: {name!r}") + + # An absolute path would land outside the chosen destination root. + if posixpath.isabs(name): + raise ValidationError(f"absolute path: {name!r}") + + # Reject ANY `..` segment in the raw path, not only entries that + # would escape after normalization. `posixpath.normpath` would + # happily collapse "foo/../bar" to "bar" — which lands in-bounds — + # but the design contract says fail closed on suspicious paths + # regardless of whether the normalized form happens to escape. A + # well-formed archive has no reason to ship `..` anywhere. + raw_segments = name.split("/") + if ".." in raw_segments: + raise ValidationError(f"path escape (.. segment): {name!r}") + + # Reject symlinks identified by the POSIX file-type bits. + if _is_symlink_entry(zi): + raise ValidationError(f"symlink entry: {name!r}") + + +# ============================================================================ +# Validation +# ============================================================================ + + +def validate_zip( + zip_path, + max_uncompressed=MAX_UNCOMPRESSED_BYTES, + min_svg_entries=MIN_SVG_ENTRIES, +): + """Open the ZIP at `zip_path`, run every structural check, and + return a `ValidatedArchive` describing what we found. + + Raises `ValidationError` on the first structural problem. Soft + issues (missing picture references) are collected and returned in + the result rather than raised. + + See DESIGN.md §4 "Validate" for the full list of checks. + """ + if not zipfile.is_zipfile(zip_path): + raise ValidationError(f"not a valid ZIP: {zip_path}") + + with zipfile.ZipFile(zip_path) as zf: + members = zf.infolist() + + # First pass: per-member safety check + accumulate uncompressed size. + total_uncompressed = 0 + for member in members: + _check_zip_member_safe(member) + total_uncompressed += member.file_size + if total_uncompressed > max_uncompressed: + raise ValidationError( + f"uncompressed total {total_uncompressed} exceeds cap " + f"{max_uncompressed} (zip-bomb guard)" + ) + + # Find the catalog JSON. There must be exactly one at the + # archive root — more than one would mean we don't know which + # one is canonical. + root_jsons = _find_root_catalog_jsons(members) + if len(root_jsons) != 1: + raise ValidationError( + f"expected exactly one root-level fragenkatalog*.json, " + f"found {len(root_jsons)}: {root_jsons}" + ) + json_filename = root_jsons[0] + + # Find the figure files. We require more than a configurable + # floor (default 100) so that we'd notice if BNetzA shipped + # only a handful of figures by accident. + svg_members = _find_svg_members(members) + if len(svg_members) <= min_svg_entries: + raise ValidationError( + f"expected more than {min_svg_entries} svgs/ file " + f"entries, found {len(svg_members)}" + ) + svg_stems = _build_svg_stem_index(svg_members) + + # Parse the catalog JSON. `utf-8-sig` tolerates an optional + # UTF-8 BOM at the very start of the file. + with zf.open(json_filename) as f: + try: + catalog = json.loads(f.read().decode("utf-8-sig")) + except json.JSONDecodeError as e: + raise ValidationError( + f"malformed JSON in {json_filename}: {e}" + ) from e + + # Pull the README into memory if it's there. It carries the + # attribution string mandated by the DL-DE license. + readme_text = "" + if "README.txt" in zf.namelist(): + readme_bytes = zf.read("README.txt") + readme_text = readme_bytes.decode( + "utf-8-sig", errors="replace", + ) + + # The JSON-shape checks happen after we've closed the ZIP; they + # only need the parsed catalog and the precomputed SVG stem index. + _validate_catalog_shape(catalog) + missing_pictures = _find_missing_picture_references(catalog, svg_stems) + attribution = _extract_attribution_from_readme(readme_text) + + return ValidatedArchive( + json_filename=json_filename, + json_data=catalog, + missing_pictures=missing_pictures, + readme_text=readme_text, + attribution=attribution, + ) + + +def _find_root_catalog_jsons(members): + """Return ZIP-member filenames that look like the catalog JSON file + at the archive root. + + "Root level" means the path contains no `/`. The actual name + encodes the edition (e.g. `fragenkatalog3b.json` for the 3rd + edition, revision b), so we match against a pattern rather than + hard-coding the current filename. + """ + pattern = re.compile(r"fragenkatalog.+\.json") + matches = [] + for member in members: + if member.is_dir(): + continue + if "/" in member.filename: + continue + if pattern.fullmatch(member.filename): + matches.append(member.filename) + return matches + + +def _find_svg_members(members): + """Return ZIP members whose normalized path starts with `svgs/`. + + Directory entries are skipped, and we deliberately don't require a + standalone `svgs/` directory record — many ZIP producers omit + directory entries and emit only file entries like + `svgs/AB108_q.svg`. + """ + result = [] + for member in members: + if member.is_dir(): + continue + if _normalize_zip_name(member.filename).startswith("svgs/"): + result.append(member) + return result + + +def _build_svg_stem_index(svg_members): + """Build a set of basenames-without-extension, keyed for picture lookup. + + The catalog references pictures by stem (e.g. `"AB109_q"`), not by + full filename (`"AB109_q.svg"`). Indexing by stem keeps the soft + cross-reference check accurate. + """ + stems = set() + for member in svg_members: + basename = posixpath.basename(_normalize_zip_name(member.filename)) + stem, _extension = posixpath.splitext(basename) + stems.add(stem) + return stems + + +# ============================================================================ +# JSON schema +# ============================================================================ + + +# Every question MUST carry these keys. Missing any is a hard failure: +# we can't usefully present a question without four answer slots plus +# the basic identifying fields. +REQUIRED_QUESTION_KEYS = ( + "number", + "class", + "question", + "answer_a", + "answer_b", + "answer_c", + "answer_d", +) + +# Optional fields that, when set, point at an image file under svgs/. +# Soft-validated: a broken reference goes into the manifest but does +# not stop the extraction. +PICTURE_FIELDS = ( + "picture_question", + "picture_a", + "picture_b", + "picture_c", + "picture_d", +) + + +def _validate_catalog_shape(catalog): + """Verify the top-level catalog object has the required structure. + + Hard validation: if the schema doesn't match we won't write a + manifest that claims this is a BNetzA catalog. + """ + if not isinstance(catalog, dict): + raise ValidationError("top-level JSON is not an object") + + for key in ("metadata", "sections"): + if key not in catalog: + raise ValidationError( + f"top-level JSON missing required key {key!r}" + ) + + metadata = catalog["metadata"] + if not isinstance(metadata, dict): + raise ValidationError("metadata is not an object") + for key in ("edition", "issued_on", "valid_from", "license"): + if key not in metadata: + raise ValidationError( + f"metadata missing required key {key!r}" + ) + + if not isinstance(catalog["sections"], list): + raise ValidationError("sections is not a list") + + for section in catalog["sections"]: + _validate_section(section) + + +def _validate_section(node): + """Recursive check for one node in the sections tree. + + Each node is either an inner section (carries nested `sections`) + or a leaf (carries `questions`). Carrying both is ambiguous, and + carrying neither means the node holds no information. We treat the + *presence* of the key as the signal, not whether the list inside + it is empty. + """ + if not isinstance(node, dict): + raise ValidationError("section is not an object") + + # Use `is not None` (rather than truthiness) so that an empty list + # still counts as "the key is present" — that's what the original + # contract calls for. + has_subsections = node.get("sections") is not None + has_questions = node.get("questions") is not None + + if has_subsections and has_questions: + raise ValidationError( + f"section {node.get('title')!r} has both " + f"sections and questions" + ) + if not has_subsections and not has_questions: + raise ValidationError( + f"section {node.get('title')!r} has neither " + f"sections nor questions" + ) + + if has_subsections: + for child in node["sections"]: + _validate_section(child) + else: + for question in node["questions"]: + _validate_question(question) + + +def _validate_question(question): + """Check that a question object has every required field.""" + if not isinstance(question, dict): + raise ValidationError("question is not an object") + for key in REQUIRED_QUESTION_KEYS: + if key not in question: + raise ValidationError( + f"question {question.get('number', '?')!r} " + f"missing required key {key!r}" + ) + + +# ============================================================================ +# Picture cross-references (soft check) +# ============================================================================ + + +def _picture_reference_resolves(reference, svg_stems): + """Return True if a picture reference matches a known SVG/PNG file. + + Accepts both forms: + - the canonical stem (`"AB109_q"`) — matches directly against + the precomputed stem index + - a filename with extension (`"AB109_q.svg"`) — defensive + fallback in case a future BNetzA edition changes the + convention; we strip the extension and try again + + Returning False here is what marks a reference as "missing" — but + that is a soft signal that goes into the manifest, not a hard + failure. + """ + if reference in svg_stems: + return True + stem, _extension = posixpath.splitext(reference) + return stem in svg_stems + + +def _find_missing_picture_references(catalog, svg_stems): + """Walk the catalog and return picture references that don't resolve. + + Soft check: a single upstream typo should not brick the downloader + for everyone until BNetzA ships a fix. Downstream consumers + decide whether to ignore, warn, or hide the affected questions. + + Returns a list of dicts with keys: question_number, field, file. + """ + missing = [] + + def walk(node): + # Inspect every question in this node, if any. + for question in node.get("questions") or []: + for field in PICTURE_FIELDS: + value = question.get(field) + if value and not _picture_reference_resolves(value, svg_stems): + missing.append({ + "question_number": question.get("number"), + "field": field, + "file": value, + }) + # Recurse into sub-sections, if any. + for child in node.get("sections") or []: + walk(child) + + walk({"sections": catalog["sections"]}) + return missing + + +# ============================================================================ +# Attribution +# ============================================================================ + + +def _extract_attribution_from_readme(readme_text): + """Pull the Quellenvermerk (attribution) string out of the README. + + The BNetzA README contains exactly one multi-line quoted block — + the Quellenvermerk that consumers must reproduce verbatim when + redistributing the data under the DL-DE license. We return the + matched text exactly as it appears (multi-line, original spacing + preserved), so a downstream consumer can reproduce it byte-for- + byte from `manifest["attribution"]`. The extracted `README.txt` + file in the per-edition directory remains the lossless source of + truth on disk regardless. + + Returns "" if no quoted block is found. + """ + match = re.search(r'"([^"]+)"', readme_text, flags=re.DOTALL) + if not match: + return "" + return match.group(1) + + +# ============================================================================ +# Slug derivation +# ============================================================================ + + +def derive_slug(metadata): + """Build a filesystem-safe, sortable directory name for this edition. + + The slug combines the issue date and the edition ordinal so that + directories sort chronologically and we never have to parse German + month names from the human-readable `edition` field. + + Example: + `{"edition": "3. Auflage, März 2024", "issued_on": "2024-03-20"}` + -> `"2024-03-20-3-auflage"` + + Both inputs are best-effort. Missing pieces become `"unknown"` so + the slug is still a valid directory name. + """ + issued_on = (metadata.get("issued_on") or "").strip() or "unknown" + edition_text = metadata.get("edition") or "" + + # The German edition string starts with the ordinal + # ("3. Auflage..."), so grab the leading run of digits. + match = re.match(r"\s*(\d+)", edition_text) + ordinal = match.group(1) if match else "unknown" + + return f"{issued_on}-{ordinal}-auflage" + + +# ============================================================================ +# Extraction +# ============================================================================ + + +def extract_zip(zip_path, destination): + """Extract every file from `zip_path` into `destination`. + + Callers must run `validate_zip()` first — this function assumes + the archive has already been checked for path traversal and + symlinks. Directory entries inside the archive are skipped; parent + directories are created on the fly. + """ + destination.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(zip_path) as zf: + for member in zf.infolist(): + if member.is_dir(): + continue + relative_path = _normalize_zip_name(member.filename) + target_path = destination / relative_path + target_path.parent.mkdir(parents=True, exist_ok=True) + with zf.open(member) as source, target_path.open("wb") as out: + shutil.copyfileobj(source, out) + + +# ============================================================================ +# Atomic file write +# ============================================================================ + + +def atomic_write_json(path, payload): + """Write `payload` as pretty JSON to `path`, atomically. + + Sequence (see DESIGN.md §4 step 7): + 1. Write the bytes to a sibling `.tmp` so any failure + doesn't leave a half-written `path`. + 2. fsync the tmp file so the bytes are on disk, not just buffered. + 3. `os.replace` performs the atomic swap. + 4. Best-effort fsync the containing directory so the rename + itself is durable on POSIX. Windows doesn't support directory + fsync — we catch and ignore that. + + A failure during step 1-2 removes the tmp file and the old `path` + stays intact. A failure after step 3 leaves the new file in place. + Either way we never end up with a corrupted `path`. + """ + parent = path.parent + parent.mkdir(parents=True, exist_ok=True) + + tmp_path = parent / (path.name + ".tmp") + encoded = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") + + # Open with low-level os.open so we can fsync the file descriptor + # before closing — important for crash safety. + fd = os.open( + tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644, + ) + try: + with os.fdopen(fd, "wb") as f: + f.write(encoded) + f.flush() + os.fsync(f.fileno()) + except Exception: + # If anything went wrong before we got to `os.replace`, drop + # the tmp file so we don't leak garbage on disk. + tmp_path.unlink(missing_ok=True) + raise + + # The atomic swap. + os.replace(tmp_path, path) + + # Best-effort: persist the rename itself. + _fsync_directory_best_effort(parent) + + +def _fsync_directory_best_effort(directory): + """fsync a directory so the most recent rename inside it is durable. + + Some platforms (notably Windows) don't support opening a directory + for fsync — that's fine, the `os.replace` itself is still atomic, + just slightly less crash-durable on those platforms. + """ + try: + dir_fd = os.open(directory, os.O_DIRECTORY) + except OSError: + return + try: + os.fsync(dir_fd) + except OSError: + pass + finally: + os.close(dir_fd) + + +# ============================================================================ +# Idempotency check +# ============================================================================ + + +def already_up_to_date(out_dir, upstream_last_modified): + """Return True if the previous run already fetched this exact version. + + Compares the upstream Last-Modified value against the one we + recorded in `manifest-latest.json` on the previous successful run. + Also verifies the per-edition `manifest.json` still exists and + parses, so a half-cleaned-up `data/` doesn't fool us. + + Returns False if anything is missing, mismatched, or unparseable — + the safe choice is always to re-download. + """ + if upstream_last_modified is None: + # Server didn't give us a Last-Modified; we can't trust the + # cache for this run. + return False + + latest_pointer = out_dir / "manifest-latest.json" + if not latest_pointer.exists(): + return False + + # Read the pointer file. Anything unparseable means we proceed + # with a fresh download. + try: + latest = json.loads(latest_pointer.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + return False + if not isinstance(latest, dict): + return False + + # The header on disk must exactly equal the one the server just + # told us about. + if latest.get("http_last_modified") != upstream_last_modified: + return False + + # Cross-check that the edition the pointer references is still on + # disk and not just a stale pointer. + cached_slug = latest.get("slug") or "" + if not cached_slug: + return False + cached_manifest = out_dir / cached_slug / "manifest.json" + if not cached_manifest.exists(): + return False + try: + json.loads(cached_manifest.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + return False + + return True + + +# ============================================================================ +# Main orchestration +# ============================================================================ + + +def _parse_args(argv): + """Build the argparse object and parse `argv`.""" + parser = argparse.ArgumentParser( + description=( + "Download the BNetzA German amateur-radio exam question " + "catalog, validate it, and extract it into a per-edition " + "directory." + ), + ) + parser.add_argument( + "--out", + type=Path, + default=DEFAULT_OUT_DIR, + help="output root directory (default: ./data)", + ) + parser.add_argument( + "--force", + action="store_true", + help=( + "re-download and replace an existing edition directory " + "when its content has changed" + ), + ) + parser.add_argument( + "--keep-zip", + action="store_true", + help=( + "keep the raw downloaded ZIP next to the extracted tree, " + "as /source.zip" + ), + ) + # Hidden flag, useful for tests that want to point at a local + # mirror or fixture URL. + parser.add_argument( + "--source-url", + default=DEFAULT_SOURCE_URL, + help=argparse.SUPPRESS, + ) + return parser.parse_args(argv) + + +def main(argv=None): + """Top-level entry point. Returns an exit code; never raises.""" + args = _parse_args(argv) + out_dir = args.out + out_dir.mkdir(parents=True, exist_ok=True) + + # ---- Step 1: HEAD + idempotency skip. + # If the upstream file hasn't been touched since our last + # successful run, there's no useful work left to do. + try: + upstream_last_modified = http_head_last_modified(args.source_url) + except (urllib.error.URLError, TimeoutError) as e: + print(f"error: HEAD request failed: {e}", file=sys.stderr) + return EXIT_ERROR + + if not args.force and already_up_to_date(out_dir, upstream_last_modified): + cached = json.loads( + (out_dir / "manifest-latest.json").read_text("utf-8"), + ) + print( + f"up to date: {cached['slug']} " + f"(Last-Modified={upstream_last_modified})" + ) + return EXIT_OK + + # ---- Step 2: stream the ZIP to a scratch location. + # We put the scratch file inside `out_dir/.tmp/` (rather than + # /tmp/) so that the later `os.replace` happens within the same + # filesystem and is therefore atomic. + scratch_dir = out_dir / ".tmp" + scratch_dir.mkdir(parents=True, exist_ok=True) + zip_scratch = scratch_dir / "download.zip" + + try: + sha256, get_last_modified, zip_size = http_get_to_file( + args.source_url, zip_scratch, max_bytes=MAX_COMPRESSED_BYTES, + ) + except (urllib.error.URLError, TimeoutError, ValidationError) as e: + print(f"error: download failed: {e}", file=sys.stderr) + zip_scratch.unlink(missing_ok=True) + return EXIT_ERROR + + # Prefer the Last-Modified from the GET; fall back to the HEAD + # value if the GET response didn't include one. + effective_last_modified = get_last_modified or upstream_last_modified + + # ---- Step 3: validate the archive. Stops here on any structural + # problem. + try: + archive = validate_zip(zip_scratch) + except ValidationError as e: + print(f"error: validation failed: {e}", file=sys.stderr) + zip_scratch.unlink(missing_ok=True) + return EXIT_ERROR + + # ---- Step 4: derive the per-edition directory name. + slug = derive_slug(archive.json_data["metadata"]) + edition_dir = out_dir / slug + edition_tmp_dir = out_dir / f"{slug}.tmp" + edition_bak_dir = out_dir / f"{slug}.bak" + + # ---- Step 5: decide what to do given any existing edition_dir. + + # When the operator did NOT pass --force, we honor any existing + # extraction: + # * same bytes already on disk: skip re-extraction, just + # refresh the latest pointer + # * different bytes already on disk: refuse to clobber + # + # When --force IS set, we always proceed to re-extract. The + # sha-match shortcut would mask a corrupted on-disk tree (the + # manifest could record the correct sha while files inside are + # missing or truncated), and that is exactly the situation + # --force exists to repair. + if not args.force and edition_dir.exists(): + existing_sha = _read_existing_zip_sha(edition_dir) + if existing_sha == sha256: + print(f"already extracted (sha256 match): {slug}") + _handle_zip_after_success( + zip_scratch, edition_dir, args.keep_zip, + ) + _write_latest_pointer( + out_dir, slug, effective_last_modified, sha256, + ) + _try_remove_empty(scratch_dir) + return EXIT_OK + + print( + f"error: {edition_dir} already exists with different " + f"content. Re-run with --force to replace.", + file=sys.stderr, + ) + zip_scratch.unlink(missing_ok=True) + return EXIT_BAD_STATE + + # ---- Step 6: refuse to clobber any stale .tmp/.bak siblings. + # Each indicates a previous crash and might hold data the operator + # wants to inspect; we never touch them without explicit human + # review. + for stale_path in (edition_tmp_dir, edition_bak_dir): + if stale_path.exists(): + print( + f"error: stale path {stale_path} exists. Inspect " + f"and remove manually before retrying.", + file=sys.stderr, + ) + zip_scratch.unlink(missing_ok=True) + return EXIT_BAD_STATE + + # ---- Step 7: extract into edition_tmp_dir, write its manifest. + try: + extract_zip(zip_scratch, edition_tmp_dir) + except Exception as e: # noqa: BLE001 (surface unexpected fails as exit 1) + print(f"error: extraction failed: {e}", file=sys.stderr) + shutil.rmtree(edition_tmp_dir, ignore_errors=True) + zip_scratch.unlink(missing_ok=True) + return EXIT_ERROR + + # Wrap the remaining steps (manifest write, atomic swap, post- + # swap housekeeping) so any failure becomes EXIT_ERROR instead of + # an exception escaping main(). The docstring promises we never + # raise; this `try` is what enforces it. + try: + # Belt-and-braces: extraction should already have written + # README.txt, but if a hypothetical future archive lacks one + # we fall back to the text we read into memory during + # validation. + readme_in_tmp = edition_tmp_dir / "README.txt" + if not readme_in_tmp.exists() and archive.readme_text: + readme_in_tmp.write_text(archive.readme_text, encoding="utf-8") + + atomic_write_json( + edition_tmp_dir / "manifest.json", + _build_manifest( + source_url=args.source_url, + archive=archive, + slug=slug, + sha256=sha256, + zip_size=zip_size, + last_modified=effective_last_modified, + ), + ) + + # ---- Step 8: atomic directory swap. + # If a previous edition_dir is present, rename it out of the + # way first. If the new rename fails we attempt to restore + # the old one so the operator doesn't end up empty-handed. + # If the restore ALSO fails we leave both `.bak/` and `.tmp/` + # on disk: the operator can recover by hand + # (`mv .bak/ /`). + if edition_dir.exists(): + os.replace(edition_dir, edition_bak_dir) + try: + os.replace(edition_tmp_dir, edition_dir) + except Exception: + os.replace(edition_bak_dir, edition_dir) + raise + shutil.rmtree(edition_bak_dir, ignore_errors=True) + else: + os.replace(edition_tmp_dir, edition_dir) + + # ---- Step 9: post-success housekeeping. + # These don't undo the extraction, but a failure here (disk + # full, permission flip) is something the operator should + # see — so we surface it as EXIT_ERROR rather than swallow. + _handle_zip_after_success( + zip_scratch, edition_dir, args.keep_zip, + ) + _write_latest_pointer( + out_dir, slug, effective_last_modified, sha256, + ) + except Exception as e: # noqa: BLE001 (main() promises not to raise) + print( + f"error: failure after extraction: {e}", + file=sys.stderr, + ) + # If the tmp dir is still on disk it's the abandoned new + # extraction (either pre-swap, or rolled back after a swap + # failure) — safe to drop. We deliberately don't touch + # edition_bak_dir: if it still exists it holds the previous + # edition's content and the operator may want it for manual + # recovery. + if edition_tmp_dir.exists(): + shutil.rmtree(edition_tmp_dir, ignore_errors=True) + zip_scratch.unlink(missing_ok=True) + return EXIT_ERROR + + _try_remove_empty(scratch_dir) + + print(f"extracted: {edition_dir}") + if archive.missing_pictures: + print( + f"note: {len(archive.missing_pictures)} missing picture " + f"reference(s) recorded in manifest (soft check)" + ) + return EXIT_OK + + +# ============================================================================ +# Small helpers used by main() +# ============================================================================ + + +def _read_existing_zip_sha(edition_dir): + """Return the `zip_sha256` from an existing edition's manifest, + or `None` if it's missing or unreadable.""" + manifest_path = edition_dir / "manifest.json" + try: + existing = json.loads(manifest_path.read_text("utf-8")) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(existing, dict): + return None + return existing.get("zip_sha256") + + +def _build_manifest( + source_url, archive, slug, sha256, zip_size, last_modified, +): + """Construct the per-edition `manifest.json` payload.""" + return { + "source_url": source_url, + "fetched_at": dt.datetime.now(dt.timezone.utc).isoformat( + timespec="seconds", + ), + "http_last_modified": last_modified, + "zip_sha256": sha256, + "zip_size": zip_size, + "json_filename": archive.json_filename, + "metadata": archive.json_data["metadata"], + "attribution": archive.attribution, + "missing_pictures": archive.missing_pictures, + "slug": slug, + } + + +def _handle_zip_after_success(zip_path, edition_dir, keep_zip): + """Either move the downloaded ZIP into `edition_dir` (when + `--keep-zip` is set) or remove it. + + Either way, the scratch ZIP is gone by the time this returns. + """ + if keep_zip and zip_path.exists(): + try: + shutil.move(str(zip_path), str(edition_dir / "source.zip")) + return + except OSError: + # Falling through to unlink: archival is best-effort and + # never worth failing the whole run over. + pass + zip_path.unlink(missing_ok=True) + + +def _write_latest_pointer(out_dir, slug, last_modified, sha256): + """Atomically update the pointer that drives the idempotency check.""" + atomic_write_json( + out_dir / "manifest-latest.json", + { + "slug": slug, + "http_last_modified": last_modified, + "zip_sha256": sha256, + }, + ) + + +def _try_remove_empty(directory): + """Remove `directory` if it's empty. Silent if it isn't or doesn't exist.""" + try: + directory.rmdir() + except OSError: + pass + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test_amateurfunk_fetch.py b/test_amateurfunk_fetch.py new file mode 100644 index 0000000..bc13d8c --- /dev/null +++ b/test_amateurfunk_fetch.py @@ -0,0 +1,627 @@ +"""Tests for amateurfunk_fetch. + +Network-bound tests are not included; the real BNetzA fetch lives in +the manual smoke-test invocation. Everything here runs against fixture +ZIPs built in temp directories. +""" +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import stat +import tempfile +import unittest +import zipfile +from pathlib import Path +from unittest import mock + +import amateurfunk_fetch as af + + +# ---------------------------------------------------------------- helpers + + +def make_minimal_json(extra_questions=()): + return { + "metadata": { + "edition": "3. Auflage, März 2024", + "issued_on": "2024-03-20", + "valid_from": "2024-06-24", + "license": "DL-DE->BY-2.0", + }, + "sections": [ + { + "title": "Technische Kenntnisse", + "sections": [ + { + "title": "Mathe", + "questions": [ + { + "number": "NA101", + "class": "1", + "question": "Frage?", + "answer_a": "A", + "answer_b": "B", + "answer_c": "C", + "answer_d": "D", + }, + *list(extra_questions), + ], + } + ], + } + ], + } + + +def make_zip( + path, + *, + json_payload=None, + json_name="fragenkatalog3b.json", + svg_count=150, + include_readme=True, + extra_files=(), + symlink_entry=None, + abs_path_entry=None, + dotdot_entry=None, +): + if json_payload is None: + json_payload = make_minimal_json() + with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf: + zf.writestr( + json_name, + json.dumps(json_payload, ensure_ascii=False).encode("utf-8"), + ) + if include_readme: + zf.writestr( + "README.txt", + 'Some preamble.\n"Attribution string, Bundesnetzagentur"\nMore text.\n', + ) + for i in range(svg_count): + zf.writestr(f"svgs/AB{i:03d}_q.svg", b"") + for name, payload in extra_files: + zf.writestr(name, payload) + if symlink_entry: + zi = zipfile.ZipInfo(symlink_entry) + zi.external_attr = (stat.S_IFLNK | 0o777) << 16 + zf.writestr(zi, b"target") + if abs_path_entry: + zf.writestr(abs_path_entry, b"x") + if dotdot_entry: + zf.writestr(dotdot_entry, b"x") + + +# ---------------------------------------------------------------- unit tests + + +class TestSlug(unittest.TestCase): + def test_normal(self): + meta = {"edition": "3. Auflage, März 2024", "issued_on": "2024-03-20"} + self.assertEqual(af.derive_slug(meta), "2024-03-20-3-auflage") + + def test_missing_ordinal(self): + meta = {"edition": "Sonderdruck", "issued_on": "2025-01-01"} + self.assertEqual(af.derive_slug(meta), "2025-01-01-unknown-auflage") + + def test_missing_issued_on(self): + meta = {"edition": "4. Auflage, Mai 2026", "issued_on": ""} + self.assertEqual(af.derive_slug(meta), "unknown-4-auflage") + + def test_multi_digit_ordinal(self): + meta = {"edition": "12. Auflage", "issued_on": "2030-01-01"} + self.assertEqual(af.derive_slug(meta), "2030-01-01-12-auflage") + + +class TestValidateZip(unittest.TestCase): + def _tmp(self): + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + return Path(d.name) + + def test_valid_minimal(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp) + arch = af.validate_zip(zp) + self.assertEqual(arch.json_filename, "fragenkatalog3b.json") + self.assertEqual(arch.missing_pictures, []) + self.assertIn("Attribution string", arch.attribution) + + def test_no_json(self): + d = self._tmp() + zp = d / "a.zip" + with zipfile.ZipFile(zp, "w") as zf: + zf.writestr("README.txt", "x") + for i in range(120): + zf.writestr(f"svgs/{i}.svg", b"x") + with self.assertRaisesRegex(af.ValidationError, "fragenkatalog"): + af.validate_zip(zp) + + def test_multiple_json(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp, extra_files=(("fragenkatalog4a.json", b"{}"),)) + with self.assertRaisesRegex(af.ValidationError, "exactly one"): + af.validate_zip(zp) + + def test_too_few_svgs(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp, svg_count=10) + with self.assertRaisesRegex(af.ValidationError, "svgs/"): + af.validate_zip(zp) + + def test_malformed_json(self): + d = self._tmp() + zp = d / "a.zip" + with zipfile.ZipFile(zp, "w") as zf: + zf.writestr("fragenkatalog3b.json", b"{not json") + for i in range(150): + zf.writestr(f"svgs/{i}.svg", b"x") + with self.assertRaisesRegex(af.ValidationError, "malformed JSON"): + af.validate_zip(zp) + + def test_missing_top_level_key(self): + d = self._tmp() + zp = d / "a.zip" + bad = make_minimal_json() + del bad["metadata"] + make_zip(zp, json_payload=bad) + with self.assertRaisesRegex(af.ValidationError, "metadata"): + af.validate_zip(zp) + + def test_missing_metadata_keys(self): + d = self._tmp() + zp = d / "a.zip" + bad = make_minimal_json() + del bad["metadata"]["edition"] + make_zip(zp, json_payload=bad) + with self.assertRaisesRegex(af.ValidationError, "edition"): + af.validate_zip(zp) + + def test_section_both_sections_and_questions(self): + d = self._tmp() + zp = d / "a.zip" + bad = make_minimal_json() + bad["sections"][0]["questions"] = [] # Inner node now has both. + make_zip(zp, json_payload=bad) + with self.assertRaisesRegex(af.ValidationError, "both"): + af.validate_zip(zp) + + def test_section_neither(self): + d = self._tmp() + zp = d / "a.zip" + bad = make_minimal_json() + del bad["sections"][0]["sections"] + make_zip(zp, json_payload=bad) + with self.assertRaisesRegex(af.ValidationError, "neither"): + af.validate_zip(zp) + + def test_question_missing_required_key(self): + d = self._tmp() + zp = d / "a.zip" + bad = make_minimal_json() + del bad["sections"][0]["sections"][0]["questions"][0]["answer_d"] + make_zip(zp, json_payload=bad) + with self.assertRaisesRegex(af.ValidationError, "answer_d"): + af.validate_zip(zp) + + def test_zipslip_absolute_path(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp, abs_path_entry="/etc/passwd") + with self.assertRaisesRegex(af.ValidationError, "absolute path"): + af.validate_zip(zp) + + def test_zipslip_dotdot(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp, dotdot_entry="../etc/passwd") + with self.assertRaisesRegex(af.ValidationError, "escape"): + af.validate_zip(zp) + + def test_zipslip_embedded_dotdot(self): + """Regression: a `..` segment embedded mid-path must still be + rejected even though `posixpath.normpath` would collapse it + into an in-bounds filename. The DESIGN contract is to reject + any `..` segment in the raw path, regardless of where it sits + or whether the normalized form happens to escape. + """ + d = self._tmp() + zp = d / "a.zip" + payload = make_minimal_json() + with zipfile.ZipFile(zp, "w") as zf: + zf.writestr( + "fragenkatalog3b.json", + json.dumps(payload, ensure_ascii=False).encode("utf-8"), + ) + zf.writestr("README.txt", 'preamble\n"Attribution"\n') + for i in range(150): + zf.writestr(f"svgs/AB{i:03d}_q.svg", b"") + # Embedded `..` that normpath simplifies to "svgs/bar.svg": + zf.writestr("svgs/foo/../bar.svg", b"") + with self.assertRaisesRegex(af.ValidationError, "escape"): + af.validate_zip(zp) + + def test_zipslip_symlink(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp, symlink_entry="link") + with self.assertRaisesRegex(af.ValidationError, "symlink"): + af.validate_zip(zp) + + def test_uncompressed_cap(self): + d = self._tmp() + zp = d / "a.zip" + make_zip(zp) + with self.assertRaisesRegex(af.ValidationError, "uncompressed"): + af.validate_zip(zp, max_uncompressed=100) + + def test_missing_picture_soft_check(self): + d = self._tmp() + zp = d / "a.zip" + payload = make_minimal_json() + payload["sections"][0]["sections"][0]["questions"][0]["picture_question"] = "missing.svg" + payload["sections"][0]["sections"][0]["questions"][0]["picture_a"] = "AB001_q.svg" # exists + make_zip(zp, json_payload=payload) + arch = af.validate_zip(zp) + self.assertEqual(len(arch.missing_pictures), 1) + self.assertEqual(arch.missing_pictures[0]["file"], "missing.svg") + self.assertEqual(arch.missing_pictures[0]["field"], "picture_question") + self.assertEqual(arch.missing_pictures[0]["question_number"], "NA101") + + def test_picture_reference_as_stem(self): + """Regression: real BNetzA JSON ships picture refs as stems + (e.g. "AB109_q"), not full filenames. Verify stems resolve.""" + d = self._tmp() + zp = d / "a.zip" + payload = make_minimal_json() + q = payload["sections"][0]["sections"][0]["questions"][0] + q["picture_question"] = "AB001_q" # stem only, matches AB001_q.svg + q["picture_a"] = "AB002_q" # stem only + q["picture_b"] = "NOPE" # stem only, no matching file + make_zip(zp, json_payload=payload) + arch = af.validate_zip(zp) + missing = {(m["field"], m["file"]) for m in arch.missing_pictures} + self.assertEqual(missing, {("picture_b", "NOPE")}) + + +# ---------------------------------------------------------------- atomic write + + +class TestAtomicWrite(unittest.TestCase): + def test_replaces_existing(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "x.json" + af.atomic_write_json(p, {"a": 1}) + af.atomic_write_json(p, {"a": 2}) + self.assertEqual(json.loads(p.read_text())["a"], 2) + self.assertEqual(list(Path(d).glob("*.tmp")), []) + + def test_no_tmp_left_on_failure(self): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "x.json" + with mock.patch("os.replace", side_effect=OSError("boom")): + with self.assertRaises(OSError): + af.atomic_write_json(p, {"a": 1}) + # os.replace failed, but os.fsync etc. succeeded; .tmp may or may + # not exist depending on platform. We only assert the final file + # is not present. + self.assertFalse(p.exists()) + + +# ---------------------------------------------------------------- end-to-end + + +def _patch_http(zip_path: Path, *, last_modified: str): + """Returns a pair of context managers patching HEAD and GET.""" + + def fake_head(url, *, timeout=60): + return last_modified + + def fake_get(url, dest, *, max_bytes, timeout=60): + shutil.copyfile(zip_path, dest) + return ( + hashlib.sha256(zip_path.read_bytes()).hexdigest(), + last_modified, + zip_path.stat().st_size, + ) + + return ( + mock.patch.object(af, "http_head_last_modified", fake_head), + mock.patch.object(af, "http_get_to_file", fake_get), + ) + + +class TestEndToEnd(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.out = Path(self.tmp.name) / "out" + self.url = "https://example.invalid/test.zip" + + def _make_default_zip(self, name="a.zip", **kwargs): + zp = Path(self.tmp.name) / name + make_zip(zp, **kwargs) + return zp + + def test_first_run_extracts_everything(self): + zp = self._make_default_zip() + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + rc = af.main(["--out", str(self.out)]) + self.assertEqual(rc, af.EXIT_OK) + + slug_dir = self.out / "2024-03-20-3-auflage" + self.assertTrue((slug_dir / "fragenkatalog3b.json").exists()) + self.assertTrue((slug_dir / "README.txt").exists()) + self.assertTrue((slug_dir / "manifest.json").exists()) + self.assertGreater(len(list((slug_dir / "svgs").iterdir())), 100) + + manifest = json.loads((slug_dir / "manifest.json").read_text()) + self.assertEqual(manifest["slug"], "2024-03-20-3-auflage") + self.assertEqual(manifest["json_filename"], "fragenkatalog3b.json") + self.assertEqual(manifest["metadata"]["edition"], "3. Auflage, März 2024") + self.assertEqual(manifest["missing_pictures"], []) + self.assertIn("Attribution string", manifest["attribution"]) + self.assertEqual(manifest["http_last_modified"], "Wed, 01 Jan 2025 00:00:00 GMT") + + latest = json.loads((self.out / "manifest-latest.json").read_text()) + self.assertEqual(latest["slug"], "2024-03-20-3-auflage") + self.assertEqual(latest["http_last_modified"], "Wed, 01 Jan 2025 00:00:00 GMT") + + # No raw ZIP left in default mode. + self.assertFalse((slug_dir / "source.zip").exists()) + + def test_keep_zip(self): + zp = self._make_default_zip() + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + af.main(["--out", str(self.out), "--keep-zip"]) + slug_dir = self.out / "2024-03-20-3-auflage" + self.assertTrue((slug_dir / "source.zip").exists()) + + def test_rerun_skips_via_last_modified(self): + zp = self._make_default_zip() + get_calls = {"n": 0} + + def fake_head(url, *, timeout=60): + return "Wed, 01 Jan 2025 00:00:00 GMT" + + def fake_get(url, dest, *, max_bytes, timeout=60): + get_calls["n"] += 1 + shutil.copyfile(zp, dest) + return ( + hashlib.sha256(zp.read_bytes()).hexdigest(), + "Wed, 01 Jan 2025 00:00:00 GMT", + zp.stat().st_size, + ) + + with mock.patch.object(af, "http_head_last_modified", fake_head), mock.patch.object( + af, "http_get_to_file", fake_get + ): + self.assertEqual(af.main(["--out", str(self.out)]), af.EXIT_OK) + self.assertEqual(get_calls["n"], 1) + self.assertEqual(af.main(["--out", str(self.out)]), af.EXIT_OK) + self.assertEqual(get_calls["n"], 1) # GET not re-issued. + + def test_rerun_with_same_content_but_new_last_modified(self): + """Server bumped Last-Modified but the ZIP bytes are identical.""" + zp = self._make_default_zip() + ph1, pg1 = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph1, pg1: + af.main(["--out", str(self.out)]) + ph2, pg2 = _patch_http(zp, last_modified="Fri, 03 Jan 2025 00:00:00 GMT") + with ph2, pg2: + rc = af.main(["--out", str(self.out)]) + self.assertEqual(rc, af.EXIT_OK) + # sha256 matched, so manifest-latest should now show the newer Last-Modified. + latest = json.loads((self.out / "manifest-latest.json").read_text()) + self.assertEqual(latest["http_last_modified"], "Fri, 03 Jan 2025 00:00:00 GMT") + + def test_force_replaces_changed_content(self): + zp1 = self._make_default_zip("a.zip") + ph, pg = _patch_http(zp1, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + af.main(["--out", str(self.out)]) + + # New ZIP, same slug (same metadata) but different bytes. + payload = make_minimal_json() + payload["sections"][0]["sections"][0]["questions"][0]["answer_a"] = "modified" + zp2 = self._make_default_zip("b.zip", json_payload=payload) + ph2, pg2 = _patch_http(zp2, last_modified="Thu, 02 Jan 2025 00:00:00 GMT") + with ph2, pg2: + # Without --force the existing non-matching dir should fail BAD_STATE. + rc = af.main(["--out", str(self.out)]) + self.assertEqual(rc, af.EXIT_BAD_STATE) + # With --force it should replace. + rc = af.main(["--out", str(self.out), "--force"]) + self.assertEqual(rc, af.EXIT_OK) + + slug_dir = self.out / "2024-03-20-3-auflage" + manifest = json.loads((slug_dir / "manifest.json").read_text()) + self.assertEqual( + hashlib.sha256(zp2.read_bytes()).hexdigest(), manifest["zip_sha256"] + ) + # No leftover .bak/.tmp directories. + self.assertFalse((self.out / "2024-03-20-3-auflage.bak").exists()) + self.assertFalse((self.out / "2024-03-20-3-auflage.tmp").exists()) + + def test_bak_collision_blocks_force(self): + zp = self._make_default_zip("a.zip") + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + af.main(["--out", str(self.out)]) + + # Plant a stale .bak directory. + bak = self.out / "2024-03-20-3-auflage.bak" + bak.mkdir() + (bak / "stale-file").write_text("leftover") + + # Different content so we don't short-circuit on sha256 match. + payload = make_minimal_json() + payload["sections"][0]["sections"][0]["questions"][0]["answer_a"] = "modified" + zp2 = self._make_default_zip("b.zip", json_payload=payload) + ph2, pg2 = _patch_http(zp2, last_modified="Thu, 02 Jan 2025 00:00:00 GMT") + with ph2, pg2: + rc = af.main(["--out", str(self.out), "--force"]) + self.assertEqual(rc, af.EXIT_BAD_STATE) + # Stale .bak left untouched (operator's call to inspect). + self.assertTrue((bak / "stale-file").exists()) + + def test_missing_picture_recorded_does_not_block(self): + payload = make_minimal_json() + payload["sections"][0]["sections"][0]["questions"][0][ + "picture_question" + ] = "missing.svg" + zp = self._make_default_zip(json_payload=payload) + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + rc = af.main(["--out", str(self.out)]) + self.assertEqual(rc, af.EXIT_OK) + slug_dir = self.out / "2024-03-20-3-auflage" + manifest = json.loads((slug_dir / "manifest.json").read_text()) + self.assertEqual(len(manifest["missing_pictures"]), 1) + + def test_validation_failure_returns_error(self): + # ZIP with a symlink should fail validation and return EXIT_ERROR. + zp = self._make_default_zip(symlink_entry="link-name") + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + rc = af.main(["--out", str(self.out)]) + self.assertEqual(rc, af.EXIT_ERROR) + # Nothing extracted. + self.assertFalse((self.out / "2024-03-20-3-auflage").exists()) + + def test_force_reextracts_when_sha_matches(self): + """--force must repair a corrupted tree even when the manifest's + recorded sha256 matches the freshly downloaded ZIP.""" + zp = self._make_default_zip() + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph, pg: + af.main(["--out", str(self.out)]) + slug_dir = self.out / "2024-03-20-3-auflage" + json_file = slug_dir / "fragenkatalog3b.json" + + # Simulate a corrupted on-disk tree: the manifest still claims + # the right sha, but a critical file is missing. + json_file.unlink() + self.assertFalse(json_file.exists()) + + # Without --force the sha-match shortcut would skip; with + # --force we expect re-extraction to restore the missing file. + ph2, pg2 = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph2, pg2: + rc = af.main(["--out", str(self.out), "--force"]) + self.assertEqual(rc, af.EXIT_OK) + self.assertTrue(json_file.exists()) + + +class TestAttribution(unittest.TestCase): + """Attribution is stored verbatim in the manifest so that consumers + can reproduce the Quellenvermerk byte-for-byte under the DL-DE + license.""" + + def test_attribution_preserves_internal_whitespace(self): + # Multi-line quoted block with deliberate spacing the verbatim + # contract has to preserve. + readme = ( + "Preamble.\n" + '"Prüfungsfragen, Bundesnetzagentur,\n' + "3. Auflage,\n" + '(www.bundesnetzagentur.de/amateurfunk)"\n' + "Trailing text.\n" + ) + result = af._extract_attribution_from_readme(readme) + self.assertIn("\n", result) + self.assertIn("3. Auflage,", result) + self.assertIn("(www.bundesnetzagentur.de/amateurfunk)", result) + # No collapsing: the original line breaks are still there. + self.assertEqual(result.count("\n"), 2) + + def test_attribution_missing_quotes(self): + self.assertEqual(af._extract_attribution_from_readme("no quotes here"), "") + self.assertEqual(af._extract_attribution_from_readme(""), "") + + +class TestPostExtractionFailures(unittest.TestCase): + """`main()` documents that it returns an exit code and never + raises. These tests force failures in the manifest-write and + directory-swap phases and verify that contract holds.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.out = Path(self.tmp.name) / "out" + + def test_manifest_write_failure_returns_error_and_cleans_tmp(self): + zp = Path(self.tmp.name) / "a.zip" + make_zip(zp) + ph, pg = _patch_http(zp, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + + # Fail the very first atomic_write_json call (which is the + # per-edition manifest in edition_tmp_dir). + def faulty_write(path, payload): + raise OSError("simulated manifest write failure") + + with ph, pg, mock.patch.object(af, "atomic_write_json", faulty_write): + rc = af.main(["--out", str(self.out)]) + + self.assertEqual(rc, af.EXIT_ERROR) + # main() promises not to raise — already satisfied above by + # the fact that we got back an int. The cleanup contract: + # neither the new edition dir nor the abandoned tmp dir + # should be left behind. + self.assertFalse((self.out / "2024-03-20-3-auflage").exists()) + self.assertFalse((self.out / "2024-03-20-3-auflage.tmp").exists()) + + def test_swap_failure_returns_error_and_restores_previous(self): + # First run: extract a baseline edition. + zp1 = Path(self.tmp.name) / "a.zip" + make_zip(zp1) + baseline_sha = hashlib.sha256(zp1.read_bytes()).hexdigest() + ph1, pg1 = _patch_http(zp1, last_modified="Wed, 01 Jan 2025 00:00:00 GMT") + with ph1, pg1: + af.main(["--out", str(self.out)]) + + slug_dir = self.out / "2024-03-20-3-auflage" + tmp_dir = self.out / "2024-03-20-3-auflage.tmp" + + # Second run with --force using a ZIP that has the same slug + # but different content. We mock os.replace to fail + # specifically on the tmp->slug rename, which exercises the + # rollback path. + payload = make_minimal_json() + payload["sections"][0]["sections"][0]["questions"][0]["answer_a"] = "DIFFERENT" + zp2 = Path(self.tmp.name) / "b.zip" + make_zip(zp2, json_payload=payload) + + real_replace = os.replace + + def selective_replace(src, dst): + # Only the tmp->slug swap fails; all other renames + # (manifest tmp writes, the rollback, etc.) pass through. + if str(src) == str(tmp_dir) and str(dst) == str(slug_dir): + raise OSError("simulated swap failure") + return real_replace(src, dst) + + ph2, pg2 = _patch_http(zp2, last_modified="Thu, 02 Jan 2025 00:00:00 GMT") + with ph2, pg2, mock.patch("os.replace", side_effect=selective_replace): + rc = af.main(["--out", str(self.out), "--force"]) + + # Contract: EXIT_ERROR, no raise, previous edition restored. + self.assertEqual(rc, af.EXIT_ERROR) + self.assertTrue(slug_dir.exists()) + self.assertFalse(tmp_dir.exists()) + self.assertFalse((self.out / "2024-03-20-3-auflage.bak").exists()) + + # The restored manifest still describes the baseline content. + restored_manifest = json.loads((slug_dir / "manifest.json").read_text()) + self.assertEqual(restored_manifest["zip_sha256"], baseline_sha) + + +if __name__ == "__main__": + unittest.main()