"""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 delete, 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 @pytest.mark.integration async def test_reseed_grants_only_pairs_of_new_roles_and_permissions(db_session: AsyncSession, tmp_path: Path) -> None: """Матрица непуста: добавляются только связи ролей/прав, вставленных этим прогоном (их снять не могли, ревью R-3); снятая администратором связь существующих роли и права не возвращается.""" seed_dir = _copy_seed(tmp_path) await run_seed(db_session, seed_dir=seed_dir) 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() core = seed_dir / "core" with (core / "permissions.yaml").open("a", encoding="utf-8") as file: file.write('- code: AUDIT_EXPORT\n object: AUDIT\n action: EXPORT\n name: "Аудит: выгрузка"\n') with (core / "roles.yaml").open("a", encoding="utf-8") as file: file.write("- code: AUDITOR\n name: Аудитор\n is_system: true\n is_active: true\n") matrix_path = core / "role_permissions.yaml" matrix = yaml.safe_load(matrix_path.read_text(encoding="utf-8")) matrix["ADMIN"].append("AUDIT_EXPORT") # существующая роль × новое право matrix["AUDITOR"] = ["AUDIT_VIEW"] # новая роль × существующее право matrix_path.write_text(yaml.safe_dump(matrix, allow_unicode=True), encoding="utf-8") report = await run_seed(db_session, seed_dir=seed_dir) rows = ( await db_session.execute( select(Role.code, Permission.code) .join(RolePermission, RolePermission.role_id == Role.id) .join(Permission, Permission.id == RolePermission.permission_id) .where(Role.code.in_(["ADMIN", "AUDITOR"]), Permission.code.in_(["AUDIT_EXPORT", "AUDIT_VIEW"])) ) ).all() assert set(rows) == {("ADMIN", "AUDIT_EXPORT"), ("ADMIN", "AUDIT_VIEW"), ("AUDITOR", "AUDIT_VIEW")} assert (await db_session.execute(expert_view)).scalar_one_or_none() is None assert await _count(db_session, RolePermission) == GRANT_COUNT - 1 + 2 stat = report.tables["role_permission"] assert (stat.inserted, stat.updated, stat.unchanged) == (2, 0, 0) assert stat.skipped_reason is not None @pytest.mark.integration async def test_reseed_refills_matrix_when_admin_removed_all_links(db_session: AsyncSession) -> None: """Пограничный случай: пустая `role_permission` — первичная загрузка, матрица заливается заново.""" await run_seed(db_session) await db_session.execute(delete(RolePermission)) report = await run_seed(db_session) assert await _count(db_session, RolePermission) == GRANT_COUNT assert report.tables["role_permission"].inserted == GRANT_COUNT assert report.tables["role_permission"].skipped_reason is None