imple/Amateurfunk-Anki
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fetcher

Browse Source
This commit is contained in:
Renat Nurgaliyev
2026-05-20 13:59:06 +02:00
parent 8c97c16885
commit 451bee6446
4 changed files with 1735 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+6
View File
@@ -0,0 +1,6 @@
*.pyc
.claude/
.codex/
.pytest_cache/
__pycache__/
data/
DESIGN.md
+14 -5
View File
@@ -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: `<number>_q.svg` for the question figure, `<number>_a.svg` ..
`<number>_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:
- `<number>_q` for the question figure (file `<number>_q.svg`),
- `<number>_a` .. `<number>_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
amateurfunk_fetch.py
+1088
View File
File diff suppressed because it is too large Load Diff
test_amateurfunk_fetch.py
+627
View File
@@ -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"<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()