"""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()