arm-core/api/tests/test_seed_dev.py

"""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"