Andrey Epifancev a04d9db478 chore(directories): отметить Task 8 (2.5) и завершение фазы 1 в плане и tasks.md

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2

2026-10-04 22:08:04 +04:00

50 KiB

Raw Permalink Blame History

Срез 1 — справочники arm-core — план исполнения, фаза 1 (модель v5.14, инфраструктура запросов, policy, API справочников)

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Branch: feat/5-slice-1-directories (worktree .wt/slice-1/arm-core). Issue: RSHB/arm-core#5. PR: RSHB/arm-core#6 (открыт, спека). Spec: openspec/changes/slice-1-directories/ — proposal.md, design.md (D-1…D-11), specs/{directories-api,directory-import,directory-policy,users-admin,data-model,audit-trail}/spec.md. Правила предметной области: ../docs/context/05-domain-rules.md §3 (нормализация алиаса), §10/§10.1 (права и видимость), §13 (импорт и ручные операции). Модель: ../docs/model/data-model-v5.14.dbml. Спека — единственный источник истины при рулингах; план сгенерирован из tasks.md change'а (номера в скобках — номера задач tasks.md; по завершении задачи отметить - [x] и там).

Goal: после фазы 1 ядро умеет вести ГК, ЮЛ/ИП с историей и алиасами, ГСК/ГСЗ с составом через /api/v1/directories/* с правами, политикой видимости по филиалу и аудитом; схема БД — DBML v5.14 (миграция 0002).

Architecture: FastAPI-эндпоинты → сервисы (одна транзакция на запрос: сессия коммитится при успехе) → репозитории с фильтром видимости → ORM. Policy — чистые функции src/domain/policy.py над Select. Все мутации пишут историю (SCD4/SCD2) и audit_event через AuditService в той же транзакции. Доменные ошибки → {"detail": {"code", "message", …}} с 404/409/422.

Tech Stack: Python 3.11, FastAPI, SQLAlchemy 2 async, Alembic, PostgreSQL 16, pydantic v2, pytest-asyncio, httpx.

Global Constraints

Имена таблиц/колонок/enum'ов — ровно как в ../docs/model/data-model-v5.14.dbml (43 таблицы, 27 enum, схема dkb); полей вне DBML нет. related_group_member.kind NOT NULL и равен related_group.kind группы (проверяет сервис). Новые значения audit_action: LEGAL_ENTITY_CHANGE (после GROUP_CHANGE), USER_CHANGE (после USER_ROLE_CHANGE) — порядок как в DBML (сверка enum'ов сравнивает упорядоченный список).
Права — данные: маршруты защищены require("<CODE>"), код роли нигде не читается. Policy (§10.1): ГО (regional_office.is_head_office) видит всё; филиал — ГК с base_regional_office_id своего филиала, их ЮЛ/ГСК и монозаёмщиков своего филиала; business_segment не используется (ASSUMPTION(OQ-86)). Объект вне видимости → 404 с тем же телом, что и несуществующий.
Все мутации: одна транзакция, audit_event (user_id инициатора, before_value/after_value, ip_address, user_agent) через AuditService.record — append-only (тест test_audit_service.py сканирует src/ на UPDATE/DELETE по AuditEvent; не добавлять).
Ошибки: {"detail": {"code": "...", "message": "...по-русски"}} для доменных 404/409/422; существующие ответы Среза 0 (401/403 строкой) не меняются. Списки — {items, total, limit, offset}, limit ≤ 200 (по умолчанию 50), offset ≥ 0, сортировка по имени.
Коды ГК/ГСК/ГСЗ — GK-NNNNNN / GSK-NNNNNN / GSZ-NNNNNN (6 цифр с нулями) из последовательностей PostgreSQL dkb.company_group_code_seq, dkb.related_group_code_seq (D-5).
История (D-6): при изменении ГК — снимок прежних значений в company_group_history (valid_from прежний, valid_to = now), у ГК valid_from = now; ЮЛ — то же в legal_entity_history с event_type; при создании история не пишется. Членство ГСК/ГСЗ — SCD2: исключение = valid_to = now, is_current = false; повторное включение — новая строка.
Деньги не затрагиваются; Decimal в домене, никаких float.
Линтеры: ruff (line-length 120, E,F,I,UP,B,PL — PLR0913 > 5 аргументов решается # noqa: PLR0913 - причина или dataclass'ом параметров; PLR2004 — именованные константы) + mypy (src.models.*, src.domain.* — strict-флаги). Тесты: pytest-asyncio auto, маркер integration для БД, fixtures для реальных данных; БД-тесты — на db_session (SAVEPOINT, откат после теста).
Все команды — из api/ (make — из корня репозитория); Python 3.11; venv api/.venv (make venv идемпотентен). PostgreSQL из compose ../docs/dev уже поднят (make up), контейнеры не перезапускать.
Язык комментариев/коммитов — русский, идентификаторы — английские. Коммит заканчивается строками Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> (или фактическая модель исполнителя) и Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2. Не пушить, не создавать PR, не запускать openspec archive.
Допущения в коде — комментарием # ASSUMPTION(OQ-nn): OQ-34 (удаление только пустой ГСК/ГСЗ), OQ-86 (ЗО не фильтрует), OQ-88 (ИНН — только длина), OQ-21 (effective_date ≥ сегодня).

Review Focus

Случаи, которых спека не называет явно, но которые ударят по пользователю; тест на каждый добавлен в задачу-владельца:

PATCH без изменений (тело с теми же значениями) — не должен плодить строку истории и аудит: Task 5 (ГК) и Task 6 (ЮЛ) — тест «no-op PATCH не пишет историю».
Фильтр inn с пробелами / ИНН с ведущими нулями — inn хранится строкой, сравнение точное после strip(); ИНН 0123456789 валиден: Task 6 — тест.
Пользователь филиала переводит ЮЛ в ГК чужого филиала — целевая ГК вне scope → 404 (не 422, не утечка): Task 6 — тест.
Добавление в ГСК закрытого ЮЛ (close_date заполнен, company_group_id = NULL) — 422 DIFFERENT_COMPANY_GROUP (ЮЛ без ГК не в той же ГК): Task 8 — тест.
offset больше total — пустой items, total настоящий, 200: Task 5 — тест списка ГК.

Задачи

Task 1: [1.1] ORM v5.14, миграция `0002_dbml_v5_14`, версия DBML

Files:

Modify: api/src/models/directories.py (RelatedGroupMember, последовательности), api/src/models/enums.py (AuditAction), api/src/models/__init__.py:10 (DBML_VERSION), api/src/models/base.py:9 (докстринг v5.14)
Create: api/alembic/versions/0002_dbml_v5_14.py
Modify: Makefile:6 (DBML ?= ../../docs/model/data-model-v5.14.dbml), api/scripts/dbml_parity.py:59 (DBML_FILE_NAME = "data-model-v5.14.dbml"), api/tests/test_migrations.py:31 (HEAD = "0002_dbml_v5_14"), api/tests/test_enums.py:18 ("5.14"), api/tests/test_check_dbml_script.py:145-154 (фиктивная версия v5.15, сообщение версия DBML 5.15 не совпадает с версией ORM 5.14), api/tests/test_models_part_01_03.py::_check_related_group_member_uniqueness
Test: api/tests/test_models_part_01_03.py, api/tests/test_migrations.py

Interfaces:

Produces: RelatedGroupMember.kind: Mapped[RelatedGroupKind] (колонка между legal_entity_id и valid_from, как в DBML); COMPANY_GROUP_CODE_SEQ = Sequence("company_group_code_seq", schema=SCHEMA, metadata=Base.metadata), RELATED_GROUP_CODE_SEQ = Sequence("related_group_code_seq", …) в src/models/directories.py; AuditAction.LEGAL_ENTITY_CHANGE, AuditAction.USER_CHANGE.
Step 1: Обновить тест уникальности членства (падает на старой схеме)

В _check_related_group_member_uniqueness (tests/test_models_part_01_03.py): INSERT в related_group_member получает колонку kind; сценарий: g1 (gsk) текущее — ок; g3 (gsz, создать третью группу) текущее для того же ЮЛ — ок (v5.14); g2 (gsk) текущее — IntegrityError; закрытая версия в g2 — ок. Докстринг: «ЮЛ — в одной ГСК и одной ГСЗ одновременно (v5.14, решение владельца 04.10.2026)».

Step 2: Запустить — убедиться, что падает

Run: cd api && .venv/bin/pytest -m integration tests/test_models_part_01_03.py -k related_group_member -v Expected: FAIL (column "kind" does not exist или IntegrityError на gsz).

Step 3: ORM

src/models/directories.py:

from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Index, Sequence, String, text
from .base import SCHEMA, Base, UUIDPrimaryMixin

#: Последовательности кодов ГК и ГСК/ГСЗ (design D-5): `GK-000001`, `GSK-000001`/`GSZ-000001`. В DBML не описываются
#: (не таблицы); создаются миграцией 0002 и `Base.metadata.create_all` (привязаны к metadata).
COMPANY_GROUP_CODE_SEQ = Sequence("company_group_code_seq", schema=SCHEMA, metadata=Base.metadata)
RELATED_GROUP_CODE_SEQ = Sequence("related_group_code_seq", schema=SCHEMA, metadata=Base.metadata)

В RelatedGroupMember.__table_args__ заменить второй индекс:

        # (legal_entity_id, kind): partial UNIQUE WHERE is_current — v5.14: ЮЛ в одной ГСК И одной ГСЗ одновременно
        Index(
            "uq_related_group_member_entity_kind_current",
            "legal_entity_id",
            "kind",
            unique=True,
            postgresql_where=text("is_current"),
        ),

и добавить колонку после legal_entity_id:

    kind: Mapped[RelatedGroupKind] = mapped_column(pg_enum("related_group_kind"), nullable=False)

src/models/enums.py: LEGAL_ENTITY_CHANGE = "LEGAL_ENTITY_CHANGE" сразу после GROUP_CHANGE, USER_CHANGE = "USER_CHANGE" сразу после USER_ROLE_CHANGE. src/models/__init__.py: DBML_VERSION = "5.14". Докстринги модулей models/*.py, scripts/check_dbml_vs_orm.py: «v5.13» → «v5.14» там, где речь о текущей версии.

Step 4: Миграция 0002

api/alembic/versions/0002_dbml_v5_14.py:

"""DBML v5.14: related_group_member.kind + partial unique (legal_entity_id, kind), audit_action += LEGAL_ENTITY_CHANGE,
USER_CHANGE, последовательности кодов ГК/ГСК (design D-1, D-5).

`ALTER TYPE … ADD VALUE` необратим — downgrade пересоздаёт тип audit_action со значениями v5.13 (строк с новыми
значениями в audit_event быть не должно; иначе downgrade падает на USING — это осознанно).

Revision ID: 0002_dbml_v5_14
Revises: 0001_dbml_v5_13
Create Date: 2026-10-04
"""

from collections.abc import Sequence

import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from sqlalchemy.schema import CreateSequence, DropSequence

from alembic import op

revision: str = "0002_dbml_v5_14"
down_revision: str | Sequence[str] | None = "0001_dbml_v5_13"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

SCHEMA = "dkb"
TABLE = "related_group_member"
OLD_INDEX = "uq_related_group_member_entity_current"
NEW_INDEX = "uq_related_group_member_entity_kind_current"
SEQUENCES = ("company_group_code_seq", "related_group_code_seq")
AUDIT_ACTION_V5_13 = (
    "LOGIN", "LOGIN_FAILED", "LOGOUT", "CONSOLIDATION_CREATE", "CONSOLIDATION_STATUS_CHANGE", "CONSOLIDATION_APPROVE",
    "CONSOLIDATION_VERIFY", "GROUP_CHANGE", "MEMBERSHIP_CHANGE", "ASSIGNMENT_CHANGE", "FILE_UPLOAD", "FILE_DELETE",
    "FILE_REPLACE", "DOCUMENT_CONFIRM", "CONSOLIDATION_LOCK", "CONSOLIDATION_LOCK_TAKEOVER", "DISCLOSURE_EDIT",
    "RECOGNIZED_DATA_EDIT", "OSV_COUNTERPARTY_MAP", "OSV_TYPE_OVERRIDE", "CORRECTION_CREATE", "CORRECTION_EDIT",
    "CORRECTION_REVOKE", "GOODWILL_EDIT", "EXPORT", "DIRECTORY_IMPORT", "USER_ROLE_CHANGE", "ROLE_PERMISSION_CHANGE",
)


def upgrade() -> None:
    kind = postgresql.ENUM(name="related_group_kind", schema=SCHEMA, create_type=False)
    op.add_column(TABLE, sa.Column("kind", kind, nullable=True), schema=SCHEMA)
    op.execute(
        f"UPDATE {SCHEMA}.{TABLE} m SET kind = g.kind FROM {SCHEMA}.related_group g WHERE g.id = m.related_group_id"
    )
    op.alter_column(TABLE, "kind", nullable=False, schema=SCHEMA)
    op.drop_index(OLD_INDEX, table_name=TABLE, schema=SCHEMA)
    op.create_index(
        NEW_INDEX, TABLE, ["legal_entity_id", "kind"], unique=True, schema=SCHEMA, postgresql_where=sa.text("is_current")
    )
    # порядок значений — как в DBML v5.14 (сверка enum'ов сравнивает упорядоченные списки)
    op.execute(f"ALTER TYPE {SCHEMA}.audit_action ADD VALUE IF NOT EXISTS 'LEGAL_ENTITY_CHANGE' AFTER 'GROUP_CHANGE'")
    op.execute(f"ALTER TYPE {SCHEMA}.audit_action ADD VALUE IF NOT EXISTS 'USER_CHANGE' AFTER 'USER_ROLE_CHANGE'")
    for name in SEQUENCES:
        op.execute(CreateSequence(sa.Sequence(name, schema=SCHEMA)))


def downgrade() -> None:
    for name in reversed(SEQUENCES):
        op.execute(DropSequence(sa.Sequence(name, schema=SCHEMA)))
    op.drop_index(NEW_INDEX, table_name=TABLE, schema=SCHEMA)
    op.create_index(OLD_INDEX, TABLE, ["legal_entity_id"], unique=True, schema=SCHEMA, postgresql_where=sa.text("is_current"))
    op.drop_column(TABLE, "kind", schema=SCHEMA)
    op.execute(f"ALTER TYPE {SCHEMA}.audit_action RENAME TO audit_action_old")
    postgresql.ENUM(*AUDIT_ACTION_V5_13, name="audit_action", schema=SCHEMA).create(op.get_bind(), checkfirst=False)
    op.execute(
        f"ALTER TABLE {SCHEMA}.audit_event ALTER COLUMN action TYPE {SCHEMA}.audit_action "
        f"USING action::text::{SCHEMA}.audit_action"
    )
    op.execute(f"DROP TYPE {SCHEMA}.audit_action_old")

alembic/env.py менять не нужно. В tests/test_migrations.py: HEAD = "0002_dbml_v5_14"; добавить тест test_downgrade_one_step (сценарий спеки «Откат на один шаг»): upgrade head → downgrade 0001_dbml_v5_13 → в information_schema.columns нет related_group_member.kind, в pg_indexes есть uq_related_group_member_entity_current, enum_range(NULL::dkb.audit_action) без LEGAL_ENTITY_CHANGE → upgrade head снова. Если test_migrated_schema_equals_orm_create_all сравнивает последовательности — они теперь есть и в metadata, и в миграции.

Step 5: Пути и версии

Makefile:6 → data-model-v5.14.dbml; scripts/dbml_parity.py:59 → "data-model-v5.14.dbml"; tests/test_enums.py → "5.14"; tests/test_check_dbml_script.py — фиктивный файл v5.15, ожидаемое сообщение с 5.15/5.14.

Step 6: Проверить

Run: cd api && .venv/bin/ruff check . && .venv/bin/ruff format --check . && .venv/bin/mypy src scripts && .venv/bin/pytest -m "not integration" → PASS. Run: make dbml-check (из корня) → пустой вывод, код 0. Run: cd api && .venv/bin/pytest -m integration tests/test_migrations.py tests/test_models_part_01_03.py -v → PASS (цикл 0001→0002→base→head, пустой autogenerate, снимок схемы = create_all, откат на шаг, уникальность gsk+gsz).

Step 7: Commit

git add -A && git commit -m "feat(model): DBML v5.14 — related_group_member.kind, audit_action +2, миграция 0002, последовательности кодов" (+ трейлеры).

Task 2: [1.2] Единица работы, аудит с контекстом запроса, доменные исключения, пагинация

Files:

Modify: api/src/db/db.py (get_async_session — commit при успехе, см. R-1 ниже)
Create: api/src/api/dependencies/audit.py, api/src/schemas/common.py
Modify: api/src/common/exceptions.py, api/src/core/exception_handlers.py
Test: api/tests/test_uow_and_errors.py (unit + integration)

Interfaces:

Produces:
- get_async_session() — та же зависимость, что в Срезе 0, но тело async with async_session_maker() as session: async with session.begin(): yield session — commit при нормальном завершении запроса, rollback при любом исключении (в т.ч. HTTPException, доменных). R-1 (design #12): вместо отдельной get_uow_session из D-2 используется одна сессия на запрос — current_user и мутации работают в одной транзакции и одном соединении; цена ошибки — read-only маршруты коммитят пустую транзакцию (безвредно).
- class DomainError(Exception): поля status: int, code: str, message: str, extra: dict[str, Any]; подклассы NotFoundError(code="NOT_FOUND", message=…) (404), ConflictError(code, message, **extra) (409), DomainValidationError(code, message, **extra) (422). Конструкторы: NotFoundError(message: str = "объект не найден"), ConflictError(code: str, message: str, **extra: Any), DomainValidationError(code: str, message: str, **extra: Any). Старый CouldntCreateError удалить (не используется).
- Обработчик DomainError → JSONResponse(status, {"detail": {"code": code, "message": message, **extra}}) (значения extra через jsonable_encoder).
- RequestAudit (src/api/dependencies/audit.py): @dataclass(frozen=True) class RequestAudit: service: AuditService; user_id: uuid.UUID; ip_address: str | None; user_agent: str | None с методом async def record(self, *, action, section, object_type, object_id=None, description=None, before_value=None, after_value=None, comment=None, result=None, changed_field=None) -> AuditEvent (делегирует AuditService.record, подставляя user_id/ip_address/user_agent). Зависимость async def get_request_audit(request: Request, session: Annotated[AsyncSession, Depends(get_async_session)], principal: Annotated[Principal, Depends(current_user)]) -> RequestAudit (ip_address = request.client.host if request.client else None, user_agent = request.headers.get("user-agent")).
- src/schemas/common.py: class Page(BaseModel, Generic[T]): items: list[T]; total: int; limit: int; offset: int; class PageParams(BaseModel): limit: int = Field(50, ge=1, le=200); offset: int = Field(0, ge=0) — используется как Annotated[PageParams, Query()]; class ErrorDetail(BaseModel): code: str; message: str (для OpenAPI responses).
Step 1: Тесты

tests/test_uow_and_errors.py:

unit: мини-FastAPI() с register_exception_handlers, маршруты бросают NotFoundError("ГК не найдена"), ConflictError("INN_EXISTS", "ИНН уже есть", legal_entity_id=uuid), DomainValidationError("INVALID_INN", "…") → 404/409/422 и тело {"detail": {"code", "message", ...}}; HTTPException(403, "x") по-прежнему {"detail": "x"}.
integration (scratch_db_url + alembic upgrade head через tests.pg_helpers.alembic_async, затем свой async_sessionmaker на этот URL с переопределением src.db.db.async_session_maker через monkeypatch): приложение с двумя маршрутами на Depends(get_async_session) — POST /ok вставляет regional_office и возвращает 201; POST /fail вставляет и бросает ConflictError. После /ok строка видна из другой сессии; после /fail — нет. Это единственный тест, который проверяет реальный commit (SAVEPOINT-сессия тестов его не ловит).
Page[RegionalOfficeOut] сериализуется; PageParams(limit=201) → ValidationError.
Step 2: Запустить — падает (нет модулей/обработчика)
Step 3: Реализация по интерфейсам выше. В exception_handlers.py обработчик DomainError регистрируется до Exception. src/db/db.py:

async def get_async_session() -> AsyncIterator[AsyncSession]:
    """Сессия запроса = одна транзакция: commit при успешном ответе, rollback при любом исключении (R-1, design #12)."""
    async with async_session_maker() as session, session.begin():
        yield session

Проверить, что src/api/endpoints/dev_auth.py и обмен (services/recognition_exchange.py) не используют get_async_session с собственным commit() (они работают через async_session_maker напрямую — не трогать).

Step 4: Тесты зелёные: pytest tests/test_uow_and_errors.py -v, затем pytest -m "not integration" и pytest -m integration tests/test_me_and_dev_token.py tests/test_auth_deps.py (регрессия Среза 0).
Step 5: Commit feat(api): транзакция на запрос, RequestAudit, доменные ошибки 404/409/422, Page.

Task 3: [1.3] Фабрики справочников и `auth_client` в conftest

Files:

Modify: api/tests/factories.py, api/tests/conftest.py, api/tests/test_me_and_dev_token.py (перевести на общие фикстуры, поведение тестов не менять)

Interfaces:

Produces (tests/factories.py, все flush, без commit):
- async def make_regional_office(session, *, name: str | None = None, is_head_office: bool = False) -> RegionalOffice
- async def make_company_group(session, *, name="ГК Тест", office: RegionalOffice | None = None, crm_id: str | None = None, code: str | None = None) -> CompanyGroup (code по умолчанию f"GK-T{uuid4().hex[:6]}", valid_from=now)
- async def make_legal_entity(session, *, name="ООО Тест", inn: str | None = None, entity_type=EntityType.ЮЛ, group: CompanyGroup | None = None, office: RegionalOffice | None = None, close_date=None) -> LegalEntity (inn по умолчанию 10 случайных цифр; regional_office_id = офис группы, если не задан)
- async def make_related_group(session, *, group: CompanyGroup, kind=RelatedGroupKind.gsk, name="ГСК Тест") -> RelatedGroup
- async def make_membership(session, *, related_group: RelatedGroup, entity: LegalEntity, is_current=True) -> RelatedGroupMember (kind = related_group.kind)
- async def make_user(session, *, email: str, office: RegionalOffice, roles: Sequence[str] = (), full_name="Тест Тестов", is_active=True) -> AppUser (роли — по Role.code из seed; UserRole строки)
Produces (tests/conftest.py):
- seeded(db_session) — await run_seed(db_session, dev=True); await db_session.flush().
- app(db_session) -> FastAPI — build_api_router(settings) с DEBUG=True, AUTH_DEV_ISSUER=True, register_exception_handlers, app.state.token_verifier = build_verifier(settings), app.state.storage = None, override get_async_session:
```
async def override_session() -> AsyncIterator[AsyncSession]:
    async with db_session.begin_nested():  # как транзакция запроса: ошибка → откат до SAVEPOINT
        yield db_session
```
- auth_client(app, seeded) -> AuthClientFactory где AuthClientFactory = Callable[[str], Awaitable[httpx.AsyncClient]]: по email запрашивает POST /api/v1/dev/token, возвращает клиент с headers={"Authorization": f"Bearer {token}"} и base_url="http://test/api/v1"; клиенты закрываются в teardown (AsyncExitStack). Dev-пользователи seed: ka_rf@armka.local (Ростовский РФ), rka@armka.local (Ростовский РФ), client_manager@armka.local (Краснодарский РФ), ka_go@armka.local, curator_go@armka.local, expert@armka.local, admin@armka.local (ГО).
Step 1: написать фабрики и фикстуры; test_me_and_dev_token.py перевести на app/auth_client там, где он строит приложение сам (тесты сочетаний флагов DEBUG/AUTH_DEV_ISSUER оставить на локальном _build_app).
Step 2: pytest -m integration tests/test_me_and_dev_token.py -v → PASS; ruff, mypy чисты.
Step 3: Commit test: фабрики справочников, фикстуры app/auth_client/seeded.

Task 4: [2.1] Policy видимости — `src/domain/policy.py`

Files:

Create: api/src/domain/__init__.py (докстринг: «чистые функции по §§ правил, без I/O»), api/src/domain/policy.py
Test: api/tests/test_domain_policy.py (без БД)

Interfaces:

Produces (src/domain/policy.py, §10.1):

@dataclass(frozen=True)
class VisibilityScope:
    """Область видимости справочников: `office_id is None` — вся система (ГО)."""
    office_id: uuid.UUID | None

    @property
    def is_global(self) -> bool: ...

def scope_for(*, is_head_office: bool, regional_office_id: uuid.UUID) -> VisibilityScope
def scope_of(principal: "UserWithAccess") -> VisibilityScope   # обёртка: scope_for(is_head_office=principal.regional_office.is_head_office, regional_office_id=principal.user.regional_office_id)
def company_groups_visible(stmt: Select[Any], scope: VisibilityScope) -> Select[Any]     # WHERE CompanyGroup.base_regional_office_id == office
def legal_entities_visible(stmt: Select[Any], scope: VisibilityScope) -> Select[Any]     # outerjoin CompanyGroup; WHERE group.base_office == office OR (le.company_group_id IS NULL AND le.regional_office_id == office)
def company_group_is_visible(group: CompanyGroup, scope: VisibilityScope) -> bool
def legal_entity_is_visible(entity: LegalEntity, group_office_id: uuid.UUID | None, scope: VisibilityScope) -> bool
def office_for_create(scope: VisibilityScope, requested: uuid.UUID | None) -> uuid.UUID | None
    # ГО: requested как есть; филиал: None → свой офис; чужой → DomainValidationError("OFFICE_NOT_ALLOWED", "ГК можно закрепить только за своим филиалом")

scope_of импортирует UserWithAccess только под TYPE_CHECKING, чтобы domain/ не тянул репозитории.

Step 1: Тесты (tests/test_domain_policy.py, без БД): scope_for(is_head_office=True, …).is_global; филиал → office_id; company_groups_visible для global возвращает тот же stmt (compile без WHERE), для филиала — str(stmt) содержит base_regional_office_id; legal_entities_visible содержит OR и IS NULL; company_group_is_visible — 4 комбинации; legal_entity_is_visible — монозаёмщик своего/чужого филиала, ЮЛ в ГК своего/чужого; office_for_create — 4 ветви, DomainValidationError.code == "OFFICE_NOT_ALLOWED".
Step 2: падает → Step 3: реализация → Step 4: pytest tests/test_domain_policy.py -v, mypy src (strict для src.domain.*) → PASS.
Step 5: Commit feat(domain): policy видимости справочников (§10.1, ASSUMPTION(OQ-86)).

Task 5: [2.2] Филиалы и ГК: репозиторий, сервис, схемы, эндпоинты

Files:

Create: api/src/repositories/directories.py, api/src/services/company_groups.py, api/src/schemas/directories.py, api/src/api/endpoints/directories/__init__.py, api/src/api/endpoints/directories/regional_offices.py, api/src/api/endpoints/directories/company_groups.py, api/src/api/dependencies/directories.py
Modify: api/src/api/router.py (include directories.router — префикс /directories, теги)
Test: api/tests/test_api_company_groups.py, api/tests/test_api_policy.py

Interfaces:

src/api/dependencies/directories.py:
- def get_scope(principal: Annotated[Principal, Depends(current_user)]) -> VisibilityScope (= scope_of(principal))
- def get_company_group_service(session, audit: Depends(get_request_audit), scope: Depends(get_scope)) -> CompanyGroupService
src/repositories/directories.py (функции, как repositories/users.py):
- async def next_code(session, seq: Sequence, prefix: str) -> str → f"{prefix}-{n:06d}" через select(seq.next_value())
- async def list_company_groups(session, scope, *, q: str | None, is_archived: bool | None, limit, offset) -> tuple[list[tuple[CompanyGroup, int]], int] (ГК + legal_entities_count через подзапрос count(legal_entity.id), total отдельным count(); сортировка CompanyGroup.name)
- async def get_company_group(session, scope, group_id) -> CompanyGroup | None (None и для невидимой)
- async def list_company_group_history(session, group_id) -> list[CompanyGroupHistory]
- async def get_regional_office(session, office_id) -> RegionalOffice | None, async def list_regional_offices(session) -> list[RegionalOffice]
src/services/company_groups.py:

class CompanyGroupService:
    def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None
    async def create(self, data: CompanyGroupCreate) -> CompanyGroup        # code из COMPANY_GROUP_CODE_SEQ; office_for_create; crm_id pre-check → ConflictError("CRM_ID_EXISTS", …); valid_from=created_at=now; updated_by=user; audit GROUP_CHANGE after_value={code,name,crm_id,base_regional_office_id}
    async def update(self, group_id: uuid.UUID, data: CompanyGroupUpdate) -> CompanyGroup  # NotFoundError; diff полей; нет изменений → вернуть без истории/аудита; иначе снимок CompanyGroupHistory(prev values, valid_from=group.valid_from, valid_to=now, changed_by=user), group.valid_from=now, updated_*; action = ASSIGNMENT_CHANGE если менялся base_regional_office_id, иначе GROUP_CHANGE; changed_field — имя поля при одном изменении; before/after — только изменённые поля
    async def set_archived(self, group_id: uuid.UUID, archived: bool) -> CompanyGroup  # GROUP_CHANGE changed_field="is_archived"
    async def get(self, group_id) -> CompanyGroup        # NotFoundError("ГК не найдена")
    async def history(self, group_id) -> list[CompanyGroupHistory]
    async def page(self, params: CompanyGroupFilter, page: PageParams) -> tuple[list[tuple[CompanyGroup, int]], int]

Константа сообщения COMPANY_GROUP_NOT_FOUND = "ГК не найдена" — одно и то же для чужой и несуществующей (spec directory-policy).

src/schemas/directories.py: RegionalOfficeOut(id, name, is_head_office, is_active); CompanyGroupCreate(name: str = Field(min_length=1, max_length=500), crm_id: str | None = None, base_regional_office_id: uuid.UUID | None = None); CompanyGroupUpdate — те же поля, все Optional, model_config = ConfigDict(extra="forbid"), семантика «поле передано» через model_fields_set; CompanyGroupOut(id, code, crm_id, name, base_regional_office_id, is_archived, valid_from, legal_entities_count: int = 0, created_at, updated_at); CompanyGroupHistoryOut(id, crm_id, name, base_regional_office_id, valid_from, valid_to, changed_by); CompanyGroupFilter(q: str | None = None, is_archived: bool | None = None).
Эндпоинты (/api/v1/directories/..., все с Depends(require("DIRECTORY_VIEW"|"DIRECTORY_EDIT"))): GET /regional-offices → list[RegionalOfficeOut]; GET /company-groups → Page[CompanyGroupOut]; POST /company-groups → 201; GET /company-groups/{id}; PATCH /company-groups/{id}; POST /company-groups/{id}/archive, /unarchive; GET /company-groups/{id}/history.
Step 1: Тесты (tests/test_api_company_groups.py, @pytest.mark.integration, auth_client):
1. ka_go создаёт две ГК → коды GK- + 6 цифр, различаются; в audit_event две GROUP_CHANGE с after_value["code"].
2. PATCH base_regional_office_id (на Краснодарский РФ по id из /regional-offices) → company_group_history 1 строка с прежним филиалом и valid_to; valid_from ГК изменился; audit_event(ASSIGNMENT_CHANGE) с before_value/after_value.
3. Дубликат crm_id → 409 {"detail": {"code": "CRM_ID_EXISTS", ...}}.
4. no-op PATCH (то же имя) → 200, история пуста, аудита нет (Review Focus 1).
5. expert (DIRECTORY_VIEW только) → POST 403, аудита нет.
6. Список: q по подстроке имени без регистра, is_archived, offset > total → items == [], total верный (Review Focus 5); limit=201 → 422.
7. archive/unarchive переключают is_archived, пишут GROUP_CHANGE changed_field="is_archived". tests/test_api_policy.py (сценарии spec directory-policy):
8. ГК А (Ростовский РФ) и ГК Б (Краснодарский РФ) через фабрики; ka_rf видит только А (total == 1); ka_go — обе.
9. ka_rf GET /company-groups/{Б} → 404, тело совпадает с GET /company-groups/{uuid4()}.
10. ka_rf создаёт ГК без филиала → base_regional_office_id = Ростовский; с чужим → 422 OFFICE_NOT_ALLOWED.
Step 2: падает → Step 3: реализация → Step 4: pytest -m integration tests/test_api_company_groups.py tests/test_api_policy.py -v; ruff, mypy → PASS.
Step 5: Commit feat(directories): филиалы и ГК — API, история, аудит, видимость.

Task 6: [2.3] ЮЛ/ИП: CRUD, перевод, вывод из состава, история

Files:

Create: api/src/domain/inn.py, api/src/services/legal_entities.py, api/src/api/endpoints/directories/legal_entities.py
Modify: api/src/repositories/directories.py, api/src/schemas/directories.py, api/src/api/dependencies/directories.py, api/src/api/endpoints/directories/__init__.py
Test: api/tests/test_domain_inn.py, api/tests/test_api_legal_entities.py

Interfaces:

src/domain/inn.py: INN_LENGTH = {EntityType.ЮЛ: 10, EntityType.ИП: 12}; def normalize_inn(raw: str) -> str (убрать пробелы); def validate_inn(inn: str, entity_type: EntityType) -> str → нормализованный ИНН или DomainValidationError("INVALID_INN", f"ИНН {entity_type.value} — ровно {n} цифр"); # ASSUMPTION(OQ-88): контрольные разряды не проверяются.
Репозиторий: list_legal_entities(session, scope, *, company_group_id, outside_group: bool | None, inn: str | None, q: str | None, closed: bool | None, limit, offset) -> tuple[list[LegalEntity], int]; get_legal_entity(session, scope, entity_id) -> LegalEntity | None (через legal_entities_visible); get_legal_entity_by_inn(session, inn) -> LegalEntity | None (без scope — для pre-check уникальности); list_legal_entity_history(session, entity_id); current_memberships(session, entity_id) -> list[RelatedGroupMember]; close_memberships(session, entity_id, now) -> list[RelatedGroupMember] (SET valid_to, is_current=false на всех текущих; возвращает закрытые).
src/services/legal_entities.py:

class LegalEntityService:
    def __init__(self, session, audit: RequestAudit, scope: VisibilityScope) -> None
    async def create(self, data: LegalEntityCreate) -> LegalEntity       # validate_inn; INN_EXISTS → ConflictError; crm_id → CRM_ID_EXISTS; company_group_id — через get_company_group(scope) иначе NotFoundError("ГК не найдена"); regional_office_id по умолчанию = офис ГК, для монозаёмщика — office_for_create; audit LEGAL_ENTITY_CHANGE
    async def update(self, entity_id, data: LegalEntityUpdate) -> LegalEntity   # name, entity_type (с повторной validate_inn), crm_id, regional_office_id; no-op без истории; снимок LegalEntityHistory(event_type=UPDATE, valid_from=entity.valid_from, valid_to=now, все поля-снимки), entity.valid_from=now; audit LEGAL_ENTITY_CHANGE
    async def move(self, entity_id, data: LegalEntityMove) -> LegalEntity        # effective_date >= today иначе DomainValidationError("EFFECTIVE_DATE_IN_PAST") ASSUMPTION(OQ-21); целевая ГК через scope → 404; та же ГК → ConflictError("SAME_COMPANY_GROUP"); снимок MOVE_TO_GROUP; close_memberships; audit LEGAL_ENTITY_CHANGE (before/after company_group_id) + MEMBERSHIP_CHANGE на каждое закрытое членство (object_type="related_group_member")
    async def withdraw(self, entity_id, data: LegalEntityWithdraw) -> LegalEntity  # event CLOSE | REMOVE_FROM_GROUP; company_group_id=None; CLOSE → close_date=effective_date; снимок с event_type; close_memberships; аудит как в move; comment → audit.comment
    async def get(self, entity_id) -> LegalEntityDetail   # entity + current memberships (related_group id/code/kind/name) + aliases
    async def history(self, entity_id) -> list[LegalEntityHistory]
    async def page(self, flt: LegalEntityFilter, page: PageParams) -> tuple[list[LegalEntity], int]

Схемы: LegalEntityCreate(inn: str, entity_type: EntityType, name: str, company_group_id: uuid | None = None, regional_office_id: uuid | None = None, crm_id: str | None = None); LegalEntityUpdate (name, entity_type, crm_id, regional_office_id — optional, extra="forbid"); LegalEntityMove(company_group_id: uuid.UUID, effective_date: datetime.date); LegalEntityWithdraw(event: Literal["CLOSE", "REMOVE_FROM_GROUP"], effective_date: datetime.date, comment: str | None = None); LegalEntityOut(id, inn, crm_id, entity_type, name, company_group_id, regional_office_id, close_date, valid_from, outside_group: bool, closed: bool, created_at, updated_at); MembershipOut(related_group_id, code, kind, name, valid_from); AliasOut(id, alias, normalized_alias, created_at); LegalEntityDetail(LegalEntityOut + memberships: list[MembershipOut], aliases: list[AliasOut]); LegalEntityHistoryOut(все поля снимка + event_type, effective_date, valid_from, valid_to, changed_by); LegalEntityFilter(company_group_id, outside_group, inn, q, closed).
Эндпоинты: GET/POST /legal-entities, GET/PATCH /legal-entities/{id}, POST /legal-entities/{id}/move, POST /legal-entities/{id}/withdraw, GET /legal-entities/{id}/history.
Step 1: Тесты. tests/test_domain_inn.py (unit): 10 цифр ЮЛ ок, 12 ИП ок, 0123456789 ок (Review Focus 2), 10 цифр для ИП → INVALID_INN, буквы → INVALID_INN, пробелы внутри убираются. tests/test_api_legal_entities.py (integration): создание ИП с 10 цифрами → 422 INVALID_INN; дубликат ИНН → 409 INN_EXISTS; монозаёмщик (ka_rf, без ГК) → outside_group == true, виден ka_rf и ka_go, не виден client_manager; move закрывает членство (фабрики make_related_group + make_membership), история MOVE_TO_GROUP с датой; move в ГК чужого филиала пользователем ka_rf → 404 (Review Focus 3); effective_date вчера → 422; withdraw CLOSE → close_date, company_group_id is None, closed == true в списке с фильтром closed=true; no-op PATCH без истории; фильтр inn=" 3528052718 " находит ЮЛ с inn="3528052718"; откат мутации — monkeypatch close_memberships на бросающую RuntimeError → 500, ЮЛ в прежней ГК, истории/аудита нет (SAVEPOINT откатился — проверяется в той же сессии).
Step 2–4: падает → реализация → pytest tests/test_domain_inn.py && pytest -m integration tests/test_api_legal_entities.py -v; ruff, mypy.
Step 5: Commit feat(directories): ЮЛ/ИП — CRUD, перевод, вывод из состава, история, аудит.

Task 7: [2.4] Алиасы ЮЛ

Files:

Create: api/src/domain/aliases.py, api/src/services/aliases.py
Modify: api/src/repositories/directories.py (find_alias_owner(session, normalized) -> LegalEntityAlias | None, list_aliases(session, entity_id)), api/src/api/endpoints/directories/legal_entities.py (маршруты алиасов), api/src/schemas/directories.py (AliasCreate(alias: str = Field(min_length=1, max_length=500)))
Test: api/tests/test_domain_aliases.py, api/tests/test_api_aliases.py

Interfaces:

src/domain/aliases.py (§3 п. 4 — «lower, trim, убрать пунктуацию/кавычки, убрать ОПФ (ООО, АО, ПАО, ИП…), схлопнуть пробелы»):

LEGAL_FORMS: frozenset[str] = frozenset({"ооо", "ао", "пао", "зао", "оао", "нао", "ип", "гуп", "муп", "фгуп", "ано", "тсж", "кфх", "пк", "спк", "сх", "сха"})  # одиночные токены ОПФ (§3); убираются только целиком, «ооо» внутри слова не трогается
_PUNCT_RE = re.compile(r"[^\w\s]", re.UNICODE)   # кавычки «»"'„“ и прочая пунктуация → пробел
def normalize_alias(raw: str) -> str:
    """§3: lower → ё→е → пунктуация и кавычки → пробел → убрать токены ОПФ → схлопнуть пробелы → strip."""

AliasService(session, audit, scope): async def add(entity_id, data) -> LegalEntityAlias (ЮЛ через scope → 404; normalize_alias; пустая нормализованная строка → DomainValidationError("EMPTY_ALIAS"); у того же ЮЛ → ConflictError("ALIAS_EXISTS"); у другого ЮЛ → ConflictError("ALIAS_TAKEN", …, legal_entity_id=owner_id); created_by, created_at; audit LEGAL_ENTITY_CHANGE object_type="legal_entity_alias"), async def remove(entity_id, alias_id) -> None (404 если не у этого ЮЛ; audit с before_value).
Маршруты: POST /legal-entities/{id}/aliases → 201 AliasOut (DIRECTORY_EDIT), DELETE /legal-entities/{id}/aliases/{alias_id} → 204 (DIRECTORY_DELETE).
Step 1: Тесты. unit: normalize_alias('ООО «Пекарня Счастливочная»') == "пекарня счастливочная"; 'Пекарня Счастливочная, ООО' → то же; 'АО "РУССКИЙ БИСКВИТ"' → "русский бисквит"; 'ИП Иванов И.И.' → "иванов и и"; ё/е; двойные пробелы; строка только из ОПФ → "". integration: сценарий спеки «Нормализация» (409 ALIAS_EXISTS), ALIAS_TAKEN с legal_entity_id владельца, DELETE → 204 и запись аудита, expert → 403 на POST.
Step 2–4: падает → реализация → тесты, ruff, mypy → PASS.
Step 5: Commit feat(directories): алиасы ЮЛ с нормализацией §3.

Task 8: [2.5] ГСК/ГСЗ и состав

Files:

Create: api/src/services/related_groups.py, api/src/api/endpoints/directories/related_groups.py
Modify: api/src/repositories/directories.py, api/src/schemas/directories.py, api/src/api/dependencies/directories.py, api/src/api/endpoints/directories/__init__.py
Test: api/tests/test_api_related_groups.py

Interfaces:

Репозиторий: list_related_groups(session, company_group_id) -> list[tuple[RelatedGroup, int]] (с числом текущих членов), get_related_group(session, scope, group_id) -> RelatedGroup | None (через ГК и company_groups_visible), current_membership_of_kind(session, entity_id, kind) -> RelatedGroupMember | None, current_members(session, related_group_id) -> list[tuple[RelatedGroupMember, LegalEntity]], has_current_members(session, related_group_id) -> bool.
RelatedGroupService(session, audit, scope):
- create(company_group_id, data: RelatedGroupCreate) -> RelatedGroup — ГК через scope → 404; code = next_code(RELATED_GROUP_CODE_SEQ, "GSK"|"GSZ"); audit MEMBERSHIP_CHANGE object_type="related_group".
- update(group_id, data: RelatedGroupUpdate) — name, created_date.
- delete(group_id) — has_current_members → ConflictError("GROUP_NOT_EMPTY", "в группе есть текущие члены") # ASSUMPTION(OQ-34); удаляет и закрытые строки состава (история группы исчезает вместе с ней); audit с before_value.
- add_member(group_id, data: MemberAdd) -> RelatedGroupMember — ЮЛ через scope → 404; entity.company_group_id != group.company_group_id (в т.ч. None) → DomainValidationError("DIFFERENT_COMPANY_GROUP"); current_membership_of_kind(entity, group.kind): та же группа → ConflictError("MEMBER_EXISTS"), другая → ConflictError("MEMBER_OF_OTHER_GROUP", related_group_id=…); строка RelatedGroupMember(kind=group.kind, valid_from=now, is_current=True, created_by=user); audit MEMBERSHIP_CHANGE object_type="related_group_member".
- remove_member(group_id, legal_entity_id) — текущее членство → valid_to=now, is_current=False; нет → 404; audit.
- get(group_id) -> RelatedGroupDetail, list_for(company_group_id) -> list[RelatedGroupOut].
Схемы: RelatedGroupCreate(kind: RelatedGroupKind, name: str, created_date: date | None = None), RelatedGroupUpdate(name?, created_date?), MemberAdd(legal_entity_id: uuid.UUID), RelatedGroupOut(id, code, company_group_id, kind, name, created_date, members_count: int), RelatedGroupMemberOut(legal_entity_id, inn, name, valid_from), RelatedGroupDetail(RelatedGroupOut + members: list[RelatedGroupMemberOut]).
Маршруты: GET/POST /company-groups/{gk_id}/related-groups; GET/PATCH /related-groups/{id}; DELETE /related-groups/{id} (DIRECTORY_DELETE); POST /related-groups/{id}/members → 201; DELETE /related-groups/{id}/members/{legal_entity_id} → 204.
Step 1: Тесты (сценарии спеки + Review Focus 4): коды GSK-/GSZ- + 6 цифр; ЮЛ в ГСК и затем в ГСЗ той же ГК — обе 201, два текущих членства с разными kind; вторая ГСК → 409 MEMBER_OF_OTHER_GROUP с id ГСК A; повтор в той же → 409 MEMBER_EXISTS; исключение → is_current=false, valid_to; повторное включение → новая строка, старая осталась; удаление непустой → 409 GROUP_NOT_EMPTY; удаление пустой → 204; ЮЛ другой ГК → 422 DIFFERENT_COMPANY_GROUP; закрытый ЮЛ (без ГК) → 422 DIFFERENT_COMPANY_GROUP; kind строки = kind группы (проверить в БД); ka_rf для группы чужой ГК → 404.
Step 2–4: падает → реализация → pytest -m integration tests/test_api_related_groups.py -v, полный make check → PASS.
Step 5: Commit feat(directories): ГСК/ГСЗ и состав (SCD2, одна группа каждого вида на ЮЛ).

Завершение фазы 1

После Task 8: make check (из корня) и make dbml-check зелёные; pytest -m integration зелёный; tasks.md change'а — группы 1–2 отмечены [x]. Фаза 2 (2026-10-04-slice-1-directories-phase2.md): импорт из ЕЦП, администрирование, фронт, документы, make check-full.

Проверки фазы 1 (Task 8, 2026-10-04): make check — 667 passed, ruff/mypy чистые; make dbml-check — пустой вывод; pytest -m integration — 229 passed, 0 skipped, 0 failed; tasks.md — группы 1–2 [x].

50 KiB Raw Permalink Blame History Unescape Escape

Срез 1 — справочники arm-core — план исполнения, фаза 1 (модель v5.14, инфраструктура запросов, policy, API справочников)

Global Constraints

Review Focus

Задачи

Task 1: [1.1] ORM v5.14, миграция 0002_dbml_v5_14, версия DBML

Task 2: [1.2] Единица работы, аудит с контекстом запроса, доменные исключения, пагинация

Task 3: [1.3] Фабрики справочников и auth_client в conftest

Task 4: [2.1] Policy видимости — src/domain/policy.py

Task 5: [2.2] Филиалы и ГК: репозиторий, сервис, схемы, эндпоинты

Task 6: [2.3] ЮЛ/ИП: CRUD, перевод, вывод из состава, история

Task 7: [2.4] Алиасы ЮЛ

Task 8: [2.5] ГСК/ГСЗ и состав

Завершение фазы 1

50 KiB

Raw Permalink Blame History

Task 1: [1.1] ORM v5.14, миграция `0002_dbml_v5_14`, версия DBML

Task 3: [1.3] Фабрики справочников и `auth_client` в conftest

Task 4: [2.1] Policy видимости — `src/domain/policy.py`