fix unable to delete tagged node issue (#586)

Author: ciur

papermerge/core/alembic/versions/88b6b2d497ea_add_ondelete_cascase_to_node_tags.py

@@ -0,0 +1,51 @@
+"""add ondelete cascade to nodes tags
+
+Revision ID: 88b6b2d497ea
+Revises: f0e0da122a9a
+Create Date: 2025-02-01 08:20:24.361716
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "88b6b2d497ea"
+down_revision: Union[str, None] = "f0e0da122a9a"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_foreign_key(
+        "nodes_user_id_fkey",
+        "nodes",
+        "users",
+        ["user_id"],
+        ["id"],
+        ondelete="CASCADE",
+        use_alter=True,
+    )
+    op.drop_constraint("nodes_tags_node_id_fkey", "nodes_tags", type_="foreignkey")
+    op.create_foreign_key(
+        None, "nodes_tags", "nodes", ["node_id"], ["id"], ondelete="CASCADE"
+    )
+    op.create_foreign_key(
+        "tag_user_id_fk", "tags", "users", ["user_id"], ["id"], use_alter=True
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint("tag_user_id_fk", "tags", type_="foreignkey")
+    op.drop_constraint(None, "nodes_tags", type_="foreignkey")
+    op.create_foreign_key(
+        "nodes_tags_node_id_fkey", "nodes_tags", "nodes", ["node_id"], ["id"]
+    )
+    op.drop_constraint("nodes_user_id_fkey", "nodes", type_="foreignkey")
+    # ### end Alembic commands ###

papermerge/core/alembic/versions/bafd773c8533_add_on_casecase_delete_for_tags_.py

@@ -0,0 +1,32 @@
+"""add on casecase=delete for tags association table
+
+Revision ID: bafd773c8533
+Revises: 88b6b2d497ea
+Create Date: 2025-02-01 08:59:18.660765
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'bafd773c8533'
+down_revision: Union[str, None] = '88b6b2d497ea'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint('nodes_tags_tag_id_fkey', 'nodes_tags', type_='foreignkey')
+    op.create_foreign_key(None, 'nodes_tags', 'tags', ['tag_id'], ['id'], ondelete='CASCADE')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'nodes_tags', type_='foreignkey')
+    op.create_foreign_key('nodes_tags_tag_id_fkey', 'nodes_tags', 'tags', ['tag_id'], ['id'])
+    # ### end Alembic commands ###

papermerge/core/features/nodes/tests/test_nodes_router.py

@@ -581,6 +581,58 @@ def test_delete_nodes_with_descendants(
     assert found is None
 
 
+def test_delete_tagged_folder(make_folder, db_session, user, auth_api_client):
+    folder = make_folder(title="My Documents", user=user, parent=user.home_folder)
+
+    nodes_dbapi.assign_node_tags(
+        db_session,
+        node_id=folder.id,
+        tags=["tag1", "tag2"],
+        user_id=user.id,
+    )
+
+    # delete tagged folder
+    response = auth_api_client.delete(f"/nodes/", json=[str(folder.id)])
+
+    assert response.status_code == 200, response.json()
+
+    # folder is not there anymore
+    q = db_session.query(orm.Folder).filter(orm.Folder.id == folder.id)
+    assert db_session.query(q.exists()).scalar() is False
+
+    # but both tags are
+    q_tag1 = db_session.query(orm.Tag).filter(orm.Tag.name == "tag1")
+    q_tag2 = db_session.query(orm.Tag).filter(orm.Tag.name == "tag2")
+    assert db_session.query(q_tag1.exists()).scalar() is True
+    assert db_session.query(q_tag2.exists()).scalar() is True
+
+
+def test_delete_tagged_document(make_document, db_session, user, auth_api_client):
+    doc = make_document(title="My Contract.pdf", user=user, parent=user.home_folder)
+
+    nodes_dbapi.assign_node_tags(
+        db_session,
+        node_id=doc.id,
+        tags=["tag1", "tag2"],
+        user_id=user.id,
+    )
+
+    # delete tagged document
+    response = auth_api_client.delete(f"/nodes/", json=[str(doc.id)])
+
+    assert response.status_code == 200, response.json()
+
+    # document is not there anymore
+    q = db_session.query(orm.Document).filter(orm.Document.id == doc.id)
+    assert db_session.query(q.exists()).scalar() is False
+
+    # but both tags are
+    q_tag1 = db_session.query(orm.Tag).filter(orm.Tag.name == "tag1")
+    q_tag2 = db_session.query(orm.Tag).filter(orm.Tag.name == "tag2")
+    assert db_session.query(q_tag1.exists()).scalar() is True
+    assert db_session.query(q_tag2.exists()).scalar() is True
+
+
 def test_get_node_tags_router_when_node_is_folder(
     make_folder, db_session, user, auth_api_client
 ):

papermerge/core/features/tags/db/api.py

@@ -95,6 +95,9 @@ def get_tag(
         error = schema.Error(messages=[str(e)])
         return None, error
 
+    if db_item is None:
+        raise EntityNotFound
+
     return schema.Tag.model_validate(db_item), None
 
 

papermerge/core/features/tags/db/orm.py

@@ -8,9 +8,11 @@
 class NodeTagsAssociation(Base):
     __tablename__ = "nodes_tags"
     id: Mapped[int] = mapped_column(primary_key=True)
-    node_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("nodes.id"))
+    node_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("nodes.id", ondelete="CASCADE")
+    )
     tag_id: Mapped[uuid.UUID] = mapped_column(
-        ForeignKey("tags.id"),
+        ForeignKey("tags.id", ondelete="CASCADE"),
     )
 
 

