arm-core/api/tests/test_seed_roles.py

"""Seed ролей, прав, матрицы §10 и ГО (spec reference-data-seed, задача 12).

Матрица §10 переписана в тесте заново и независимо от `seed/core/role_permissions.yaml`.
"""

import shutil
import subprocess
import sys
from pathlib import Path

import pytest
import yaml
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from scripts.seed import SeedError, SeedReport, load_seed, run_seed
from src.models.access import AppUser, Permission, Role, RolePermission
from src.models.directories import RegionalOffice

API_DIR = Path(__file__).resolve().parent.parent
SEED_DIR = API_DIR / "seed"

# docs/context/05-domain-rules.md §10: роль -> множество (объект, действие). Пояснения в скобках — policy, не права.
MATRIX_S10: dict[str, set[tuple[str, str]]] = {
    "ADMIN": {
        ("DIRECTORY", "VIEW"),
        ("DIRECTORY", "EDIT"),
        ("DIRECTORY", "DELETE"),
        ("DIRECTORY", "ADMIN"),
        ("EX", "VIEW"),
        ("USERS", "ADMIN"),
        ("AUDIT", "VIEW"),
    },
    "CURATOR_GO": {
        ("DIRECTORY", "VIEW"),
        ("DIRECTORY", "EDIT"),
        ("CONSOLIDATION", "VIEW"),
        ("KFO", "VIEW"),
        ("EX", "VIEW"),
    },
    "KA_RF": {
        ("DIRECTORY", "VIEW"),
        ("DIRECTORY", "EDIT"),
        ("DOCUMENT", "VIEW"),
        ("DOCUMENT", "EDIT"),
        ("CONSOLIDATION", "VIEW"),
        ("CONSOLIDATION", "EDIT"),
        ("KFO", "VIEW"),
        ("KFO", "EXPORT"),
        ("EX", "VIEW"),
        ("EXPORT", "EXPORT"),
    },
    "RKA": {
        ("DIRECTORY", "VIEW"),
        ("DIRECTORY", "EDIT"),
        ("DOCUMENT", "VIEW"),
        ("CONSOLIDATION", "VIEW"),
        ("CONSOLIDATION", "EDIT"),
        ("KFO", "VIEW"),
        ("KFO", "APPROVE"),
        ("EX", "VIEW"),
        ("EXPORT", "EXPORT"),
    },
    "KA_GO": {
        ("DIRECTORY", "VIEW"),
        ("DIRECTORY", "EDIT"),
        ("DOCUMENT", "VIEW"),
        ("DOCUMENT", "EDIT"),
        ("CONSOLIDATION", "VIEW"),
        ("CONSOLIDATION", "EDIT"),
        ("KFO", "VIEW"),
        ("KFO", "VERIFY"),
        ("KFO", "EXPORT"),
        ("EX", "VIEW"),
        ("EXPORT", "EXPORT"),
    },
    "EXPERT": {
        ("DIRECTORY", "VIEW"),
        ("DOCUMENT", "VIEW"),
        ("CONSOLIDATION", "VIEW"),
        ("KFO", "VIEW"),
        ("EX", "VIEW"),
    },
    "CLIENT_MANAGER": {
        ("DIRECTORY", "VIEW"),
        ("DOCUMENT", "VIEW"),
        ("DOCUMENT", "EDIT"),
        ("CONSOLIDATION", "VIEW"),
        ("KFO", "VIEW"),
    },
}
ROLE_COUNT = 7
PERMISSION_COUNT = 16
GRANT_COUNT = sum(len(pairs) for pairs in MATRIX_S10.values())  # 52
ROLE_CODES = {"KA_RF", "KA_GO", "RKA", "CLIENT_MANAGER", "CURATOR_GO", "EXPERT", "ADMIN"}


async def _count(session: AsyncSession, model: type) -> int:
    return int((await session.execute(select(func.count()).select_from(model))).scalar_one())


async def _counts(session: AsyncSession) -> dict[str, int]:
    return {
        "role": await _count(session, Role),
        "permission": await _count(session, Permission),
        "role_permission": await _count(session, RolePermission),
        "regional_office": await _count(session, RegionalOffice),
    }


def _copy_seed(tmp_path: Path) -> Path:
    target = tmp_path / "seed"
    shutil.copytree(SEED_DIR, target)
    return target


@pytest.mark.integration
async def test_first_run_loads_roles_permissions_and_head_office(db_session: AsyncSession) -> None:
    report = await run_seed(db_session)

    counts = await _counts(db_session)
    assert counts["role"] == ROLE_COUNT
    assert counts["permission"] == PERMISSION_COUNT
    assert counts["role_permission"] == GRANT_COUNT
    assert counts["regional_office"] == 1

    roles = (await db_session.execute(select(Role))).scalars().all()
    assert {r.code for r in roles} == ROLE_CODES
    assert all(r.is_system and r.is_active and r.name for r in roles)

    head = (await db_session.execute(select(RegionalOffice))).scalar_one()
    assert head.name == "Головной офис"
    assert head.is_head_office is True

    assert isinstance(report, SeedReport)
    assert report.tables["role"].inserted == ROLE_COUNT
    assert report.tables["role_permission"].inserted == GRANT_COUNT


