"""Dev-seed (spec reference-data-seed, требование «Dev-данные только по явному флагу», задача 15).""" import os import re 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 scripts.seed_lib.model import SeedReport from src.core.config import get_settings from src.models.access import AppUser, Permission, Role, RolePermission, UserRole from src.models.directories import ( CompanyGroup, LegalEntity, LegalEntityAlias, RegionalOffice, RelatedGroup, RelatedGroupMember, ) from src.models.enums import BusinessSegment, EntityType, RelatedGroupKind 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 = 3 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 = {"Ростовский региональный филиал", "Краснодарский региональный филиал", "Санкт-Петербургский РФ"} #: dev-ГК под документы fixtures (seed/dev/company_groups.yaml): выгрузка ЕЦП «Русский Бисквит» + владельцы форм ДКБ DEV_GROUPS = {"Русский Бисквит", "Унагранде", "Новороссийский прокатный завод", "Агрофирма Межениновская"} DEV_ENTITY_COUNT = 24 DEV_ALIAS_COUNT = 10 DEV_RELATED_GROUP_COUNT = 2 DEV_MEMBER_COUNT = 4 UNAGRANDE_ENTITY_COUNT = 2 # ЮЛ ГК «Унагранде» (тест переноса ГК на другой филиал) 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), "company_group": await _count(session, CompanyGroup), "legal_entity": await _count(session, LegalEntity), "legal_entity_alias": await _count(session, LegalEntityAlias), "related_group": await _count(session, RelatedGroup), "related_group_member": await _count(session, RelatedGroupMember), } 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, CompanyGroup) == 0 and await _count(db_session, LegalEntity) == 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_cli_no_dev_groups_flag_skips_company_groups(monkeypatch: pytest.MonkeyPatch) -> None: """`--dev --no-dev-groups`: загрузчик получает `dev_groups=False` (пустые справочники для импорта ЕЦП).""" import scripts.seed as seed_module # noqa: PLC0415 - модуль CLI подменяется точечно seen: dict[str, object] = {} async def fake_run_cli(seed_dir: Path, *, dev: bool, dev_groups: bool) -> SeedReport: seen.update(dev=dev, dev_groups=dev_groups) return SeedReport() monkeypatch.setattr(seed_module, "_run_cli", fake_run_cli) assert seed_module.main(["--dev", "--no-dev-groups"]) == 0 assert seen == {"dev": True, "dev_groups": False} assert seed_module.main(["--dev"]) == 0 assert seen == {"dev": True, "dev_groups": True} 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 # ---- dev-ГК под документы fixtures ---- @pytest.mark.integration async def test_dev_flag_loads_company_groups_matching_fixture_documents(db_session: AsyncSession) -> None: """Требование «Dev-данные только по явному флагу», сценарий «Dev-ГК под документы fixtures».""" await run_seed(db_session, dev=True) groups = {group.name: group for group in (await db_session.scalars(select(CompanyGroup))).all()} offices = {office.id: office.name for office in (await db_session.scalars(select(RegionalOffice))).all()} entities = {entity.inn: entity for entity in (await db_session.scalars(select(LegalEntity))).all()} aliases = (await db_session.scalars(select(LegalEntityAlias))).all() related = (await db_session.scalars(select(RelatedGroup))).all() members = (await db_session.scalars(select(RelatedGroupMember))).all() assert set(groups) == DEV_GROUPS rb = groups["Русский Бисквит"] assert rb.crm_id == "1-AZFL6BQ" and re.fullmatch(r"GK-\d{6}", rb.code) assert offices[rb.base_regional_office_id] == "Санкт-Петербургский РФ" assert all(re.fullmatch(r"GK-\d{6}", group.code) for group in groups.values()) assert len(entities) == DEV_ENTITY_COUNT rb_entity = entities["3528052718"] assert rb_entity.company_group_id == rb.id and rb_entity.regional_office_id == rb.base_regional_office_id assert rb_entity.entity_type is EntityType.ЮЛ and entities["352819549855"].entity_type is EntityType.ИП assert len(aliases) == DEV_ALIAS_COUNT happy = entities["3528104476"] assert {alias.normalized_alias for alias in aliases if alias.legal_entity_id == happy.id} == { "юг финанс пекарня счастливочная", "пекарня счастливочная", } assert {(group.name, group.kind) for group in related} == { ("ГСК Русский Бисквит", RelatedGroupKind.gsk), ("ГСК Унагранде", RelatedGroupKind.gsk), } assert all(re.fullmatch(r"GSK-\d{6}", group.code) for group in related) assert len(members) == DEV_MEMBER_COUNT assert all( member.is_current and member.valid_to is None and member.kind is RelatedGroupKind.gsk for member in members ) gsk_rb = next(group for group in related if group.name == "ГСК Русский Бисквит") assert {member.legal_entity_id for member in members if member.related_group_id == gsk_rb.id} == { rb_entity.id, happy.id, } @pytest.mark.integration async def test_changed_group_office_updates_group_and_keeps_codes(db_session: AsyncSession, tmp_path: Path) -> None: await run_seed(db_session, dev=True) before = {group.name: group.code for group in (await db_session.scalars(select(CompanyGroup))).all()} target = _copy_seed(tmp_path) groups_file = target / "dev" / "company_groups.yaml" text = groups_file.read_text(encoding="utf-8") assert text.count("regional_office: Головной офис") == 1 groups_file.write_text( text.replace("regional_office: Головной офис", "regional_office: Ростовский региональный филиал"), encoding="utf-8", ) report = await run_seed(db_session, dev=True, seed_dir=target) after = {group.name: group.code for group in (await db_session.scalars(select(CompanyGroup))).all()} assert after == before assert report.tables["company_group"].updated == 1 and report.tables["company_group"].inserted == 0 assert report.tables["legal_entity"].updated == UNAGRANDE_ENTITY_COUNT # ЮЛ переехали вслед за ГК assert report.tables["legal_entity_alias"].inserted == 0 and report.tables["related_group_member"].inserted == 0 def test_loader_reads_dev_company_groups() -> None: data = load_seed(dev=True) assert {group.name for group in data.company_groups} == DEV_GROUPS assert {office.name for office in data.regional_offices} == {HEAD_OFFICE, *DEV_BRANCHES} assert {office.name for office in load_seed(dev=True, dev_groups=False).regional_offices} == { HEAD_OFFICE, *DEV_BRANCHES, } - {"Санкт-Петербургский РФ"} assert sum(len(group.legal_entities) for group in data.company_groups) == DEV_ENTITY_COUNT assert ( sum(len(entity.aliases) for group in data.company_groups for entity in group.legal_entities) == DEV_ALIAS_COUNT ) assert sum(len(group.related_groups) for group in data.company_groups) == DEV_RELATED_GROUP_COUNT @pytest.mark.parametrize( ("old", "new", "expected"), [ ( 'members: ["3249004697", "7728757141"]', 'members: ["3249004697", "9999999999"]', "не описаны в legal_entities", ), ('{inn: "2315995868"', '{inn: "231599586"', "ровно 10 цифр"), ( 'aliases: [\'ОБЩЕСТВО С ОГРАНИЧЕННОЙ ОТВЕТСТВЕННОСТЬЮ "АГРОФИРМА "МЕЖЕНИНОВСКАЯ"\']', 'aliases: ["ООО"]', "из одних ОПФ", ), ( ' - {kind: gsk, name: ГСК Унагранде, members: ["3249004697", "7728757141"]}', ' - {kind: gsk, name: ГСК Унагранде, members: ["3249004697", "7728757141"]}\n' ' - {kind: gsk, name: ГСК Унагранде 2, members: ["3249004697"]}', "gsk", ), ], ) def test_validator_rejects_malformed_dev_company_groups(tmp_path: Path, old: str, new: str, expected: str) -> None: target = _copy_seed(tmp_path) groups_file = target / "dev" / "company_groups.yaml" text = groups_file.read_text(encoding="utf-8") assert text.count(old) == 1, old groups_file.write_text(text.replace(old, new), encoding="utf-8") with pytest.raises(SeedError, match=expected): load_seed(target, dev=True) def test_dev_seed_carries_no_invented_kpp() -> None: """В данных seed КПП нет ни у одного ЮЛ (в выгрузке ЕЦП и фикстурах его нет) — значений не выдумываем.""" data = load_seed(dev=True) assert all(entity.kpp is None for group in data.company_groups for entity in group.legal_entities) def test_loader_accepts_optional_kpp_and_normalizes_it(tmp_path: Path) -> None: target = _copy_seed(tmp_path) groups_file = target / "dev" / "company_groups.yaml" text = groups_file.read_text(encoding="utf-8") old = '{inn: "2315995868"' assert text.count(old) == 1 groups_file.write_text(text.replace(old, '{kpp: " 2315 ab 001 ", inn: "2315995868"'), encoding="utf-8") entities = {e.inn: e for g in load_seed(target, dev=True).company_groups for e in g.legal_entities} assert entities["2315995868"].kpp == "2315AB001" @pytest.mark.parametrize("kpp", ["23159500", "A31599500", "231599500Z"]) def test_validator_rejects_malformed_kpp(tmp_path: Path, kpp: str) -> None: target = _copy_seed(tmp_path) groups_file = target / "dev" / "company_groups.yaml" text = groups_file.read_text(encoding="utf-8") old = '{inn: "2315995868"' groups_file.write_text(text.replace(old, '{kpp: "' + kpp + '", inn: "2315995868"'), encoding="utf-8") with pytest.raises(SeedError, match="КПП"): load_seed(target, dev=True) @pytest.mark.integration async def test_seed_writes_kpp_and_keeps_it_when_data_has_none(db_session: AsyncSession, tmp_path: Path) -> None: target = _copy_seed(tmp_path) groups_file = target / "dev" / "company_groups.yaml" text = groups_file.read_text(encoding="utf-8") old = '{inn: "2315995868"' groups_file.write_text(text.replace(old, '{kpp: "231501001", inn: "2315995868"'), encoding="utf-8") await run_seed(db_session, dev=True, seed_dir=target) entity = (await db_session.scalars(select(LegalEntity).where(LegalEntity.inn == "2315995868"))).one() assert entity.kpp == "231501001" groups_file.write_text(text, encoding="utf-8") # в данных КПП убран — seed ранее записанный КПП не стирает await run_seed(db_session, dev=True, seed_dir=target) await db_session.refresh(entity) assert entity.kpp == "231501001"