628 lines
25 KiB
Python
628 lines
25 KiB
Python
"""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"<svg/>")
|
|
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"<svg/>")
|
|
# Embedded `..` that normpath simplifies to "svgs/bar.svg":
|
|
zf.writestr("svgs/foo/../bar.svg", b"<svg/>")
|
|
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()
|