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.
56 lines
2.2 KiB
Python
56 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
|
|
def test_second_attach_of_same_filename_returns_existing_key(tmp_path: Path):
|
|
s = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage")
|
|
key = s.create(Source(title="T", url="https://x/1"))
|
|
f = tmp_path / "attachment_1.pdf"
|
|
f.write_bytes(b"%PDF")
|
|
k1 = s.attach_file(key, f, title="attachment_1.pdf")
|
|
k2 = s.attach_file(key, f, title="attachment_1.pdf")
|
|
assert k1 == k2
|
|
n = s._con().execute("SELECT count(*) FROM attachments").fetchone()[0]
|
|
assert n == 1
|
|
# exactly one storage copy
|
|
assert len(list((tmp_path / "storage").iterdir())) == 1
|
|
|
|
|
|
def test_different_filename_creates_second_row(tmp_path: Path):
|
|
s = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage")
|
|
key = s.create(Source(title="T", url="https://x/1"))
|
|
f1 = tmp_path / "a.pdf"
|
|
f1.write_bytes(b"a")
|
|
f2 = tmp_path / "b.pdf"
|
|
f2.write_bytes(b"b")
|
|
assert s.attach_file(key, f1) != s.attach_file(key, f2)
|
|
|
|
|
|
def test_explicit_title_dedupes_across_different_source_files(tmp_path: Path):
|
|
"""refs #680: the dedupe key is (item_id, filename) where filename
|
|
is ``title or path.name`` — an explicit title must dedupe on its
|
|
own value even when the two calls point at physically different
|
|
source files (different path.name, different bytes)."""
|
|
s = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage")
|
|
key = s.create(Source(title="T", url="https://x/1"))
|
|
f1 = tmp_path / "original_name_1.pdf"
|
|
f1.write_bytes(b"first version")
|
|
f2 = tmp_path / "totally_different_name.pdf"
|
|
f2.write_bytes(b"second version, different bytes and path.name")
|
|
|
|
k1 = s.attach_file(key, f1, title="Comment text")
|
|
k2 = s.attach_file(key, f2, title="Comment text")
|
|
|
|
assert k1 == k2
|
|
n = s._con().execute("SELECT count(*) FROM attachments").fetchone()[0]
|
|
assert n == 1
|
|
# only the first file's bytes were ever copied into storage
|
|
assert len(list((tmp_path / "storage").iterdir())) == 1
|
|
stored = next((tmp_path / "storage").iterdir())
|
|
copied = next(stored.iterdir())
|
|
assert copied.read_bytes() == b"first version"
|