feat: dev-seed (2 РФ и по пользователю на роль) по флагу --dev, проверка ссылок до записи

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-03 21:41:07 +04:00
parent 0f800a1bd0
commit cb2723c3a2
8 changed files with 486 additions and 11 deletions

View File

@ -1,7 +1,7 @@
"""Загрузчик справочников продукта: `python -m scripts.seed [--dev]` (запуск из `api/`). """Загрузчик справочников продукта: `python -m scripts.seed [--dev]` (запуск из `api/`).
Spec reference-data-seed; design D-4 и решение #18 (R-5): `run_seed(session, *, dev)` не делает commit — коммитит CLI. Spec reference-data-seed; design D-4 и решение #18 (R-5): `run_seed(session, *, dev)` не делает commit — коммитит CLI.
Источник данных — `api/seed/core/*.yaml|csv` (+ `api/seed/dev/` при `--dev`, задача 15). Источник данных — `api/seed/core/*.yaml|csv` (+ `api/seed/dev/` при `--dev`).
Загрузка двухфазная: Загрузка двухфазная:
1. `load_seed` (`seed_lib/loaders.py`) читает ВСЕ файлы в память и проверяет их (обязательные поля, enum'ы из 1. `load_seed` (`seed_lib/loaders.py`) читает ВСЕ файлы в память и проверяет их (обязательные поля, enum'ы из
@ -23,6 +23,13 @@ ASSUMPTION(OQ-46)): `seed_lib/loaders_rules.py` проверяет, что ка
`seed_lib/apply_rules.py` — upsert (составы счетов и чек-лист приводятся к файлу без удалений; ключи без уникального `seed_lib/apply_rules.py` — upsert (составы счетов и чек-лист приводятся к файлу без удалений; ключи без уникального
индекса — в Python, NULL в `account_id` не плодит дубли). Общие помощники чтения — `seed_lib/readers.py`. индекса — в Python, NULL в `account_id` не плодит дубли). Общие помощники чтения — `seed_lib/readers.py`.
Dev-данные (`--dev`, `seed/dev/`, задача 15): 2 региональных филиала (`regional_offices.yaml`, upsert по `name`)
и по одному пользователю на каждую системную роль (`users.yaml`, `<роль>@armka.local`, upsert `app_user` по `email`,
`user_role` по (user_id, role_id)). Ссылки пользователей на роли и филиалы проверяются до записи. Без `--dev` ни
пользователи, ни dev-филиалы не загружаются. ASSUMPTION: dev-данные — только для локальной разработки и тестовых
стендов; в прод `--dev` не запускать (защиты флага нет, спека её не требует). Удалений нет: роль, убранная из
`users.yaml`, остаётся назначенной.
Расширение (задачи 13–15): поле в `SeedData` (`seed_lib/model.py`), читатель в `load_seed`, шаг в `apply_seed`. Расширение (задачи 13–15): поле в `SeedData` (`seed_lib/model.py`), читатель в `load_seed`, шаг в `apply_seed`.
""" """
@ -65,7 +72,11 @@ async def _run_cli(data: SeedData) -> SeedReport:
def main(argv: list[str] | None = None) -> int: def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="python -m scripts.seed", description="Загрузка справочников продукта.") parser = argparse.ArgumentParser(prog="python -m scripts.seed", description="Загрузка справочников продукта.")
parser.add_argument("--dev", action="store_true", help="дополнительно загрузить dev-данные из seed/dev/") parser.add_argument(
"--dev",
action="store_true",
help="дополнительно загрузить dev-данные из seed/dev/ (только для разработки, не для прода)",
)
parser.add_argument("--seed-dir", type=Path, default=DEFAULT_SEED_DIR, help="каталог seed (по умолчанию api/seed)") parser.add_argument("--seed-dir", type=Path, default=DEFAULT_SEED_DIR, help="каталог seed (по умолчанию api/seed)")
args = parser.parse_args(argv) args = parser.parse_args(argv)
try: try:

