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