2c83005564
При непустой `role_permission` вставляются пары, у которых роль или право вставлены этим же прогоном (`_upsert` возвращает вставленные ключи): их администратор снять не мог. Прочие пары по-прежнему не трогаются — снятые права не возвращаются. Убрана мёртвая ветка `unchanged` (R-9): при пустой таблице совпадений быть не может, при заполненной сверка не нужна. Тесты: новое право и новая роль во временном seed-каталоге → повторный seed выдаёт только их пары, снятая связь не возвращается; администратор снял все связи → матрица заливается заново. README, CLAUDE.md, комментарий `role_permissions.yaml` и докстринг `seed.py` уточнены. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
429 lines
17 KiB
Python
429 lines
17 KiB
Python
"""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
|