View File

@ -3,13 +3,13 @@
import uuid import uuid
from typing import Any from typing import Any
from sqlalchemy import func, select from sqlalchemy import func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from scripts.seed_lib.apply_rules import apply_elimination_rules, apply_required_document_rules from scripts.seed_lib.apply_rules import apply_elimination_rules, apply_required_document_rules
from scripts.seed_lib.model import SeedData, SeedReport, TableStat from scripts.seed_lib.model import SeedData, SeedReport, TableStat
from src.models.access import Permission, Role, RolePermission from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
from src.models.accounting import ChartOfAccount from src.models.accounting import ChartOfAccount
from src.models.directories import RegionalOffice from src.models.directories import RegionalOffice
@ -132,6 +132,76 @@ async def _apply_chart_of_accounts(session: AsyncSession, data: SeedData, stat:
await session.execute(statement, rows) await session.execute(statement, rows)
async def _apply_users(session: AsyncSession, data: SeedData, user_stat: TableStat, link_stat: TableStat) -> None:
"""Dev-пользователи: upsert `app_user` по `email`, затем `user_role` по (user_id, role_id).
`created_at` ставится при вставке, `updated_at` — при вставке и при изменении данных (триггера в БД нет);
неизменившиеся строки не трогаются. Назначения ролей только добавляются: роль, убранная из файла, остаётся
назначенной (удалений нет). `external_id` не задаётся (SSO-субъект dev-пользователя неизвестен).
"""
if not data.users:
return
office_ids = {
name: id_ for id_, name in (await session.execute(select(RegionalOffice.id, RegionalOffice.name))).all()
}
role_ids = {code: id_ for id_, code in (await session.execute(select(Role.id, Role.code))).all()}
stored = {
row.email: row
for row in (
await session.execute(
select(
AppUser.id,
AppUser.email,
AppUser.full_name,
AppUser.employee_no,
AppUser.regional_office_id,
AppUser.business_segment,
AppUser.is_active,
)
)
).all()
}
user_ids: dict[str, Any] = {}
for user in data.users:
values = {
"full_name": user.full_name,
"employee_no": user.employee_no,
"regional_office_id": office_ids[user.regional_office],
"business_segment": user.business_segment,
"is_active": user.is_active,
}
row = stored.get(user.email)
if row is None:
user_stat.inserted += 1
user_ids[user.email] = uuid.uuid4()
await session.execute(
pg_insert(AppUser).values(
id=user_ids[user.email], email=user.email, **values, created_at=func.now(), updated_at=func.now()
)
)
continue
user_ids[user.email] = row.id
if any(getattr(row, name) != value for name, value in values.items()):
user_stat.updated += 1
await session.execute(update(AppUser).where(AppUser.id == row.id).values(**values, updated_at=func.now()))
else:
user_stat.unchanged += 1
existing = set((await session.execute(select(UserRole.user_id, UserRole.role_id))).all())
new_rows = []
for user in data.users:
for role_code in user.role_codes:
pair = (user_ids[user.email], role_ids[role_code])
if pair in existing:
link_stat.unchanged += 1
else:
link_stat.inserted += 1
new_rows.append({"id": uuid.uuid4(), "user_id": pair[0], "role_id": pair[1]})
if new_rows:
await session.execute(
pg_insert(UserRole).on_conflict_do_nothing(index_elements=["user_id", "role_id"]), new_rows
)
async def apply_seed(session: AsyncSession, data: SeedData) -> SeedReport: async def apply_seed(session: AsyncSession, data: SeedData) -> SeedReport:
"""Фаза 2: upsert набора в порядке зависимостей. Без commit — транзакцией управляет вызывающий.""" """Фаза 2: upsert набора в порядке зависимостей. Без commit — транзакцией управляет вызывающий."""
report = SeedReport() report = SeedReport()
@ -180,4 +250,6 @@ async def apply_seed(session: AsyncSession, data: SeedData) -> SeedReport:
session, data, report.stat("elimination_rule"), report.stat("elimination_rule_account") session, data, report.stat("elimination_rule"), report.stat("elimination_rule_account")
) )
await apply_required_document_rules(session, data, report.stat("required_document_rule")) await apply_required_document_rules(session, data, report.stat("required_document_rule"))
# dev-пользователи — после ролей и филиалов (в том числе dev-филиалов, они входят в `regional_offices`)
await _apply_users(session, data, report.stat("app_user"), report.stat("user_role"))
return report return report

