test(persistence): PR8 T8.2 + T8.3 — ServiceAccount cascade / restrict

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>
This commit is contained in:
1445043649
2026-05-14 14:07:16 +08:00
parent 1fb07e48e6
commit bb7289781e
@@ -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()