@pytest.mark.integration
async def test_second_run_does_not_change_row_counts(db_session: AsyncSession) -> None:
    await run_seed(db_session)
    first = await _counts(db_session)

    report = await run_seed(db_session)

    assert await _counts(db_session) == first
    assert all(stat.inserted == 0 and stat.updated == 0 for stat in report.tables.values())
    # матрица прав непуста — шаг пропущен целиком (issue #9 п.7)
    assert report.tables["role_permission"].skipped_reason is not None
    assert report.tables["role_permission"].unchanged == 0
    assert "role_permission: пропущено" in report.format()


@pytest.mark.integration
async def test_first_run_fills_matrix_without_skip(db_session: AsyncSession) -> None:
    report = await run_seed(db_session)

    assert report.tables["role_permission"].skipped_reason is None
    assert "пропущено" not in report.format()


@pytest.mark.integration
async def test_reseed_does_not_restore_permission_revoked_by_admin(db_session: AsyncSession) -> None:
    """Связь «роль × право», снятая администратором, повторным seed не возвращается; роли и права — upsert."""
    await run_seed(db_session)
    expert_view = (
        select(RolePermission)
        .join(Role, Role.id == RolePermission.role_id)
        .join(Permission, Permission.id == RolePermission.permission_id)
        .where(Role.code == "EXPERT", Permission.code == "EX_VIEW")
    )
    await db_session.delete((await db_session.execute(expert_view)).scalar_one())
    await db_session.flush()
    await db_session.execute(update(Role).where(Role.code == "EXPERT").values(name="Изменено вручную"))

    report = await run_seed(db_session)

    assert (await db_session.execute(expert_view)).scalar_one_or_none() is None
    assert await _count(db_session, RolePermission) == GRANT_COUNT - 1
    assert report.tables["role_permission"].skipped_reason is not None
    assert report.tables["role"].updated == 1  # роли по-прежнему приводятся к файлу


@pytest.mark.integration
async def test_role_permission_matches_matrix_s10(db_session: AsyncSession) -> None:
    await run_seed(db_session)

    rows = (
        await db_session.execute(
            select(Role.code, Permission.object_type, Permission.action)
            .join(RolePermission, RolePermission.role_id == Role.id)
            .join(Permission, Permission.id == RolePermission.permission_id)
        )
    ).all()
    actual: dict[str, set[tuple[str, str]]] = {}
    for code, obj, action in rows:
        actual.setdefault(code, set()).add((obj.value, action.value))

    assert set(actual) == set(MATRIX_S10)
    for role, expected in MATRIX_S10.items():
        assert actual[role] == expected, role


@pytest.mark.integration
async def test_permission_codes_and_grant_metadata(db_session: AsyncSession) -> None:
    await run_seed(db_session)

    perms = (await db_session.execute(select(Permission))).scalars().all()
    for perm in perms:
        expected = (
            perm.action.value
            if perm.object_type.value == perm.action.value
            else (f"{perm.object_type.value}_{perm.action.value}")
        )
        assert perm.code == expected
    assert "EXPORT" in {p.code for p in perms}

    grants = (await db_session.execute(select(RolePermission))).scalars().all()
    assert all(g.granted_by is None and g.granted_at is not None for g in grants)


@pytest.mark.integration
async def test_role_name_change_updates_row_instead_of_inserting(db_session: AsyncSession, tmp_path: Path) -> None:
    seed_dir = _copy_seed(tmp_path)
    await run_seed(db_session, seed_dir=seed_dir)

    roles_path = seed_dir / "core" / "roles.yaml"
    roles = yaml.safe_load(roles_path.read_text(encoding="utf-8"))
    for role in roles:
        if role["code"] == "EXPERT":
            role["name"] = "Эксперт (переименован)"
    roles_path.write_text(yaml.safe_dump(roles, allow_unicode=True), encoding="utf-8")

    report = await run_seed(db_session, seed_dir=seed_dir)

    assert await _count(db_session, Role) == ROLE_COUNT
    expert = (await db_session.execute(select(Role).where(Role.code == "EXPERT"))).scalar_one()
    assert expert.name == "Эксперт (переименован)"
    assert report.tables["role"].updated == 1
    assert report.tables["role"].inserted == 0


@pytest.mark.integration
async def test_unknown_permission_in_matrix_fails_before_any_write(db_session: AsyncSession, tmp_path: Path) -> None:
    seed_dir = _copy_seed(tmp_path)
    matrix_path = seed_dir / "core" / "role_permissions.yaml"
    matrix = yaml.safe_load(matrix_path.read_text(encoding="utf-8"))
    matrix["EXPERT"].append("NO_SUCH_PERMISSION")
    matrix_path.write_text(yaml.safe_dump(matrix, allow_unicode=True), encoding="utf-8")
    before = await _counts(db_session)

    with pytest.raises(SeedError, match="NO_SUCH_PERMISSION"):
        await run_seed(db_session, seed_dir=seed_dir)

    assert (
        await _counts(db_session) == before == {"role": 0, "permission": 0, "role_permission": 0, "regional_office": 0}
    )