papermerge/core/features/tags/tests/test_router_tags.py

@@ -1,8 +1,11 @@
+import uuid
+
 from sqlalchemy import func
 
 from papermerge.core.db.engine import Session
 from papermerge.core import orm, schema
 from papermerge.core.tests.types import AuthTestClient
+from papermerge.core.features.nodes.db import api as nodes_dbapi
 
 
 def test_create_tag_route(auth_api_client: AuthTestClient, db_session: Session):
@@ -49,6 +52,19 @@ def test_get_tag_route(auth_api_client: AuthTestClient, make_tag, user):
     assert response.status_code == 200, response.json()
 
 
+def test_get_non_existing_tag(auth_api_client: AuthTestClient, make_tag, user):
+    """
+    In this use case valid UUID is passed - however there is no tag
+    with such UUID
+    """
+
+    response = auth_api_client.get(
+        f"/tags/{uuid.uuid4()}",
+    )
+
+    assert response.status_code == 404, response.json()
+
+
 def test_delete_tag_route(auth_api_client: AuthTestClient, db_session, make_tag, user):
     tag: schema.Tag = make_tag(name="draft", bg_color="red", user=user)
 
@@ -133,3 +149,57 @@ def test_get_all_tags_no_pagination_per_user(make_tag, make_api_client):
     items_user_b = [schema.Tag(**item) for item in response.json()]
 
     assert len(items_user_b) == user_b_tags_count
+
+
+def test_delete_tag_which_has_associated_folder(
+    make_folder, make_tag, db_session, user, auth_api_client
+):
+    folder = make_folder(title="My Documents", user=user, parent=user.home_folder)
+    tag = make_tag(name="important", user=user)
+
+    nodes_dbapi.assign_node_tags(
+        db_session,
+        node_id=folder.id,
+        tags=["important"],
+        user_id=user.id,
+    )
+
+    # delete tagged folder
+    response = auth_api_client.delete(f"/tags/{tag.id}")
+
+    assert response.status_code == 204
+
+    # folder still exists
+    q = db_session.query(orm.Folder).filter(orm.Folder.id == folder.id)
+    assert db_session.query(q.exists()).scalar() is True
+
+    # but tag was deleted
+    q_tag = db_session.query(orm.Tag).filter(orm.Tag.name == "important")
+    assert db_session.query(q_tag.exists()).scalar() is False
+
+
+def test_delete_tag_which_has_associated_document(
+    make_document, make_tag, db_session, user, auth_api_client
+):
+    doc = make_document(title="My Contract", user=user, parent=user.home_folder)
+    tag = make_tag(name="important", user=user)
+
+    nodes_dbapi.assign_node_tags(
+        db_session,
+        node_id=doc.id,
+        tags=["important"],
+        user_id=user.id,
+    )
+
+    # delete tagged folder
+    response = auth_api_client.delete(f"/tags/{tag.id}")
+
+    assert response.status_code == 204
+
+    # document still exists
+    q = db_session.query(orm.Document).filter(orm.Document.id == doc.id)
+    assert db_session.query(q.exists()).scalar() is True
+
+    # but tag was deleted
+    q_tag = db_session.query(orm.Tag).filter(orm.Tag.name == "important")
+    assert db_session.query(q_tag.exists()).scalar() is False