diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 64c9852d73..d895d1210c 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -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 diff --git a/tests/plugins/test_kanban_attachments.py b/tests/plugins/test_kanban_attachments.py index 077c18117a..ade77f31a4 100644 --- a/tests/plugins/test_kanban_attachments.py +++ b/tests/plugins/test_kanban_attachments.py @@ -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 /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() - -