From bb7289781ee63c74f9423026c095e76cf0faf600 Mon Sep 17 00:00:00 2001
From: 1445043649 <>
Date: Thu, 14 May 2026 14:07:16 +0800
Subject: [PATCH] =?UTF-8?q?test(persistence):=20PR8=20T8.2=20+=20T8.3=20?=
 =?UTF-8?q?=E2=80=94=20ServiceAccount=20cascade=20/=20restrict?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

T8.2 test_cascade_on_workspace_delete: seed user → workspace → SA,
delete workspace, assert SA row is gone (ondelete CASCADE on
workspace_id FK).

T8.3 test_restrict_on_created_by_user_delete: seed user → workspace
→ SA, then attempt DELETE FROM users WHERE id = created_by, assert
IntegrityError raises and the SA row survives (ondelete RESTRICT on
created_by FK).

SQLite enforces FKs because the engine's connect-listener turns on
`PRAGMA foreign_keys = ON` for every new connection — see engine.py
init_engine().

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 backend/tests/test_service_account_schema.py | 82 ++++++++++++++++++++
 1 file changed, 82 insertions(+)

diff --git a/backend/tests/test_service_account_schema.py b/backend/tests/test_service_account_schema.py
index 777463b1..1d3e8b6c 100644
--- a/backend/tests/test_service_account_schema.py
+++ b/backend/tests/test_service_account_schema.py
@@ -13,6 +13,8 @@ from __future__ import annotations
 from datetime import UTC, datetime
 
 import pytest
+from sqlalchemy import delete
+from sqlalchemy.exc import IntegrityError
 
 from deerflow.persistence.service_account import ServiceAccountRow
 from deerflow.persistence.user.model import UserRow
@@ -92,3 +94,83 @@ async def test_insert_smoke(tmp_path):
         assert row.created_by == "u-alice"
     finally:
         await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# T8.2 — CASCADE on workspace delete
+# ---------------------------------------------------------------------------
+
+
+async def test_cascade_on_workspace_delete(tmp_path):
+    """Deleting the parent workspace removes the service_account row (FK CASCADE)."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf)
+        await _seed_workspace(sf)
+        now = datetime.now(UTC)
+        async with sf() as session:
+            session.add(
+                ServiceAccountRow(
+                    id="sa-2",
+                    workspace_id="w-1",
+                    name="bot",
+                    role="member",
+                    identity_mode="collapsed",
+                    status="active",
+                    created_by="u-alice",
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+            await session.commit()
+
+        async with sf() as session:
+            await session.execute(delete(WorkspaceRow).where(WorkspaceRow.id == "w-1"))
+            await session.commit()
+
+        async with sf() as session:
+            row = await session.get(ServiceAccountRow, "sa-2")
+        assert row is None
+    finally:
+        await _cleanup()
+
+
+# ---------------------------------------------------------------------------
+# T8.3 — RESTRICT on created_by user delete
+# ---------------------------------------------------------------------------
+
+
+async def test_restrict_on_created_by_user_delete(tmp_path):
+    """Deleting the creator user is blocked while their service_account survives."""
+    sf = await _setup(tmp_path)
+    try:
+        await _seed_user(sf)
+        await _seed_workspace(sf)
+        now = datetime.now(UTC)
+        async with sf() as session:
+            session.add(
+                ServiceAccountRow(
+                    id="sa-3",
+                    workspace_id="w-1",
+                    name="bot",
+                    role="member",
+                    identity_mode="collapsed",
+                    status="active",
+                    created_by="u-alice",
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+            await session.commit()
+
+        with pytest.raises(IntegrityError):
+            async with sf() as session:
+                await session.execute(delete(UserRow).where(UserRow.id == "u-alice"))
+                await session.commit()
+
+        # The service_account is still there after the rollback.
+        async with sf() as session:
+            row = await session.get(ServiceAccountRow, "sa-3")
+        assert row is not None
+    finally:
+        await _cleanup()