f61eae8656
- _init_schema skips the duplicate-attachments GROUP BY scan once the unique index already exists (checks sqlite_master first). - upsert_status merges extra_json per key instead of replacing the whole column: a stored key now survives unless the incoming payload sets that specific key to a non-empty value. - docket_counts is one grouped query (a single items scan, left-joined against the enriched-tag set) instead of two separate url LIKE scans. - fingerprint_files widens `except FileNotFoundError` to `except OSError` so NotADirectoryError/PermissionError are handled the same way as a plain missing file. - backfill_details and backfill_from_mirror shared an identical sealed-guard block; factored into one _sealed_backfill_skip helper. Adds coverage: extra_json per-key merge (both directions), docket_counts, the widened OSError catch, attach_file's explicit-title dedupe path (two different source files, same explicit title), a dedupe-migration conflict test for a missing kept file, and a test confirming _init_schema's new short-circuit actually skips the scan.
128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
"""bib.dockets — pure docket state helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from bib.dockets import Docket, fingerprint_files, quiet_days, should_seal
|
|
|
|
|
|
def _d(**kw) -> Docket:
|
|
base = dict(
|
|
id="CMS-2026-2377",
|
|
rule_cms_id="CMS-1848-P",
|
|
fr_document_id="CMS-2026-2377-0001",
|
|
fr_object_id="0900006482921ba1",
|
|
comment_end_date="2026-09-14",
|
|
pull_watermark="",
|
|
last_pull_at="",
|
|
last_pull_new=None,
|
|
sealed_at="",
|
|
seal_reason="",
|
|
counts_json="{}",
|
|
)
|
|
base.update(kw)
|
|
return Docket(**base)
|
|
|
|
|
|
class TestShouldSeal:
|
|
def test_seals_after_quiet_period_with_no_new(self):
|
|
d = _d(last_pull_at="2026-10-15T03:00:00Z", last_pull_new=0)
|
|
assert should_seal(d, date(2026, 10, 15), 30) is True
|
|
|
|
def test_not_before_quiet_period(self):
|
|
d = _d(last_pull_at="2026-10-13T03:00:00Z", last_pull_new=0)
|
|
assert should_seal(d, date(2026, 10, 13), 30) is False
|
|
|
|
def test_not_when_last_pull_found_new(self):
|
|
d = _d(last_pull_at="2026-10-20T03:00:00Z", last_pull_new=3)
|
|
assert should_seal(d, date(2026, 10, 20), 30) is False
|
|
|
|
def test_not_when_last_pull_predates_quiet_period(self):
|
|
# Pull happened before end+quiet even though today is well past it.
|
|
d = _d(last_pull_at="2026-09-20T03:00:00Z", last_pull_new=0)
|
|
assert should_seal(d, date(2026, 12, 1), 30) is False
|
|
|
|
def test_not_without_close_date(self):
|
|
d = _d(
|
|
comment_end_date="", last_pull_at="2027-01-01T00:00:00Z", last_pull_new=0
|
|
)
|
|
assert should_seal(d, date(2027, 1, 1), 30) is False
|
|
|
|
def test_not_when_already_sealed(self):
|
|
d = _d(
|
|
sealed_at="2026-10-15T00:00:00Z",
|
|
last_pull_at="2026-10-15T03:00:00Z",
|
|
last_pull_new=0,
|
|
)
|
|
assert should_seal(d, date(2026, 10, 15), 30) is False
|
|
|
|
def test_not_when_never_pulled(self):
|
|
d = _d(last_pull_at="", last_pull_new=None)
|
|
assert should_seal(d, date(2027, 1, 1), 30) is False
|
|
|
|
|
|
class TestFingerprintFiles:
|
|
def test_order_independent(self, tmp_path: Path):
|
|
a = tmp_path / "a.pdf"
|
|
b = tmp_path / "b.pdf"
|
|
a.write_bytes(b"aa")
|
|
b.write_bytes(b"bbb")
|
|
assert fingerprint_files([a, b]) == fingerprint_files([b, a])
|
|
|
|
def test_changes_when_size_changes(self, tmp_path: Path):
|
|
a = tmp_path / "a.pdf"
|
|
a.write_bytes(b"aa")
|
|
f1 = fingerprint_files([a])
|
|
a.write_bytes(b"aaaa")
|
|
assert fingerprint_files([a]) != f1
|
|
|
|
def test_changes_when_mtime_changes(self, tmp_path: Path):
|
|
a = tmp_path / "a.pdf"
|
|
a.write_bytes(b"aa")
|
|
f1 = fingerprint_files([a])
|
|
os.utime(a, ns=(1_000_000_000_000_000_000, 1_000_000_000_000_000_000))
|
|
assert fingerprint_files([a]) != f1
|
|
|
|
def test_missing_file_is_recorded_not_fatal(self, tmp_path: Path):
|
|
assert fingerprint_files([tmp_path / "nope.pdf"]) == fingerprint_files(
|
|
[tmp_path / "nope.pdf"]
|
|
)
|
|
assert fingerprint_files([tmp_path / "nope.pdf"]) != fingerprint_files([])
|
|
|
|
def test_empty_is_stable(self):
|
|
assert fingerprint_files([]) == fingerprint_files([])
|
|
assert len(fingerprint_files([])) == 64
|
|
|
|
def test_non_missing_oserror_is_recorded_not_fatal(self, tmp_path: Path):
|
|
"""refs #680: widened from FileNotFoundError to OSError — a path
|
|
under a non-directory (NotADirectoryError, not
|
|
FileNotFoundError) must not raise, same as a missing file."""
|
|
f = tmp_path / "not_a_dir.pdf"
|
|
f.write_bytes(b"x")
|
|
bogus = f / "nested.pdf" # stat() raises NotADirectoryError
|
|
assert fingerprint_files([bogus]) == fingerprint_files([bogus])
|
|
assert fingerprint_files([bogus]) == fingerprint_files(
|
|
[tmp_path / "nested.pdf"] # same name, both "missing"
|
|
)
|
|
|
|
|
|
class TestQuietDays:
|
|
def test_default_when_section_missing(self, monkeypatch):
|
|
import bib.dockets as mod
|
|
from conf import _Cfg
|
|
|
|
monkeypatch.setattr(mod, "_cfg", lambda: _Cfg({}))
|
|
assert quiet_days() == 30
|
|
|
|
def test_reads_section(self, monkeypatch):
|
|
import bib.dockets as mod
|
|
from conf import _Cfg
|
|
|
|
monkeypatch.setattr(
|
|
mod, "_cfg", lambda: _Cfg({"comments": {"seal_quiet_days": 7}})
|
|
)
|
|
assert quiet_days() == 7
|