"""Dev-seed (spec reference-data-seed, требование «Dev-данные только по явному флагу», задача 15).""" import os import shutil import subprocess import sys from pathlib import Path import pytest from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from scripts.seed import DEV_FORBIDDEN_MESSAGE, SeedError, check_dev_allowed, load_seed, run_seed from src.core.config import get_settings from src.models.access import AppUser, Permission, Role, RolePermission, UserRole from src.models.directories import RegionalOffice from src.models.enums import BusinessSegment API_DIR = Path(__file__).resolve().parent.parent SEED_DIR = API_DIR / "seed" ROLE_CODES = ["KA_RF", "KA_GO", "RKA", "CLIENT_MANAGER", "CURATOR_GO", "EXPERT", "ADMIN"] USER_COUNT = len(ROLE_CODES) DEV_BRANCH_COUNT = 2 DEV_EMAILS = {f"{code.lower()}@armka.local" for code in ROLE_CODES} HEAD_OFFICE = "Головной офис" #: роли ГО — пользователь в Головном офисе, роли РФ — в региональном филиале HEAD_OFFICE_ROLES = {"KA_GO", "CURATOR_GO", "EXPERT", "ADMIN"} DEV_BRANCHES = {"Ростовский региональный филиал", "Краснодарский региональный филиал"} async def _count(session: AsyncSession, model: type) -> int: return int(await session.scalar(select(func.count()).select_from(model)) or 0) async def _counts(session: AsyncSession) -> dict[str, int]: return { "app_user": await _count(session, AppUser), "user_role": await _count(session, UserRole), "regional_office": await _count(session, RegionalOffice), "role": await _count(session, Role), "permission": await _count(session, Permission), "role_permission": await _count(session, RolePermission), } def _copy_seed(tmp_path: Path) -> Path: target = tmp_path / "seed" shutil.copytree(SEED_DIR, target) return target # ---- интеграция ---- @pytest.mark.integration async def test_without_dev_flag_only_head_office_and_no_users(db_session: AsyncSession) -> None: await run_seed(db_session, dev=False) assert await _count(db_session, AppUser) == 0 assert await _count(db_session, UserRole) == 0 offices = (await db_session.execute(select(RegionalOffice.name, RegionalOffice.is_head_office))).all() assert [(o.name, o.is_head_office) for o in offices] == [(HEAD_OFFICE, True)] @pytest.mark.integration async def test_dev_flag_loads_branches_and_one_user_per_role(db_session: AsyncSession) -> None: report = await run_seed(db_session, dev=True) offices = {o.name: o for o in (await db_session.scalars(select(RegionalOffice))).all()} assert set(offices) == {HEAD_OFFICE, *DEV_BRANCHES} assert [o.name for o in offices.values() if o.is_head_office] == [HEAD_OFFICE] assert all(offices[name].is_active and not offices[name].is_head_office for name in DEV_BRANCHES) users = (await db_session.scalars(select(AppUser))).all() assert {u.email for u in users} == DEV_EMAILS assert all(u.email == u.email.lower() for u in users) # v5.15: email в нижнем регистре assert all(u.is_active and u.external_id is None and u.full_name for u in users) assert all(u.business_segment is BusinessSegment.DKB for u in users) assert all(u.created_at is not None and u.updated_at is not None for u in users) links = ( await db_session.execute( select(AppUser.email, Role.code, AppUser.regional_office_id) .join(UserRole, UserRole.user_id == AppUser.id) .join(Role, Role.id == UserRole.role_id) ) ).all() assert len(links) == len(DEV_EMAILS) # ровно одна роль на пользователя office_name = {o.id: o.name for o in offices.values()} for email, role_code, office_id in links: assert email == f"{role_code.lower()}@armka.local" if role_code in HEAD_OFFICE_ROLES: assert office_name[office_id] == HEAD_OFFICE else: assert office_name[office_id] in DEV_BRANCHES assert report.tables["regional_office"].inserted == DEV_BRANCH_COUNT + 1 assert report.tables["app_user"].inserted == USER_COUNT assert report.tables["user_role"].inserted == USER_COUNT @pytest.mark.integration async def test_second_dev_run_changes_nothing(db_session: AsyncSession) -> None: await run_seed(db_session, dev=True) before = await _counts(db_session) stamps = {u.email: u.updated_at for u in (await db_session.scalars(select(AppUser))).all()} report = await run_seed(db_session, dev=True) assert await _counts(db_session) == before assert before["app_user"] == USER_COUNT assert before["user_role"] == USER_COUNT for table in ("regional_office", "app_user", "user_role", "role_permission"): stat = report.tables[table] assert (stat.inserted, stat.updated) == (0, 0) assert report.tables["role_permission"].skipped_reason is not None # матрица непуста — не трогается db_session.expire_all() assert {u.email: u.updated_at for u in (await db_session.scalars(select(AppUser))).all()} == stamps @pytest.mark.integration async def test_changed_user_data_updates_row_and_keeps_old_role_link(db_session: AsyncSession, tmp_path: Path) -> None: await run_seed(db_session, dev=True) seed_dir = _copy_seed(tmp_path) users_file = seed_dir / "dev" / "users.yaml" text = users_file.read_text(encoding="utf-8") users_file.write_text( text.replace("Dev: Эксперт", "Dev: Эксперт ГО").replace("roles: [ADMIN]", "roles: [KA_GO]"), encoding="utf-8", ) report = await run_seed(db_session, dev=True, seed_dir=seed_dir) assert (report.tables["app_user"].inserted, report.tables["app_user"].updated) == (0, 1) assert (report.tables["user_role"].inserted, report.tables["user_role"].unchanged) == (1, USER_COUNT - 1) assert await _count(db_session, AppUser) == USER_COUNT expert = await db_session.scalar(select(AppUser).where(AppUser.email == "expert@armka.local")) assert expert is not None assert expert.full_name == "Dev: Эксперт ГО" assert await _count(db_session, UserRole) == USER_COUNT + 1 @pytest.mark.integration async def test_seed_stores_email_in_lower_case_and_stays_idempotent(db_session: AsyncSession, tmp_path: Path) -> None: """v5.15: email из файла seed хранится как `strip().lower()`; повторный прогон не плодит строк.""" seed_dir = _copy_seed(tmp_path) users_file = seed_dir / "dev" / "users.yaml" text = users_file.read_text(encoding="utf-8") assert "email: expert@armka.local" in text users_file.write_text(text.replace("email: expert@armka.local", "email: Expert@Armka.LOCAL"), encoding="utf-8") await run_seed(db_session, dev=True, seed_dir=seed_dir) emails = set((await db_session.scalars(select(AppUser.email))).all()) assert emails == DEV_EMAILS report = await run_seed(db_session, dev=True, seed_dir=seed_dir) assert (report.tables["app_user"].inserted, report.tables["app_user"].updated) == (0, 0) assert await _count(db_session, AppUser) == USER_COUNT @pytest.mark.integration @pytest.mark.parametrize( ("old", "new", "expected"), [ ("roles: [EXPERT]", "roles: [NO_SUCH_ROLE]", "NO_SUCH_ROLE"), ( "regional_office: Головной офис\n business_segment: DKB\n is_active: true\n roles: [EXPERT]", "regional_office: Несуществующий филиал\n business_segment: DKB\n is_active: true\n roles: [EXPERT]", "Несуществующий филиал", ), ], ids=["unknown-role", "unknown-branch"], ) async def test_unknown_reference_fails_before_any_write( db_session: AsyncSession, tmp_path: Path, old: str, new: str, expected: str ) -> None: seed_dir = _copy_seed(tmp_path) users_file = seed_dir / "dev" / "users.yaml" text = users_file.read_text(encoding="utf-8") assert old in text users_file.write_text(text.replace(old, new), encoding="utf-8") before = await _counts(db_session) with pytest.raises(SeedError, match=expected): await run_seed(db_session, dev=True, seed_dir=seed_dir) assert await _counts(db_session) == before == dict.fromkeys(before, 0) # ---- валидатор без БД ---- def test_loader_without_dev_does_not_read_dev_files(tmp_path: Path) -> None: seed_dir = _copy_seed(tmp_path) shutil.rmtree(seed_dir / "dev") data = load_seed(seed_dir) assert data.users == [] assert [o.name for o in data.regional_offices] == [HEAD_OFFICE] def test_loader_reads_repository_dev_seed() -> None: data = load_seed(SEED_DIR, dev=True) assert {o.name for o in data.regional_offices} == {HEAD_OFFICE, *DEV_BRANCHES} assert {u.email for u in data.users} == DEV_EMAILS assert all(len(u.role_codes) == 1 and u.email == f"{u.role_codes[0].lower()}@armka.local" for u in data.users) @pytest.mark.parametrize( ("file", "old", "new", "expected"), [ ("users.yaml", ' full_name: "Dev: Эксперт"\n', "", "full_name"), ( "users.yaml", "business_segment: DKB\n is_active: true\n roles: [EXPERT]", "business_segment: XX\n is_active: true\n roles: [EXPERT]", "business_segment", ), ("users.yaml", "roles: [EXPERT]", "roles: []", "ровно одна"), ("users.yaml", "roles: [EXPERT]", "roles: [EXPERT, ADMIN]", "ровно одна"), ("users.yaml", "email: expert@armka.local", "email: admin@armka.local", "admin@armka.local"), ("users.yaml", "is_active: true\n roles: [EXPERT]", "is_active: да\n roles: [EXPERT]", "is_active"), ("users.yaml", " employee_no: DEV-006\n", " employee_no: DEV-006\n extra: 1\n", "extra"), ("regional_offices.yaml", "Краснодарский региональный филиал", "Ростовский региональный филиал", "Ростовский"), ("regional_offices.yaml", "Краснодарский региональный филиал", HEAD_OFFICE, HEAD_OFFICE), ( "regional_offices.yaml", "Ростовский региональный филиал\n is_head_office: false", "Р\n is_head_office: true", "головным", ), ], ids=[ "no-full-name", "bad-segment", "empty-roles", "two-roles", "duplicate-email", "bad-bool", "unknown-key", "duplicate-dev-branch", "dev-branch-duplicates-head-office", "dev-branch-is-head-office", ], ) def test_validator_rejects_malformed_dev_seed(tmp_path: Path, file: str, old: str, new: str, expected: str) -> None: seed_dir = _copy_seed(tmp_path) path = seed_dir / "dev" / file text = path.read_text(encoding="utf-8") assert old in text path.write_text(text.replace(old, new, 1), encoding="utf-8") with pytest.raises(SeedError, match=expected): load_seed(seed_dir, dev=True) def test_missing_dev_file_is_seed_error_only_with_dev_flag(tmp_path: Path) -> None: seed_dir = _copy_seed(tmp_path) (seed_dir / "dev" / "users.yaml").unlink() load_seed(seed_dir) with pytest.raises(SeedError, match="users.yaml"): load_seed(seed_dir, dev=True) # ---- `--dev` только при DEBUG=true (OQ-60) ---- def test_dev_allowed_only_with_debug() -> None: settings = get_settings() assert check_dev_allowed(settings.model_copy(update={"DEBUG": True})) is None assert check_dev_allowed(settings.model_copy(update={"DEBUG": False})) == DEV_FORBIDDEN_MESSAGE assert "DEBUG=true" in DEV_FORBIDDEN_MESSAGE def test_cli_dev_without_debug_exits_with_code_2_before_db() -> None: """`--dev` при DEBUG=false: код 2 и текст в stderr; до БД дело не доходит (переменная окружения важнее ENV_FILE).""" env = {**os.environ, "ENV_FILE": ".env.test", "DEBUG": "false"} result = subprocess.run( # noqa: S603 [sys.executable, "-m", "scripts.seed", "--dev"], cwd=API_DIR, capture_output=True, text=True, check=False, env=env, ) assert result.returncode == 2 # noqa: PLR2004 assert DEV_FORBIDDEN_MESSAGE in result.stderr assert "Seed применён" not in result.stdout