fix: soft-deleted permission readd

2026-08-25 22:49:19 +08:00
parent 90955a2c1d
commit 498f06ed39
5 changed files with 485 additions and 142 deletions
@@ -0,0 +1,32 @@
+"""Shared test bootstrap.
+
+Loads the repo-root ``.env`` so modules that import ``common.config``
+(which is a process-wide singleton) can resolve ``APP_CONFIG_SECRET_KEY``
+at collection time. Without it, any test importing ``backend.api.*`` fails
+with a RuntimeError about ENC(...) ciphertext.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent.parent
+
+
+def _bootstrap() -> None:
+    if os.environ.get("APP_CONFIG_SECRET_KEY"):
+        return
+    env_file = _REPO_ROOT / ".env"
+    if not env_file.is_file():
+        return
+    try:
+        from dotenv import dotenv_values
+    except ImportError:
+        return
+    value = dotenv_values(env_file).get("APP_CONFIG_SECRET_KEY")
+    if value:
+        os.environ["APP_CONFIG_SECRET_KEY"] = value
+
+
+_bootstrap()
@@ -0,0 +1,252 @@
+"""Regression tests for the ``role_permissions`` re-add (upsert) flow.
+
+Reproduces the PATCH /api/v1/platform/roles/admin/permissions bug where
+re-adding a permission that was previously soft-deleted collides with the
+composite ``(role_id, permission_id)`` PRIMARY KEY: the soft-deleted row
+still occupies its PK slot, so a plain INSERT raises a duplicate-key
+IntegrityError (MySQL 1062).
+
+Uses sqlite in-memory + aiosqlite (``create_async_engine``) and drives
+:func:`_apply_role_permission_set` directly — no HTTP layer, no MySQL.
+"""
+
+from __future__ import annotations
+
+import datetime
+
+import pytest
+from sqlalchemy import func, insert, select, text
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from common.db.models import Permissions, RolePermissions, Roles
+from backend.api.platform._permission_set import _apply_role_permission_set
+
+# -- realistic admin permission codes (matches the reported repro) ----------
+SYSTEM_VIEW = "system:view"
+SYSTEM_USER_VIEW = "system:user:view"
+SYSTEM_PROJECT_VIEW = "system:project:view"
+
+# Raw DDL mirrors the ORM models (roles/permissions/role_permissions) with
+# sqlite-compatible types — the MySQL dialects (CHAR/TINYINT/DATETIME(fsp))
+# cannot be compiled by SQLite's DDL compiler, so the tables are created by
+# hand and then driven through the ORM at runtime.
+_DDL = [
+    """
+    CREATE TABLE roles (
+        role_id VARCHAR(26) PRIMARY KEY,
+        role_code VARCHAR(64) NOT NULL UNIQUE,
+        role_name VARCHAR(100) NOT NULL,
+        role_scope VARCHAR(16) NOT NULL,
+        is_builtin INTEGER NOT NULL DEFAULT 0,
+        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        description VARCHAR(500),
+        is_deleted INTEGER NOT NULL DEFAULT 0,
+        deleted_at DATETIME
+    )
+    """,
+    """
+    CREATE TABLE permissions (
+        permission_id VARCHAR(26) PRIMARY KEY,
+        permission_code VARCHAR(128) NOT NULL UNIQUE,
+        permission_name VARCHAR(100) NOT NULL,
+        module_code VARCHAR(64) NOT NULL,
+        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        description VARCHAR(500),
+        is_deleted INTEGER NOT NULL DEFAULT 0,
+        deleted_at DATETIME
+    )
+    """,
+    """
+    CREATE TABLE role_permissions (
+        role_id VARCHAR(26) NOT NULL,
+        permission_id VARCHAR(26) NOT NULL,
+        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+        is_deleted INTEGER NOT NULL DEFAULT 0,
+        deleted_at DATETIME,
+        PRIMARY KEY (role_id, permission_id)
+    )
+    """,
+]
+
+
+@pytest.fixture
+async def session():
+    """Async in-memory sqlite session with the three identity tables."""
+    engine = create_async_engine(
+        "sqlite+aiosqlite://",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        for ddl in _DDL:
+            await conn.execute(text(ddl))
+    Session = async_sessionmaker(engine, expire_on_commit=False)
+    async with Session() as db:
+        yield db
+    await engine.dispose()
+
+
+# -- helpers ---------------------------------------------------------------
+
+
+async def _add_permission(session, permission_id: str, code: str) -> None:
+    session.add(
+        Permissions(
+            permission_id=permission_id,
+            permission_code=code,
+            permission_name=code,
+            module_code="system",
+            description=None,
+        )
+    )
+    await session.flush()
+
+
+async def _add_role(session, role_id: str, role_code: str) -> Roles:
+    role = Roles(
+        role_id=role_id,
+        role_code=role_code,
+        role_name=role_code,
+        role_scope="platform",
+        is_builtin=1,
+        description=None,
+    )
+    session.add(role)
+    await session.flush()
+    return role
+
+
+async def _link(session, role_id: str, permission_id: str, *, active: bool) -> None:
+    """Insert (or directly mark) a role_permissions row with a given state."""
+    await session.execute(
+        insert(RolePermissions),
+        [
+            {
+                "role_id": role_id,
+                "permission_id": permission_id,
+                "is_deleted": 0 if active else 1,
+                "deleted_at": None if active else datetime.datetime(2026, 1, 1),
+            }
+        ],
+    )
+    await session.flush()
+
+
+async def _active_count(session, role_id: str) -> int:
+    return int(
+        await session.scalar(
+            select(func.count())
+            .select_from(RolePermissions)
+            .where(
+                RolePermissions.role_id == role_id,
+                RolePermissions.is_deleted == 0,
+            )
+        )
+        or 0
+    )
+
+
+async def _active_row_count(session, role_id: str) -> int:
+    return int(
+        await session.scalar(
+            select(func.count())
+            .select_from(RolePermissions)
+            .where(RolePermissions.role_id == role_id)
+        )
+        or 0
+    )
+
+
+# -- tests -----------------------------------------------------------------
+
+
+async def test_happy_path_soft_delete_then_readd(session) -> None:
+    """Soft-delete two permissions, then re-add them — no IntegrityError."""
+    role = await _add_role(session, "R" * 26, "admin")
+    await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
+    await _add_permission(session, "P2" * 13, SYSTEM_USER_VIEW)
+    await _add_permission(session, "P3" * 13, SYSTEM_PROJECT_VIEW)
+
+    view_id = "P1" * 13
+    user_view_id = "P2" * 13
+    project_view_id = "P3" * 13
+    for pid in (view_id, user_view_id, project_view_id):
+        await _link(session, role.role_id, pid, active=True)
+
+    # 1st PATCH: shrink to only system:view → the other two get soft-deleted.
+    codes = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
+    assert codes == [SYSTEM_VIEW]
+    assert await _active_count(session, role.role_id) == 1
+    assert await _active_row_count(session, role.role_id) == 3
+
+    # 2nd PATCH: restore the full set → the soft-deleted rows must be
+    # revived (UPDATE), not INSERTed over (would raise IntegrityError).
+    codes = await _apply_role_permission_set(
+        session,
+        role,
+        [SYSTEM_VIEW, SYSTEM_USER_VIEW, SYSTEM_PROJECT_VIEW],
+    )
+    assert codes == [SYSTEM_PROJECT_VIEW, SYSTEM_USER_VIEW, SYSTEM_VIEW]
+    assert await _active_count(session, role.role_id) == 3
+    assert await _active_row_count(session, role.role_id) == 3
+
+    states = (
+        await session.execute(
+            select(RolePermissions.is_deleted, RolePermissions.deleted_at).where(
+                RolePermissions.role_id == role.role_id
+            )
+        )
+    ).all()
+    assert all(is_deleted == 0 and deleted_at is None for is_deleted, deleted_at in states)
+
+
+async def test_fresh_insert_no_history(session) -> None:
+    """Brand-new role: a never-before-linked permission is plain INSERTed."""
+    role = await _add_role(session, "R" * 26, "admin")
+    await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
+
+    codes = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
+    assert codes == [SYSTEM_VIEW]
+    assert await _active_count(session, role.role_id) == 1
+    assert await _active_row_count(session, role.role_id) == 1
+
+
+async def test_mixed_revive_and_fresh_insert(session) -> None:
+    """One historically soft-deleted row is revived, one fresh row inserted."""
+    role = await _add_role(session, "R" * 26, "admin")
+    await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
+    await _add_permission(session, "P2" * 13, SYSTEM_USER_VIEW)
+    await _link(session, role.role_id, "P1" * 13, active=False)  # 历史软删行
+
+    codes = await _apply_role_permission_set(
+        session, role, [SYSTEM_VIEW, SYSTEM_USER_VIEW]
+    )
+    assert codes == [SYSTEM_USER_VIEW, SYSTEM_VIEW]
+    assert await _active_count(session, role.role_id) == 2
+    assert await _active_row_count(session, role.role_id) == 2
+
+    revived = (
+        await session.execute(
+            select(RolePermissions.is_deleted, RolePermissions.deleted_at).where(
+                RolePermissions.role_id == role.role_id,
+                RolePermissions.permission_id == "P1" * 13,
+            )
+        )
+    ).one()
+    assert revived[0] == 0 and revived[1] is None
+
+
+async def test_same_set_is_noop(session) -> None:
+    """PATCHing the exact same active set again changes nothing."""
+    role = await _add_role(session, "R" * 26, "admin")
+    await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
+    await _link(session, role.role_id, "P1" * 13, active=True)
+
+    first = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
+    second = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
+    assert first == [SYSTEM_VIEW]
+    assert second == [SYSTEM_VIEW]
+    assert await _active_count(session, role.role_id) == 1
+    assert await _active_row_count(session, role.role_id) == 1