fix: clean up folder and tag assignments on tree soft delete

When a tree is soft-deleted, folder assignments and tag assignments are
now removed from junction tables. Tag usage counts are decremented with
a floor of zero to prevent negative counts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
chihlasm
2026-02-06 00:23:24 -05:00
parent 02e00963e1
commit 5ae22e041f
2 changed files with 100 additions and 3 deletions

View File

@@ -3,15 +3,15 @@ from typing import Annotated, Optional
from uuid import UUID from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, status, Query from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, or_, true as sa_true from sqlalchemy import select, func, or_, true as sa_true, update
from sqlalchemy.orm import selectinload from sqlalchemy.orm import selectinload
from app.core.database import get_db from app.core.database import get_db
from app.models.tree import Tree from app.models.tree import Tree
from app.models.user import User from app.models.user import User
from app.models.category import TreeCategory from app.models.category import TreeCategory
from app.models.tag import TreeTag from app.models.tag import TreeTag, tree_tag_assignments
from app.models.folder import UserFolder from app.models.folder import UserFolder, user_folder_trees
from app.schemas.tree import TreeCreate, TreeUpdate, TreeResponse, TreeListResponse, CategoryInfo from app.schemas.tree import TreeCreate, TreeUpdate, TreeResponse, TreeListResponse, CategoryInfo
from app.api.deps import get_current_active_user, require_engineer_or_admin, require_admin from app.api.deps import get_current_active_user, require_engineer_or_admin, require_admin
from app.core.permissions import can_edit_tree, can_access_tree from app.core.permissions import can_edit_tree, can_access_tree
@@ -522,6 +522,31 @@ async def delete_tree(
tree.is_active = False tree.is_active = False
tree.deleted_at = datetime.now(timezone.utc) tree.deleted_at = datetime.now(timezone.utc)
tree.deleted_by = current_user.id tree.deleted_by = current_user.id
# Clean up folder assignments
await db.execute(
user_folder_trees.delete().where(user_folder_trees.c.tree_id == tree.id)
)
# Decrement usage_count on associated tags (floor at 0)
tag_ids_result = await db.execute(
select(tree_tag_assignments.c.tag_id).where(
tree_tag_assignments.c.tree_id == tree.id
)
)
tag_ids = [row[0] for row in tag_ids_result.fetchall()]
if tag_ids:
await db.execute(
update(TreeTag)
.where(TreeTag.id.in_(tag_ids))
.values(usage_count=func.greatest(TreeTag.usage_count - 1, 0))
)
# Clean up tag assignments
await db.execute(
tree_tag_assignments.delete().where(tree_tag_assignments.c.tree_id == tree.id)
)
await log_audit(db, current_user.id, "tree.delete", "tree", tree.id, await log_audit(db, current_user.id, "tree.delete", "tree", tree.id,
{"tree_name": tree.name}) {"tree_name": tree.name})
await db.commit() await db.commit()

View File

@@ -2,6 +2,10 @@
import pytest import pytest
from httpx import AsyncClient from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.folder import user_folder_trees
from app.models.tag import tree_tag_assignments
class TestTrees: class TestTrees:
@@ -203,6 +207,74 @@ class TestTrees:
tree_ids = [t["id"] for t in list_response.json()] tree_ids = [t["id"] for t in list_response.json()]
assert private_tree_id in tree_ids assert private_tree_id in tree_ids
@pytest.mark.asyncio
async def test_delete_tree_cleans_up_folder_and_tag_assignments(
self, client: AsyncClient, auth_headers: dict, admin_auth_headers: dict, test_db: AsyncSession
):
"""Test that soft-deleting a tree removes folder and tag junction entries."""
# Create a tree
tree_data = {
"name": "Cascade Test Tree",
"description": "Will be deleted",
"tree_structure": {
"id": "root",
"type": "solution",
"title": "Test",
"description": "Test tree"
}
}
create_resp = await client.post("/api/v1/trees", json=tree_data, headers=auth_headers)
assert create_resp.status_code == 201
tree_id = create_resp.json()["id"]
# Create a folder and add the tree to it
folder_resp = await client.post(
"/api/v1/folders", json={"name": "Test Folder"}, headers=auth_headers
)
assert folder_resp.status_code == 201
folder_id = folder_resp.json()["id"]
add_resp = await client.post(
f"/api/v1/folders/{folder_id}/trees",
json={"tree_id": tree_id},
headers=auth_headers
)
assert add_resp.status_code == 201
# Add tags to the tree
tag_resp = await client.post(
f"/api/v1/tags/trees/{tree_id}",
json={"tags": ["cascade-test-tag"]},
headers=auth_headers
)
assert tag_resp.status_code == 200
# Verify junction rows exist
folder_rows = await test_db.execute(
select(user_folder_trees).where(user_folder_trees.c.tree_id == tree_id)
)
assert len(folder_rows.fetchall()) > 0
tag_rows = await test_db.execute(
select(tree_tag_assignments).where(tree_tag_assignments.c.tree_id == tree_id)
)
assert len(tag_rows.fetchall()) > 0
# Delete the tree (admin only)
del_resp = await client.delete(f"/api/v1/trees/{tree_id}", headers=admin_auth_headers)
assert del_resp.status_code == 204
# Verify junction rows are gone
folder_rows_after = await test_db.execute(
select(user_folder_trees).where(user_folder_trees.c.tree_id == tree_id)
)
assert len(folder_rows_after.fetchall()) == 0
tag_rows_after = await test_db.execute(
select(tree_tag_assignments).where(tree_tag_assignments.c.tree_id == tree_id)
)
assert len(tag_rows_after.fetchall()) == 0
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_create_tree_unauthorized(self, client: AsyncClient): async def test_create_tree_unauthorized(self, client: AsyncClient):
"""Test that creating a tree without auth fails.""" """Test that creating a tree without auth fails."""