fix(bib): attach_file is idempotent on (item, filename) (refs #615)
This commit is contained in:
@@ -637,6 +637,14 @@ class Store:
|
|||||||
if row is None:
|
if row is None:
|
||||||
raise KeyError(f"Item not found: {item_key}")
|
raise KeyError(f"Item not found: {item_key}")
|
||||||
|
|
||||||
|
filename = title or path.name
|
||||||
|
dup = con.execute(
|
||||||
|
"SELECT key FROM attachments WHERE item_id = ? AND filename = ?",
|
||||||
|
(row["id"], filename),
|
||||||
|
).fetchone()
|
||||||
|
if dup:
|
||||||
|
return dup["key"]
|
||||||
|
|
||||||
att_key = _generate_key()
|
att_key = _generate_key()
|
||||||
dest_dir = self._storage / att_key
|
dest_dir = self._storage / att_key
|
||||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||||
@@ -652,7 +660,7 @@ class Store:
|
|||||||
(
|
(
|
||||||
row["id"],
|
row["id"],
|
||||||
att_key,
|
att_key,
|
||||||
title or path.name,
|
filename,
|
||||||
content_type,
|
content_type,
|
||||||
# resolve(): through a symlinked storage dir (a git
|
# resolve(): through a symlinked storage dir (a git
|
||||||
# worktree's data/ pointing at the real data dir) the
|
# worktree's data/ pointing at the real data dir) the
|
||||||
|
|||||||
28
tests/bib/test_store_attach_idempotent.py
Normal file
28
tests/bib/test_store_attach_idempotent.py
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
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)
|
||||||
Reference in New Issue
Block a user