fix(kanban): preserve shared attachment blobs
This commit is contained in:
@@ -1901,11 +1901,16 @@ def delete_attachment(conn: sqlite3.Connection, attachment_id: int) -> Optional[
|
||||
if att is None:
|
||||
return None
|
||||
conn.execute("DELETE FROM task_attachments WHERE id = ?", (attachment_id,))
|
||||
has_remaining_blob_reference = conn.execute(
|
||||
"SELECT 1 FROM task_attachments WHERE stored_path = ? LIMIT 1",
|
||||
(att.stored_path,),
|
||||
).fetchone() is not None
|
||||
_append_event(conn, att.task_id, "attachment_removed", {"filename": att.filename})
|
||||
with contextlib.suppress(OSError):
|
||||
p = Path(att.stored_path)
|
||||
if p.is_file():
|
||||
p.unlink()
|
||||
if not has_remaining_blob_reference:
|
||||
with contextlib.suppress(OSError):
|
||||
p = Path(att.stored_path)
|
||||
if p.is_file():
|
||||
p.unlink()
|
||||
return att
|
||||
|
||||
|
||||
|
||||
@@ -111,6 +111,7 @@ def test_add_list_get_delete_attachment(kanban_home, tmp_path):
|
||||
conn.close()
|
||||
|
||||
|
||||
|
||||
def test_delete_attachment_missing_returns_none(kanban_home):
|
||||
conn = kbc.connect()
|
||||
try:
|
||||
@@ -119,6 +120,36 @@ def test_delete_attachment_missing_returns_none(kanban_home):
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_delete_attachment_keeps_blob_referenced_by_another_row(kanban_home):
|
||||
conn = kbc.connect()
|
||||
try:
|
||||
task_id = _make_task(conn)
|
||||
blob = kb.task_attachments_dir(task_id) / "shared.txt"
|
||||
blob.parent.mkdir(parents=True, exist_ok=True)
|
||||
blob.write_bytes(b"shared attachment")
|
||||
|
||||
first_id = kb.add_attachment(
|
||||
conn,
|
||||
task_id,
|
||||
filename="first.txt",
|
||||
stored_path=str(blob),
|
||||
size=blob.stat().st_size,
|
||||
)
|
||||
second_id = kb.add_attachment(
|
||||
conn,
|
||||
task_id,
|
||||
filename="second.txt",
|
||||
stored_path=str(blob),
|
||||
size=blob.stat().st_size,
|
||||
)
|
||||
|
||||
assert kb.delete_attachment(conn, first_id) is not None
|
||||
assert blob.exists(), "a surviving attachment row must keep its shared blob"
|
||||
assert kb.get_attachment(conn, second_id) is not None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_attachments_root_is_per_board(kanban_home, monkeypatch):
|
||||
# default board uses <root>/kanban/attachments
|
||||
default_root = kb.attachments_root(board="default")
|
||||
@@ -292,5 +323,3 @@ def test_cli_attach_attachments_and_rm(kanban_home, tmp_path):
|
||||
assert kb.list_attachments(conn, task_id) == []
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user