"""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