253 lines
8.7 KiB
Python
253 lines
8.7 KiB
Python
"""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
|