View File

@ -16,16 +16,19 @@ from scripts.seed_lib.model import (
RoleSeed, RoleSeed,
SeedData, SeedData,
SeedError, SeedError,
UserSeed,
account_codes, account_codes,
) )
from scripts.seed_lib.readers import bool_field, checked_records, ensure_unique, place, read_yaml, str_field from scripts.seed_lib.readers import bool_field, checked_records, ensure_unique, place, read_yaml, str_field
from src.models.enums import AccountKind, AccountRowType, PermissionAction, PermissionObject from src.models.enums import AccountKind, AccountRowType, BusinessSegment, PermissionAction, PermissionObject
DEFAULT_SEED_DIR = Path(__file__).resolve().parent.parent.parent / "seed" DEFAULT_SEED_DIR = Path(__file__).resolve().parent.parent.parent / "seed"
def _load_regional_offices(core: Path, seed_dir: Path) -> list[RegionalOfficeSeed]: def _load_regional_offices(
path = core / "regional_office.yaml" directory: Path, seed_dir: Path, filename: str = "regional_office.yaml"
) -> list[RegionalOfficeSeed]:
path = directory / filename
records = checked_records( records = checked_records(
read_yaml(path, seed_dir), {"name", "is_head_office", "is_active"}, {"name"}, path, seed_dir read_yaml(path, seed_dir), {"name", "is_head_office", "is_active"}, {"name"}, path, seed_dir
) )
@ -246,9 +249,73 @@ def load_seed(seed_dir: Path | None = None, *, dev: bool = False) -> SeedData:
data.elimination_rules = load_elimination_rules(core, seed_dir, codes) data.elimination_rules = load_elimination_rules(core, seed_dir, codes)
data.required_document_rules = load_required_document_rules(core, seed_dir, codes) data.required_document_rules = load_required_document_rules(core, seed_dir, codes)
if dev: if dev:
_load_dev(data, seed_dir / "dev") _load_dev(data, seed_dir / "dev", seed_dir)
return data return data
def _load_dev(data: SeedData, dev_dir: Path) -> None: def _load_dev_regional_offices(data: SeedData, dev_dir: Path, seed_dir: Path) -> None:
"""Заглушка расширения: dev-данные (филиалы, пользователи по ролям) появятся в задаче 15.""" """Dev-филиалы (`dev/regional_offices.yaml`) дописываются к филиалам набора; имена уникальны во всём наборе."""
path = dev_dir / "regional_offices.yaml"
offices = _load_regional_offices(dev_dir, seed_dir, "regional_offices.yaml")
for office in offices:
if office.is_head_office:
raise SeedError(
f"{place(path, seed_dir)}: филиал «{office.name}» объявлен головным офисом, в dev — только РФ"
)
data.regional_offices.extend(offices)
ensure_unique([o.name for o in data.regional_offices], "филиал", path, seed_dir)
def _load_dev_users(data: SeedData, dev_dir: Path, seed_dir: Path) -> None:
"""Dev-пользователи (`dev/users.yaml`): ссылки на роли и филиалы проверяются по набору до любой записи в БД."""
path = dev_dir / "users.yaml"
records = checked_records(
read_yaml(path, seed_dir),
{"email", "full_name", "employee_no", "regional_office", "business_segment", "is_active", "roles"},
{"email", "full_name", "regional_office", "roles"},
path,
seed_dir,
)
role_codes = {r.code for r in data.roles}
office_names = {o.name for o in data.regional_offices}
users = []
for index, record in enumerate(records):
where = place(path, seed_dir, index, str(record.get("email")))
office = str(str_field(record, "regional_office", where))
if office not in office_names:
raise SeedError(
f"{where}: филиал «{office}» не описан ни в core/regional_office.yaml, ни в dev/regional_offices.yaml"
)
roles = record["roles"]
if not isinstance(roles, list) or not all(isinstance(code, str) for code in roles):
raise SeedError(f"{where}: поле «roles» должно быть списком кодов ролей")
if len(roles) != 1:
raise SeedError(f"{where}: у dev-пользователя должна быть ровно одна роль, указано {len(roles)}")
if roles[0] not in role_codes:
raise SeedError(f"{where}: роль «{roles[0]}» не описана в roles.yaml")
segment_text = str_field(record, "business_segment", where, optional=True)
try:
segment = BusinessSegment(segment_text) if segment_text is not None else None
except ValueError:
raise SeedError(
f"{where}: неизвестный business_segment «{segment_text}» (допустимы DKB, DMSB, DM)"
) from None
users.append(
UserSeed(
email=str(str_field(record, "email", where)),
full_name=str(str_field(record, "full_name", where)),
employee_no=str_field(record, "employee_no", where, optional=True),
regional_office=office,
business_segment=segment,
is_active=bool_field(record, "is_active", where, default=True),
role_codes=(roles[0],),
)
)
ensure_unique([u.email for u in users], "email", path, seed_dir)
data.users = users
def _load_dev(data: SeedData, dev_dir: Path, seed_dir: Path) -> None:
"""Dev-данные (задача 15): филиалы, затем пользователи — они ссылаются на филиалы и роли набора."""
_load_dev_regional_offices(data, dev_dir, seed_dir)
_load_dev_users(data, dev_dir, seed_dir)