@pytest.mark.integration
async def test_unknown_role_in_matrix_fails_before_any_write(db_session: AsyncSession, tmp_path: Path) -> None:
    seed_dir = _copy_seed(tmp_path)
    matrix_path = seed_dir / "core" / "role_permissions.yaml"
    matrix = yaml.safe_load(matrix_path.read_text(encoding="utf-8"))
    matrix["GHOST_ROLE"] = ["EX_VIEW"]
    matrix_path.write_text(yaml.safe_dump(matrix, allow_unicode=True), encoding="utf-8")

    with pytest.raises(SeedError, match="GHOST_ROLE"):
        await run_seed(db_session, seed_dir=seed_dir)

    assert await _count(db_session, Role) == 0


@pytest.mark.integration
async def test_without_dev_flag_no_users_are_loaded(db_session: AsyncSession) -> None:
    await run_seed(db_session)

    assert await _count(db_session, AppUser) == 0
    assert await _count(db_session, RegionalOffice) == 1


@pytest.mark.integration
async def test_dev_flag_is_accepted(db_session: AsyncSession) -> None:
    """Флаг --dev не ломает загрузку core-справочников (dev-данные проверяет test_seed_dev.py)."""
    await run_seed(db_session, dev=True)

    assert await _count(db_session, Role) == ROLE_COUNT


# ---- валидатор без БД ----


def _write(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def test_loader_reads_repository_seed() -> None:
    data = load_seed(SEED_DIR)

    assert len(data.roles) == ROLE_COUNT
    assert len(data.permissions) == PERMISSION_COUNT
    assert len(data.regional_offices) == 1
    assert sum(len(codes) for codes in data.role_permissions.values()) == GRANT_COUNT


@pytest.mark.parametrize(
    ("file", "mutate", "expected"),
    [
        (
            "roles.yaml",
            lambda text: text.replace("code: KA_RF\n", "code: KA_RF\n  unknown_field: 1\n", 1),
            "unknown_field",
        ),
        ("roles.yaml", lambda text: text.replace("is_system: true", "is_system: yes-please", 1), "is_system"),
        ("permissions.yaml", lambda text: text.replace("action: VIEW", "action: LOOK", 1), "LOOK"),
        ("permissions.yaml", lambda text: text.replace("code: DIRECTORY_VIEW", "code: WRONG_CODE", 1), "WRONG_CODE"),
        ("roles.yaml", lambda text: text.replace("code: ADMIN", "code: KA_RF", 1), "KA_RF"),
        ("regional_office.yaml", lambda text: text.replace("- name: Головной офис\n  ", "- ", 1), "name"),
        ("roles.yaml", lambda text: "- [broken", "roles.yaml"),
    ],
)
def test_validator_rejects_malformed_seed(
    tmp_path: Path,
    file: str,
    mutate,
    expected: str,  # noqa: ANN001
) -> None:
    seed_dir = _copy_seed(tmp_path)
    path = seed_dir / "core" / file
    path.write_text(mutate(path.read_text(encoding="utf-8")), encoding="utf-8")

    with pytest.raises(SeedError, match=expected):
        load_seed(seed_dir)


def test_validator_rejects_duplicate_permission_in_role(tmp_path: Path) -> None:
    seed_dir = _copy_seed(tmp_path)
    path = seed_dir / "core" / "role_permissions.yaml"
    path.write_text(path.read_text(encoding="utf-8").replace("ADMIN:\n", "ADMIN:\n  - EX_VIEW\n", 1), encoding="utf-8")

    with pytest.raises(SeedError, match="EX_VIEW"):
        load_seed(seed_dir)


def test_missing_seed_file_is_seed_error(tmp_path: Path) -> None:
    with pytest.raises(SeedError, match="roles.yaml"):
        load_seed(tmp_path)


def test_cli_validation_error_exits_with_code_1_without_db(tmp_path: Path) -> None:
    """CLI с битым каталогом seed: код 1 и сообщение в stderr; до БД дело не доходит."""
    result = subprocess.run(  # noqa: S603
        [sys.executable, "-m", "scripts.seed", "--seed-dir", str(tmp_path)],
        cwd=API_DIR,
        capture_output=True,
        text=True,
        check=False,
    )

    assert result.returncode == 1
    assert "roles.yaml" in result.stderr


def test_cli_help() -> None:
    result = subprocess.run(  # noqa: S603
        [sys.executable, "-m", "scripts.seed", "--help"], cwd=API_DIR, capture_output=True, text=True, check=False
    )

    assert result.returncode == 0
    assert "--dev" in result.stdout