spencermandClaude Opus 5.5
6adaf7bf25
FILE checks the archive folder and copies the packet inside the filing transaction. A failed check or copy rolls the filing back, so the packet stays in its previous stage; the person filing gets 503 archive_failed, and in its own transaction the server writes an archive.filing_blocked audit row and raises the critical archive_failed condition to Admin for that packet (who tried, folder, error). The next successful filing clears it. The archive-copy job now only catches up filings made while the archive was off. PLAN question 53, CHANGELOG, FEED and SECURITY updated. Co-Authored-By: Claude Opus 5.5 <[email protected]>
493 lines
22 KiB
Python
493 lines
22 KiB
Python
"""The archive location (Gitea issue 7): the path rules, the check, the File locations API, the copy FILE makes in
|
|
its own transaction, filing refused when that copy fails (Gitea issue 57), and the archive-copy job (idempotence,
|
|
failure and backoff, the Admin condition, old seals queued). Synthetic data only; the archive root is a temp
|
|
folder."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from collections.abc import Iterator
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from closehelpers import at_complete, file_packet, filed, q1, qall, rich_complete, vclose
|
|
from fakedir import FakeDirectory
|
|
from pshelpers import make_settings, needs_pg
|
|
from shophelpers import Shop, ok, open_shop, refused, run
|
|
|
|
from packetstation.domain import archive
|
|
from packetstation.settings import Settings
|
|
from packetstation.worker import jobs, runs
|
|
from packetstation.worker.jobs import archive_copy
|
|
from packetstation.worker.scheduler import Scheduler
|
|
|
|
URL = "/api/admin/file-locations"
|
|
|
|
|
|
def share(tmp_path: Path) -> Path:
|
|
root = tmp_path / "share"
|
|
(root / "packets").mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
|
|
# ---------------------------------------------------------------- settings and path rules
|
|
|
|
|
|
def test_archive_root_blank_is_off(tmp_path: Path) -> None:
|
|
assert make_settings(tmp_path).archive_root is None
|
|
assert make_settings(tmp_path, archive_root="").archive_root is None
|
|
assert make_settings(tmp_path, archive_root=" ").archive_root is None
|
|
assert archive.root_of(make_settings(tmp_path, archive_root=" ")) is None
|
|
assert make_settings(tmp_path, archive_root="/archive").archive_root == Path("/archive")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("given", "why"),
|
|
[
|
|
("", "Enter a folder"),
|
|
(" ", "Enter a folder"),
|
|
("..", "'..'"),
|
|
("packets/../..", "'..'"),
|
|
("a\\..\\b", "'..'"),
|
|
("pack\x00ets", "NUL"),
|
|
("x" * 401, "longer than"),
|
|
],
|
|
)
|
|
def test_relative_dir_refusals(tmp_path: Path, given: str, why: str) -> None:
|
|
with pytest.raises(archive.PathRefused, match=why):
|
|
archive.relative_dir(share(tmp_path), given)
|
|
|
|
|
|
def test_relative_dir_clean_and_absolute(tmp_path: Path) -> None:
|
|
root = share(tmp_path)
|
|
assert archive.relative_dir(root, "packets") == "packets"
|
|
assert archive.relative_dir(root, "./packets//2026/") == "packets/2026"
|
|
assert archive.relative_dir(root, "packets\\2026") == "packets/2026"
|
|
assert archive.relative_dir(root, str(root / "packets" / "2026")) == "packets/2026"
|
|
with pytest.raises(archive.PathRefused, match="must lie under"):
|
|
archive.relative_dir(root, str(tmp_path / "elsewhere"))
|
|
with pytest.raises(archive.PathRefused, match="must lie under"):
|
|
archive.relative_dir(root, "/etc")
|
|
with pytest.raises(archive.PathRefused, match="not the root"):
|
|
archive.relative_dir(root, str(root))
|
|
|
|
|
|
def test_resolve_dir(tmp_path: Path) -> None:
|
|
root = share(tmp_path)
|
|
rel, real = archive.resolve_dir(root, "packets")
|
|
assert rel == "packets" and real == (root / "packets").resolve()
|
|
with pytest.raises(archive.PathRefused, match="does not exist"):
|
|
archive.resolve_dir(root, "nope")
|
|
(root / "afile").write_bytes(b"x")
|
|
with pytest.raises(archive.PathRefused, match="not a folder"):
|
|
archive.resolve_dir(root, "afile")
|
|
with pytest.raises(archive.PathRefused, match="is not there"):
|
|
archive.resolve_dir(tmp_path / "unmounted", "packets")
|
|
|
|
|
|
def test_resolve_dir_refuses_a_link_out(tmp_path: Path) -> None:
|
|
root = share(tmp_path)
|
|
outside = tmp_path / "outside"
|
|
outside.mkdir()
|
|
try:
|
|
os.symlink(outside, root / "escape", target_is_directory=True)
|
|
except (OSError, NotImplementedError):
|
|
pytest.skip("this OS or user cannot make symlinks")
|
|
with pytest.raises(archive.PathRefused, match="outside the archive root"):
|
|
archive.resolve_dir(root, "escape")
|
|
(outside / "deep").mkdir()
|
|
with pytest.raises(archive.PathRefused, match="outside the archive root"):
|
|
archive.resolve_dir(root, "escape/deep")
|
|
# a link that stays inside the root is fine
|
|
os.symlink(root / "packets", root / "inside", target_is_directory=True)
|
|
assert archive.resolve_dir(root, "inside")[1] == (root / "packets").resolve()
|
|
|
|
|
|
def test_check_and_probe(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
root = share(tmp_path)
|
|
rel, res = archive.check(root, "packets")
|
|
assert rel == "packets" and res.ok and res.free_bytes and res.resolved == str((root / "packets").resolve())
|
|
assert list((root / "packets").iterdir()) == [] # the probe is gone
|
|
assert archive.check(None, "packets") == (None, archive.CheckResult(False, None, None, archive.OFF_MESSAGE))
|
|
rel, res = archive.check(root, "../x")
|
|
assert rel is None and not res.ok and "'..'" in res.message
|
|
|
|
def broken_fsync(fd: int) -> None:
|
|
raise OSError(5, "Input/output error")
|
|
|
|
monkeypatch.setattr(archive.os, "fsync", broken_fsync)
|
|
rel, res = archive.check(root, "packets")
|
|
assert rel == "packets" and not res.ok and "Cannot write" in res.message and "Input/output error" in res.message
|
|
assert list((root / "packets").iterdir()) == []
|
|
|
|
|
|
def test_backoff() -> None:
|
|
assert [archive.backoff(n) for n in (1, 2, 3, 4, 9)] == [
|
|
timedelta(minutes=5),
|
|
timedelta(minutes=15),
|
|
timedelta(minutes=60),
|
|
timedelta(minutes=60),
|
|
timedelta(minutes=60),
|
|
]
|
|
|
|
|
|
def test_safe_names() -> None:
|
|
assert archive.safe_part("L5/12:3", "x") == "L5_12_3"
|
|
assert archive.safe_part("..", "x") == "x"
|
|
taken: set[str] = set()
|
|
assert [archive._unique(n, taken) for n in ("a.pdf", "A.pdf", "a.pdf", "manifest.json", "b")] == [
|
|
"a.pdf",
|
|
"A (2).pdf",
|
|
"a (3).pdf",
|
|
"manifest (2).json",
|
|
"b",
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------- with PostgreSQL
|
|
|
|
|
|
@pytest.fixture
|
|
def on_settings(pg_settings: Settings, tmp_path: Path) -> Settings:
|
|
return pg_settings.model_copy(update={"archive_root": share(tmp_path)})
|
|
|
|
|
|
@pytest.fixture
|
|
def shop(on_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
|
|
yield from open_shop(on_settings, fake_dir)
|
|
|
|
|
|
@pytest.fixture
|
|
def off_shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
|
|
yield from open_shop(pg_settings, fake_dir)
|
|
|
|
|
|
def run_copy(settings: Settings) -> dict[str, Any]:
|
|
sch = Scheduler(settings, build="pytest", jobs={archive.JOB: archive_copy.SPEC})
|
|
return runs.run_job(sch, archive.JOB, "schedule")
|
|
|
|
|
|
def save_dir(settings: Settings, archive_dir: str = "packets") -> None:
|
|
run(
|
|
settings,
|
|
"INSERT INTO settings (key, value) VALUES ('files', jsonb_build_object('archive_dir', %s::text))"
|
|
" ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, version = settings.version + 1",
|
|
(archive_dir,),
|
|
)
|
|
|
|
|
|
def copies(settings: Settings) -> list[Any]:
|
|
return qall(settings, "SELECT seal_id, status, attempts, last_error, next_try_at, folder FROM archive_copies")
|
|
|
|
|
|
def audits(settings: Settings, action: str) -> list[dict[str, Any]]:
|
|
return [r[0] for r in qall(settings, "SELECT detail FROM audit_events WHERE action = %s ORDER BY id", (action,))]
|
|
|
|
|
|
def sha(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
@needs_pg
|
|
def test_api_off(off_shop: Shop) -> None:
|
|
ann = off_shop("admin.ann")
|
|
body = ok(ann.get(URL))
|
|
assert body == {
|
|
"enabled": False,
|
|
"root": None,
|
|
"archive_dir": None,
|
|
"version": 0,
|
|
"resolved": None,
|
|
"last_check": None,
|
|
"copies": {"pending": 0, "failed": 0, "done": 0, "recent_failures": []},
|
|
}
|
|
for r in (
|
|
ann.post(f"{URL}/check", json={"archive_dir": "packets"}),
|
|
ann.put(URL, json={"archive_dir": "packets", "version": 0}),
|
|
ann.post(f"{URL}/retry"),
|
|
):
|
|
refused(r, 409, "archive_off", archive.OFF_MESSAGE)
|
|
# Admin only, password session
|
|
assert off_shop("clerk.clara").get(URL).status_code == 403
|
|
|
|
|
|
@needs_pg
|
|
def test_api_check_save_and_retry(shop: Shop) -> None:
|
|
s = shop.settings
|
|
root = Path(str(s.archive_root))
|
|
ann = shop("admin.ann")
|
|
body = ok(ann.get(URL))
|
|
assert body["enabled"] is True and body["root"] == str(root) and body["archive_dir"] is None
|
|
|
|
refused(ann.post(f"{URL}/check", json={"archive_dir": "../etc"}), 422, "bad_path")
|
|
refused(ann.post(f"{URL}/check", json={"archive_dir": "missing"}), 422, "bad_path")
|
|
good = ok(ann.post(f"{URL}/check", json={"archive_dir": str(root / "packets")}))
|
|
assert good["ok"] is True and good["resolved"] == str((root / "packets").resolve()) and good["free_bytes"] > 0
|
|
assert ok(ann.get(URL))["archive_dir"] is None # the check saves nothing
|
|
checks = audits(s, "archive.checked")
|
|
assert [(c["source"], c["ok"]) for c in checks] == [("admin", False), ("admin", False), ("admin", True)]
|
|
assert checks[2]["archive_dir"] == "packets"
|
|
|
|
refused(ann.put(URL, json={"archive_dir": "missing", "version": 0}), 422, "bad_path")
|
|
refused(ann.put(URL, json={"archive_dir": "/elsewhere", "version": 0}), 422, "bad_path")
|
|
saved = ok(ann.put(URL, json={"archive_dir": str(root / "packets"), "version": 0}))
|
|
assert saved["archive_dir"] == "packets" and saved["version"] == 1
|
|
assert saved["resolved"] == str((root / "packets").resolve())
|
|
assert saved["last_check"]["ok"] is True and saved["last_check"]["by"] == "Ann Admin"
|
|
assert set(saved["last_check"]) == {"ok", "at", "by", "message"}
|
|
conflict = refused(ann.put(URL, json={"archive_dir": "packets", "version": 0}), 409, "version_conflict")
|
|
assert conflict["current"]["version"] == 1
|
|
(root / "other").mkdir()
|
|
assert ok(ann.put(URL, json={"archive_dir": "other", "version": 1}))["archive_dir"] == "other"
|
|
assert [a["archive_dir"] for a in audits(s, "admin.file_locations_updated")] == [
|
|
{"from": None, "to": "packets"},
|
|
{"from": "packets", "to": "other"},
|
|
]
|
|
# a check that fails after the path rules (the probe) answers check_failed on save and keeps the old folder
|
|
(root / "ro").mkdir()
|
|
orig = archive.probe
|
|
try:
|
|
archive.probe = lambda folder: archive.CheckResult(False, str(folder), None, "Cannot write in it.")
|
|
r = refused(ann.put(URL, json={"archive_dir": "ro", "version": 2}), 422, "check_failed", "Cannot write in it.")
|
|
assert r["check"]["ok"] is False
|
|
finally:
|
|
archive.probe = orig
|
|
assert ok(ann.get(URL))["archive_dir"] == "other"
|
|
# saving asks for a copy run; Retry now resets the backoff and asks again (202)
|
|
r = ann.post(f"{URL}/retry")
|
|
assert r.status_code == 202 and r.json()["requested"] is True
|
|
assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 3
|
|
assert audits(s, "admin.archive_retry")
|
|
assert shop("clerk.clara").post(f"{URL}/retry").status_code == 403
|
|
|
|
|
|
@needs_pg
|
|
def test_copy_job(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
s = shop.settings
|
|
root = Path(str(s.archive_root))
|
|
events: list[tuple[str, str, dict[str, Any]]] = []
|
|
monkeypatch.setattr(jobs, "raise_event", lambda conn, t, **kw: events.append(("raise", t, kw)))
|
|
monkeypatch.setattr(jobs, "resolve_event", lambda conn, t, **kw: events.append(("resolve", t, kw)))
|
|
|
|
# no folder saved: the job idles, nothing queued
|
|
row = run_copy(s)
|
|
assert row["status"] == "ok" and row["counts"]["skipped"] == "no archive folder saved"
|
|
save_dir(s)
|
|
|
|
pkt = rich_complete(shop)
|
|
pid = pkt["id"]
|
|
# rich_complete keeps one short open (a packet from before change C7); close it so FILE accepts
|
|
run(s, "UPDATE bom_shorts SET closed_at = now() WHERE packet_id = %s", (pid,))
|
|
out = ok(file_packet(shop, vclose(shop, pkt)))
|
|
assert out["status"] == "filed"
|
|
# FILE made the copy itself, in the filing transaction: nothing waits for the worker
|
|
assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 0
|
|
seal_id, book_id = q1(s, "SELECT id, databook_asset_id FROM packet_seals WHERE packet_id = %s", (pid,))
|
|
((sid, status, attempts, err, nxt, folder),) = copies(s)
|
|
assert (sid, status, attempts, err, nxt) == (seal_id, "done", 1, None, None)
|
|
project, unit = q1(
|
|
s,
|
|
"SELECT pr.number, pr.number || u.lot FROM packets p JOIN units u ON u.id = p.unit_id"
|
|
" JOIN projects pr ON pr.id = u.project_id WHERE p.id = %s",
|
|
(pid,),
|
|
)
|
|
assert folder.startswith(f"packets/{project}/{unit}/") and folder.endswith(f"-seal{seal_id}")
|
|
where = root.joinpath(*folder.split("/"))
|
|
manifest = json.loads((where / "manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["packet_id"] == pid and manifest["seal_id"] == seal_id and manifest["project"] == project
|
|
assert manifest["unit"] == unit and manifest["build"] == s.ps_build and manifest["filed_by"] == "Clara Clerk"
|
|
assert manifest["sealed_by"] == "Clara Clerk" and len(manifest["seal_hash"]) == 64 and manifest["key_id"]
|
|
kinds = [f["kind"] for f in manifest["files"]]
|
|
assert kinds[0] == "databook" and manifest["files"][0]["asset_id"] == str(book_id)
|
|
assert {"databook", "page", "doc", "qc_doc", "mtr"} <= set(kinds), kinds
|
|
assert kinds.count("mtr") == 2
|
|
names = [f["name"] for f in manifest["files"]]
|
|
assert len({n.lower() for n in names}) == len(names)
|
|
assert sorted(p.name for p in where.iterdir()) == sorted([*names, "manifest.json"])
|
|
for f in manifest["files"]:
|
|
assert sha(where / f["name"]) == f["sha256"] and (where / f["name"]).stat().st_size == f["size"]
|
|
assert q1(s, "SELECT manifest_sha256 FROM archive_copies")[0] == sha(where / "manifest.json")
|
|
(done,) = audits(s, "archive.copied")
|
|
assert done["seal_id"] == seal_id and done["folder"] == folder and done["files"] == len(names)
|
|
assert done["source"] == "filing"
|
|
|
|
# idempotent: a run copies nothing; a row set back to pending finds the identical files and is done
|
|
row = run_copy(s)
|
|
assert row["status"] == "ok" and row["counts"]["queued"] == 0 and row["counts"]["copied"] == 0, row
|
|
assert events == [("resolve", "archive_failed", {"subject": "archive", "title": "Archive copy failing"})]
|
|
events.clear()
|
|
run(s, "UPDATE archive_copies SET status = 'pending'")
|
|
before = {p.name: p.stat().st_mtime_ns for p in where.iterdir()}
|
|
assert run_copy(s)["counts"]["copied"] == 1
|
|
assert {p.name: p.stat().st_mtime_ns for p in where.iterdir()} == before
|
|
assert copies(s)[0][1] == "done"
|
|
|
|
# a different file in the way: never overwritten; failed with a 5-minute backoff, then 15, then 60 and raised
|
|
run(s, "UPDATE archive_copies SET status = 'pending', folder = NULL, done_at = NULL, attempts = 0")
|
|
(where / "manifest.json").unlink()
|
|
victim = where / names[1]
|
|
victim.write_bytes(b"something else")
|
|
events.clear()
|
|
t0 = datetime.now(UTC)
|
|
row = run_copy(s)
|
|
assert row["status"] == "degraded" and row["counts"]["failed"] == 1
|
|
_, status, attempts, err, nxt, _ = copies(s)[0]
|
|
assert (status, attempts) == ("failed", 1) and "not overwritten" in err and names[1] in err
|
|
assert timedelta(minutes=4) < nxt - t0 < timedelta(minutes=6)
|
|
assert victim.read_bytes() == b"something else" and not (where / "manifest.json").exists()
|
|
assert events == [] # one failure: not raised yet, not cleared either
|
|
assert run_copy(s)["counts"]["failed"] == 0 # not due
|
|
for n, wait in ((2, 15), (3, 60)):
|
|
run(s, "UPDATE archive_copies SET next_try_at = now() - interval '1 second'")
|
|
t0 = datetime.now(UTC)
|
|
run_copy(s)
|
|
_, status, attempts, _, nxt, _ = copies(s)[0]
|
|
assert (status, attempts) == ("failed", n)
|
|
assert timedelta(minutes=wait - 1) < nxt - t0 < timedelta(minutes=wait + 1)
|
|
raised = [e for e in events if e[0] == "raise"]
|
|
assert len(raised) == 1 and raised[0][1] == "archive_failed" and raised[0][2]["subject"] == "archive"
|
|
assert raised[0][2]["condition"] is True and f"seal {seal_id}" in raised[0][2]["body"]
|
|
assert len(audits(s, "archive.copy_failed")) == 3
|
|
st = ok(shop("admin.ann").get(URL))["copies"]
|
|
assert (st["failed"], st["done"], st["pending"]) == (1, 0, 0)
|
|
(f,) = st["recent_failures"]
|
|
assert (
|
|
f["seal_id"] == seal_id
|
|
and f["packet_id"] == pid
|
|
and f["attempts"] == 3
|
|
and "not overwritten" in f["last_error"]
|
|
)
|
|
assert f["packet_label"].startswith(project) and f["next_try_at"]
|
|
|
|
# the Admin removes the stray file and presses Retry now: done, condition cleared
|
|
victim.unlink()
|
|
events.clear()
|
|
assert shop("admin.ann").post(f"{URL}/retry").status_code == 202
|
|
row = run_copy(s)
|
|
assert row["counts"]["copied"] == 1 and copies(s)[0][1] == "done"
|
|
assert sha(victim) == manifest["files"][1]["sha256"]
|
|
assert events == [("resolve", "archive_failed", {"subject": "archive", "title": "Archive copy failing"})]
|
|
|
|
|
|
@needs_pg
|
|
def test_old_seals_are_queued_when_the_archive_is_turned_on(off_shop: Shop, tmp_path: Path) -> None:
|
|
s = off_shop.settings
|
|
pkt = filed(off_shop)
|
|
# archive off: FILE asks for nothing, and the job does nothing
|
|
assert q1(s, "SELECT count(*) FROM job_requests WHERE job = 'archive-copy'")[0] == 0
|
|
assert run_copy(s)["counts"] == {"enabled": False, "queued": 0, "copied": 0, "failed": 0}
|
|
assert copies(s) == []
|
|
|
|
on = s.model_copy(update={"archive_root": share(tmp_path)})
|
|
save_dir(on)
|
|
row = run_copy(on)
|
|
assert row["counts"]["queued"] == 1 and row["counts"]["copied"] == 1
|
|
seal = q1(s, "SELECT id FROM packet_seals WHERE packet_id = %s", (pkt["id"],))[0]
|
|
assert copies(s)[0][:2] == (seal, "done")
|
|
|
|
|
|
def open_filing_events(settings: Settings, pid: int) -> list[Any]:
|
|
return qall(
|
|
settings,
|
|
"SELECT title, body, detail, severity, raised_count FROM notification_events"
|
|
" WHERE event_type = 'archive_failed' AND subject = %s AND resolved_at IS NULL",
|
|
(f"filing:{pid}",),
|
|
)
|
|
|
|
|
|
@needs_pg
|
|
def test_filing_is_refused_when_the_archive_folder_is_gone(shop: Shop) -> None:
|
|
s = shop.settings
|
|
root = Path(str(s.archive_root))
|
|
save_dir(s, "gone") # saved, then the folder went away
|
|
pkt = vclose(shop, at_complete(shop))
|
|
pid = pkt["id"]
|
|
|
|
body = refused(file_packet(shop, pkt), 503, "archive_failed")
|
|
assert body["detail"].startswith("The packet could not be copied to the archive: The archive folder check failed:")
|
|
assert "does not exist" in body["detail"] and "Nothing was filed, and Admin has been told." in body["detail"]
|
|
# the whole filing rolled back: still complete, no seal, no data book, no copy row
|
|
view = ok(shop("clerk.clara").get(f"/api/packets/{pid}"))
|
|
assert view["status"] == "complete" and view["seal"] is None
|
|
assert q1(s, "SELECT count(*) FROM packet_seals WHERE packet_id = %s", (pid,))[0] == 0
|
|
assert q1(s, "SELECT databook_at FROM packets WHERE id = %s", (pid,))[0] is None
|
|
assert copies(s) == []
|
|
# recorded on its own: the audit row and one open Admin condition with the details
|
|
(blocked,) = audits(s, "archive.filing_blocked")
|
|
assert blocked["packet_id"] == pid and blocked["archive_dir"] == "gone" and blocked["by"] == "Clara Clerk"
|
|
assert blocked["status"] == "complete" and "does not exist" in blocked["error"]
|
|
((title, text, detail, severity, raised),) = open_filing_events(s, pid)
|
|
assert title.startswith("Filing blocked: archive copy failed: ") and blocked["packet_label"] in title
|
|
assert "Clara Clerk tried to file" in text and "Archive folder: gone" in text and "does not exist" in text
|
|
assert severity == "critical" and raised == 1 and detail["packet_id"] == pid
|
|
|
|
# a second try fails the same way: the open condition only counts it
|
|
refused(file_packet(shop, pkt), 503, "archive_failed")
|
|
assert open_filing_events(s, pid)[0][4] == 2 and len(audits(s, "archive.filing_blocked")) == 2
|
|
|
|
# the folder comes back: the filing goes through, the copy is done and the condition is cleared
|
|
(root / "gone").mkdir()
|
|
out = ok(file_packet(shop, pkt))
|
|
assert out["status"] == "filed"
|
|
((_, status, attempts, err, _, folder),) = copies(s)
|
|
assert (status, attempts, err) == ("done", 1, None) and folder.startswith("gone/")
|
|
assert open_filing_events(s, pid) == []
|
|
cleared = q1(
|
|
s,
|
|
"SELECT title, body FROM notification_events WHERE event_type = 'archive_failed' AND subject = %s"
|
|
" AND recovery_of IS NOT NULL",
|
|
(f"filing:{pid}",),
|
|
)
|
|
assert cleared[0].startswith("Cleared: Filing blocked") and cleared[1] == "The packet was filed and copied."
|
|
|
|
|
|
@needs_pg
|
|
def test_filing_is_refused_when_the_copy_fails(shop: Shop, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
s = shop.settings
|
|
save_dir(s)
|
|
|
|
def boom(*a: Any, **k: Any) -> Any:
|
|
raise archive.CopyError("A different file is already at Data_Book.pdf in the archive; it was not overwritten.")
|
|
|
|
monkeypatch.setattr(archive, "copy_seal", boom)
|
|
pkt = vclose(shop, at_complete(shop))
|
|
body = refused(file_packet(shop, pkt), 503, "archive_failed")
|
|
assert "it was not overwritten. Nothing was filed" in body["detail"]
|
|
assert q1(s, "SELECT status FROM packets WHERE id = %s", (pkt["id"],))[0] == "complete"
|
|
assert q1(s, "SELECT count(*) FROM packet_seals WHERE packet_id = %s", (pkt["id"],))[0] == 0
|
|
((_, text, detail, _, _),) = open_filing_events(s, pkt["id"])
|
|
assert "not overwritten" in text and detail["archive_dir"] == "packets"
|
|
|
|
# anything unexpected blocks the filing the same way
|
|
monkeypatch.setattr(archive, "copy_seal", lambda *a, **k: 1 / 0)
|
|
body = refused(file_packet(shop, pkt), 503, "archive_failed")
|
|
assert "ZeroDivisionError" in body["detail"]
|
|
|
|
|
|
@needs_pg
|
|
def test_filing_with_no_folder_saved_copies_nothing(shop: Shop) -> None:
|
|
# ARCHIVE_ROOT is set but the Admin has not saved a folder: the archive is not in use yet, so filing goes ahead
|
|
pkt = filed(shop)
|
|
assert pkt["status"] == "filed" and copies(shop.settings) == []
|
|
|
|
|
|
@needs_pg
|
|
def test_doctor_archive_check(pg_settings: Settings, tmp_path: Path) -> None:
|
|
from packetstation import doctor
|
|
|
|
assert doctor.check_archive(pg_settings)[0].status == "SKIP"
|
|
on = pg_settings.model_copy(update={"archive_root": share(tmp_path)})
|
|
(c,) = doctor.check_archive(on)
|
|
assert c.status == "SKIP" and "no archive folder saved" in c.detail
|
|
save_dir(on)
|
|
(c,) = doctor.check_archive(on)
|
|
assert c.status == "PASS" and "writable" in c.detail
|
|
save_dir(on, "gone")
|
|
(c,) = doctor.check_archive(on)
|
|
assert c.status == "FAIL" and "does not exist" in c.detail
|
|
assert [(a["source"], a["ok"]) for a in audits(on, "archive.checked")] == [("doctor", True), ("doctor", False)]
|