View File

@ -6,6 +6,7 @@ from dataclasses import dataclass, field
from src.models.enums import ( from src.models.enums import (
AccountKind, AccountKind,
AccountRowType, AccountRowType,
BusinessSegment,
ConsolidationApproach, ConsolidationApproach,
DocumentType, DocumentType,
PermissionAction, PermissionAction,
@ -92,6 +93,19 @@ class RequiredDocumentRuleSeed:
is_mandatory: bool is_mandatory: bool
@dataclass(frozen=True)
class UserSeed:
"""Dev-пользователь; `regional_office` — имя филиала, `role_codes` — коды ролей (в dev-наборе ровно одна)."""
email: str
full_name: str
employee_no: str | None
regional_office: str
business_segment: BusinessSegment | None
is_active: bool
role_codes: tuple[str, ...]
@dataclass @dataclass
class SeedData: class SeedData:
"""Проверенный набор seed; ничего не знает о БД.""" """Проверенный набор seed; ничего не знает о БД."""
@ -105,6 +119,8 @@ class SeedData:
accounts: dict[str, AccountSeed] = field(default_factory=dict) accounts: dict[str, AccountSeed] = field(default_factory=dict)
elimination_rules: list[EliminationRuleSeed] = field(default_factory=list) elimination_rules: list[EliminationRuleSeed] = field(default_factory=list)
required_document_rules: list[RequiredDocumentRuleSeed] = field(default_factory=list) required_document_rules: list[RequiredDocumentRuleSeed] = field(default_factory=list)
#: только при `--dev` (seed/dev/users.yaml); при `--dev` dev-филиалы дописываются в `regional_offices`
users: list[UserSeed] = field(default_factory=list)
def account_codes(data: SeedData) -> frozenset[str]: def account_codes(data: SeedData) -> frozenset[str]:

View File

@ -0,0 +1,9 @@
# DEV-данные (только `python -m scripts.seed --dev`; в прод-режиме не загружаются). Региональные филиалы для
# локальной разработки — названия условные. Головной офис — в core/regional_office.yaml.
# Натуральный ключ upsert — name; удаление seed не выполняет.
- name: Ростовский региональный филиал
is_head_office: false
is_active: true
- name: Краснодарский региональный филиал
is_head_office: false
is_active: true

58
api/seed/dev/users.yaml Normal file
View File

@ -0,0 +1,58 @@
# DEV-данные (только `python -m scripts.seed --dev`): по одному пользователю на каждую системную роль ТЗ 8.1.
# Email — `<код роли в нижнем регистре>@armka.local`; dev-вход идёт по email (JWT РАИСА в dev подменяется),
# поэтому external_id не задаётся (NULL). Натуральный ключ upsert — email.
# regional_office — ИМЯ филиала: ГО — из core/regional_office.yaml, РФ — из dev/regional_offices.yaml.
# Раскладка: роли ГО (KA_GO, CURATOR_GO, EXPERT, ADMIN) — в Головном офисе; роли РФ (KA_RF, RKA, CLIENT_MANAGER) —
# в РФ: КА и РКА в Ростовском (один РФ, чтобы проходить маршрут «КА РФ -> РКА РФ»), КМ — в Краснодарском
# (проверка границ «свой РФ / чужой РФ» policy-слоя).
# Роли пользователя в файле приводятся к строке user_role добавлением; снятые из файла роли seed НЕ удаляет
# (смена роли в файле — удалить старую связь вручную через UI администратора).
- email: ka_rf@armka.local
full_name: "Dev: Аналитик РФ"
employee_no: DEV-001
regional_office: Ростовский региональный филиал
business_segment: DKB
is_active: true
roles: [KA_RF]
- email: ka_go@armka.local
full_name: "Dev: Аналитик ГО"
employee_no: DEV-002
regional_office: Головной офис
business_segment: DKB
is_active: true
roles: [KA_GO]
- email: rka@armka.local
full_name: "Dev: Руководитель аналитика РФ"
employee_no: DEV-003
regional_office: Ростовский региональный филиал
business_segment: DKB
is_active: true
roles: [RKA]
- email: client_manager@armka.local
full_name: "Dev: Клиентский менеджер"
employee_no: DEV-004
regional_office: Краснодарский региональный филиал
business_segment: DKB
is_active: true
roles: [CLIENT_MANAGER]
- email: curator_go@armka.local
full_name: "Dev: Куратор КП ГО"
employee_no: DEV-005
regional_office: Головной офис
business_segment: DKB
is_active: true
roles: [CURATOR_GO]
- email: expert@armka.local
full_name: "Dev: Эксперт"
employee_no: DEV-006
regional_office: Головной офис
business_segment: DKB
is_active: true
roles: [EXPERT]
- email: admin@armka.local
full_name: "Dev: Администратор"
employee_no: DEV-007
regional_office: Головной офис
business_segment: DKB
is_active: true
roles: [ADMIN]

242
api/tests/test_seed_dev.py Normal file
View File

@ -0,0 +1,242 @@
"""Dev-seed (spec reference-data-seed, требование «Dev-данные только по явному флагу», задача 15)."""
import shutil
from pathlib import Path
import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from scripts.seed import SeedError, load_seed, run_seed
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.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"):
stat = report.tables[table]
assert (stat.inserted, stat.updated) == (0, 0)
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
@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)

View File

@ -248,7 +248,7 @@ async def test_without_dev_flag_no_users_are_loaded(db_session: AsyncSession) ->
@pytest.mark.integration @pytest.mark.integration
async def test_dev_flag_is_accepted(db_session: AsyncSession) -> None: async def test_dev_flag_is_accepted(db_session: AsyncSession) -> None:
"""Каркас --dev: до задачи 15 dev-данных нет, но флаг не должен падать.""" """Флаг --dev не ломает загрузку core-справочников (dev-данные проверяет test_seed_dev.py)."""
await run_seed(db_session, dev=True) await run_seed(db_session, dev=True)
assert await _count(db_session, Role) == ROLE_COUNT assert await _count(db_session, Role) == ROLE_COUNT