From d142c6f69eb038c80b3328327ea12e12f8643398 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 4 Oct 2026 21:17:58 +0400 Subject: [PATCH] =?UTF-8?q?feat(domain):=20policy=20=D0=B2=D0=B8=D0=B4?= =?UTF-8?q?=D0=B8=D0=BC=D0=BE=D1=81=D1=82=D0=B8=20=D1=81=D0=BF=D1=80=D0=B0?= =?UTF-8?q?=D0=B2=D0=BE=D1=87=D0=BD=D0=B8=D0=BA=D0=BE=D0=B2=20(=C2=A710.1,?= =?UTF-8?q?=20ASSUMPTION(OQ-86))?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Первый модуль src/domain/ (чистые функции, без I/O, mypy strict): VisibilityScope, scope_for/scope_of, SQL-фильтры company_groups_visible/legal_entities_visible, проверки company_group_is_visible/legal_entity_is_visible, office_for_create (чужой филиал → 422 OFFICE_NOT_ALLOWED). Код роли не читается; business_segment не фильтрует (ASSUMPTION(OQ-86)). Принадлежность ЮЛ ГК филиала — подзапросом IN, а не JOIN (число строк не меняется, нет конфликта с JOIN вызывающего). Тесты: unit без БД (компиляция Select под PostgreSQL, все ветки) + один integration-тест семантики фильтров и их согласованности с проверками объектов. Отмечены шаги Task 4 в плане и 2.1 в tasks.md. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/src/domain/__init__.py | 1 + api/src/domain/policy.py | 109 +++++++ api/tests/test_domain_policy.py | 270 ++++++++++++++++++ .../2026-10-04-slice-1-directories-phase1.md | 6 +- openspec/changes/slice-1-directories/tasks.md | 2 +- 5 files changed, 384 insertions(+), 4 deletions(-) create mode 100644 api/src/domain/__init__.py create mode 100644 api/src/domain/policy.py create mode 100644 api/tests/test_domain_policy.py diff --git a/api/src/domain/__init__.py b/api/src/domain/__init__.py new file mode 100644 index 0000000..993f891 --- /dev/null +++ b/api/src/domain/__init__.py @@ -0,0 +1 @@ +"""Доменный слой: чистые функции по §§ правил `05-domain-rules.md`, без I/O (ни сессий БД, ни сети, ни файлов).""" diff --git a/api/src/domain/policy.py b/api/src/domain/policy.py new file mode 100644 index 0000000..f6b96f9 --- /dev/null +++ b/api/src/domain/policy.py @@ -0,0 +1,109 @@ +"""Policy видимости справочников (§10.1 `05-domain-rules.md`, spec directory-policy). + +Правило формулируется через данные, а не через роль (инвариант «права — данные», §10): пользователь головного +офиса (`regional_office.is_head_office`) видит всё; пользователь филиала — ГК с `base_regional_office_id` своего +филиала, их ЮЛ и ГСК/ГСЗ, а также монозаёмщиков своего филиала (`company_group_id IS NULL AND regional_office_id = …`). +Код роли (`role.code`) здесь не читается. Право (`require`) отвечает «может ли вообще», policy — «какие именно +записи»; объект вне видимости вызывающий превращает в 404 с тем же телом, что у несуществующего. + +ASSUMPTION(OQ-86): ЗО (`app_user.business_segment`) в Срезе 1 не фильтрует — у ГК и ЮЛ нет поля сегмента (DBML v5.14). +""" + +import uuid +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from sqlalchemy import Select, and_, or_, select + +from src.common.exceptions import DomainValidationError +from src.models.directories import CompanyGroup, LegalEntity + +if TYPE_CHECKING: + # Только для типов: домен не тянет репозитории в рантайме + from src.repositories.users import UserWithAccess + + +@dataclass(frozen=True) +class VisibilityScope: + """Область видимости справочников: `office_id is None` — вся система (ГО).""" + + office_id: uuid.UUID | None + + @property + def is_global(self) -> bool: + """Пользователь головного офиса — видит все записи (§10.1).""" + return self.office_id is None + + +def scope_for(*, is_head_office: bool, regional_office_id: uuid.UUID) -> VisibilityScope: + """Область видимости по данным о филиале пользователя (§10.1): ГО — вся система, иначе — свой филиал. + + ASSUMPTION(OQ-86): `business_segment` не учитывается. + """ + return VisibilityScope(office_id=None if is_head_office else regional_office_id) + + +def scope_of(principal: "UserWithAccess") -> VisibilityScope: + """Область видимости текущего пользователя (§10.1); роли и права `principal` не участвуют.""" + return 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]: + """Фильтр выборки ГК (§10.1): для филиала — `company_group.base_regional_office_id = <филиал>`. + + `stmt` должен ссылаться на `CompanyGroup` (выборка ГК или запрос, уже присоединивший ГК). Для ГО `stmt` не меняется. + """ + if scope.office_id is None: + return stmt + return stmt.where(CompanyGroup.base_regional_office_id == scope.office_id) + + +def legal_entities_visible(stmt: Select[Any], scope: VisibilityScope) -> Select[Any]: + """Фильтр выборки ЮЛ (§10.1): для филиала — ЮЛ в ГК своего филиала ИЛИ монозаёмщик своего филиала. + + Принадлежность ГК филиалу проверяется подзапросом `company_group_id IN (SELECT id … WHERE base_regional_office_id = + <филиал>)`, а не JOIN: число строк не меняется, а запрос вызывающего может сам присоединять `company_group` без + конфликта имён. Филиал самого ЮЛ в группе не важен — видимость определяет ГК. Для ГО `stmt` не меняется. + """ + if scope.office_id is None: + return stmt + groups_of_office = select(CompanyGroup.id).where(CompanyGroup.base_regional_office_id == scope.office_id) + return stmt.where( + or_( + LegalEntity.company_group_id.in_(groups_of_office), + and_(LegalEntity.company_group_id.is_(None), LegalEntity.regional_office_id == scope.office_id), + ) + ) + + +def company_group_is_visible(group: CompanyGroup, scope: VisibilityScope) -> bool: + """Видна ли ГК пользователю (§10.1); то же правило, что :func:`company_groups_visible`.""" + return scope.office_id is None or group.base_regional_office_id == scope.office_id + + +def legal_entity_is_visible(entity: LegalEntity, group_office_id: uuid.UUID | None, scope: VisibilityScope) -> bool: + """Виден ли ЮЛ пользователю (§10.1); то же правило, что :func:`legal_entities_visible`. + + `group_office_id` — `base_regional_office_id` ГК этого ЮЛ (для монозаёмщика игнорируется). + """ + if scope.office_id is None: + return True + if entity.company_group_id is not None: + return group_office_id == scope.office_id + return entity.regional_office_id == scope.office_id + + +def office_for_create(scope: VisibilityScope, requested: uuid.UUID | None) -> uuid.UUID | None: + """Филиал закрепления новой ГК (spec directory-policy «Политика применяется до прав на изменение», §10.1). + + ГО — `requested` как есть (в т.ч. `None`); филиал — `None` превращается в свой филиал, свой принимается, чужой → + 422 `OFFICE_NOT_ALLOWED`. + """ + if scope.office_id is None: + return requested + if requested is None or requested == scope.office_id: + return scope.office_id + raise DomainValidationError("OFFICE_NOT_ALLOWED", "ГК можно закрепить только за своим филиалом") diff --git a/api/tests/test_domain_policy.py b/api/tests/test_domain_policy.py new file mode 100644 index 0000000..557293f --- /dev/null +++ b/api/tests/test_domain_policy.py @@ -0,0 +1,270 @@ +"""Policy видимости справочников (§10.1 `05-domain-rules.md`, spec directory-policy, task 2.1). + +Unit-часть — без БД: область видимости из данных о филиале, SQL-фильтры (компиляция `Select` под PostgreSQL с +подставленными значениями), проверки отдельных объектов и выбор филиала при создании ГК. +Integration-часть — семантика SQL-фильтров на настоящей БД: какие именно ГК и ЮЛ попадают в выборку, и что SQL-фильтр +совпадает с проверкой отдельного объекта (`*_is_visible`). +""" + +import re +import uuid +from types import SimpleNamespace +from typing import Any, cast + +import pytest +from fastapi import status as http_status +from sqlalchemy import Select, func, select +from sqlalchemy.dialects import postgresql +from sqlalchemy.ext.asyncio import AsyncSession + +from src.common.exceptions import DomainValidationError +from src.domain.policy import ( + VisibilityScope, + company_group_is_visible, + company_groups_visible, + legal_entities_visible, + legal_entity_is_visible, + office_for_create, + scope_for, + scope_of, +) +from src.models.directories import CompanyGroup, LegalEntity +from tests.factories import make_company_group, make_legal_entity, make_regional_office + +OFFICE_X = uuid.UUID("00000000-0000-0000-0000-0000000000aa") +OFFICE_Y = uuid.UUID("00000000-0000-0000-0000-0000000000bb") +GLOBAL = VisibilityScope(office_id=None) +BRANCH_X = VisibilityScope(office_id=OFFICE_X) + + +def _sql(stmt: Select[Any]) -> str: + """SQL под PostgreSQL с подставленными значениями, пробелы схлопнуты в один.""" + compiled = stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}) + return re.sub(r"\s+", " ", str(compiled)).strip() + + +def _where(stmt: Select[Any]) -> str: + """Только часть после WHERE (или пустая строка).""" + sql = _sql(stmt) + return sql.split(" WHERE ", 1)[1] if " WHERE " in sql else "" + + +# --- область видимости --- + + +def test_head_office_scope_is_global() -> None: + scope = scope_for(is_head_office=True, regional_office_id=OFFICE_X) + assert scope.is_global + assert scope.office_id is None + + +def test_branch_scope_is_bound_to_its_office() -> None: + scope = scope_for(is_head_office=False, regional_office_id=OFFICE_X) + assert not scope.is_global + assert scope.office_id == OFFICE_X + + +@pytest.mark.parametrize(("is_head_office", "expected"), [(True, GLOBAL), (False, BRANCH_X)]) +def test_scope_of_principal_uses_office_data_not_role(is_head_office: bool, expected: VisibilityScope) -> None: + # Роль заведомо «чужая» уровню: область зависит только от филиала (права — данные, код роли не читается) + principal = SimpleNamespace( + user=SimpleNamespace(regional_office_id=OFFICE_X), + regional_office=SimpleNamespace(id=OFFICE_X, is_head_office=is_head_office), + roles=("KA_RF",) if is_head_office else ("KA_GO",), + permissions=frozenset({"DIRECTORY_VIEW"}), + ) + assert scope_of(cast(Any, principal)) == expected + + +# --- SQL-фильтр ГК --- + + +def test_company_groups_global_returns_statement_unchanged() -> None: + stmt = select(CompanyGroup) + filtered = company_groups_visible(stmt, GLOBAL) + assert filtered is stmt + assert _where(filtered) == "" + + +def test_company_groups_branch_filters_by_base_office() -> None: + filtered = company_groups_visible(select(CompanyGroup), BRANCH_X) + assert _where(filtered) == f"dkb.company_group.base_regional_office_id = '{OFFICE_X}'" + + +def test_company_groups_filter_composes_with_existing_where() -> None: + stmt = select(CompanyGroup).where(CompanyGroup.is_archived.is_(False)) + where = _where(company_groups_visible(stmt, BRANCH_X)) + assert where == ( + f"dkb.company_group.is_archived IS false AND dkb.company_group.base_regional_office_id = '{OFFICE_X}'" + ) + + +# --- SQL-фильтр ЮЛ --- + + +def test_legal_entities_global_returns_statement_unchanged() -> None: + stmt = select(LegalEntity) + filtered = legal_entities_visible(stmt, GLOBAL) + assert filtered is stmt + assert _where(filtered) == "" + + +def test_legal_entities_branch_filter_is_group_of_office_or_mono_borrower_of_office() -> None: + where = _where(legal_entities_visible(select(LegalEntity), BRANCH_X)) + office = f"'{OFFICE_X}'" + # ЮЛ в ГК своего филиала ИЛИ (монозаёмщик И филиал ЮЛ — свой); единственное условие на запрос, обе ветви — по X + assert where == ( + "dkb.legal_entity.company_group_id IN (SELECT dkb.company_group.id FROM dkb.company_group " + f"WHERE dkb.company_group.base_regional_office_id = {office}) " + f"OR dkb.legal_entity.company_group_id IS NULL AND dkb.legal_entity.regional_office_id = {office}" + ) + assert " OR " in where + assert "IS NULL" in where + + +def test_legal_entities_branch_filter_does_not_multiply_rows_or_clash_with_caller_join() -> None: + # Вызывающий сам присоединяет ГК (например, для имени группы): фильтр не добавляет второй FROM/JOIN company_group + stmt = select(LegalEntity, CompanyGroup.name).outerjoin( + CompanyGroup, CompanyGroup.id == LegalEntity.company_group_id + ) + sql = _sql(legal_entities_visible(stmt, BRANCH_X)) + from_clause = sql.split(" WHERE ", 1)[0].split(" FROM ", 1)[1] + assert from_clause == ( + "dkb.legal_entity LEFT OUTER JOIN dkb.company_group ON dkb.company_group.id = dkb.legal_entity.company_group_id" + ) + + +def test_legal_entities_filter_works_for_count_statement() -> None: + stmt = select(func.count()).select_from(LegalEntity) + sql = _sql(legal_entities_visible(stmt, BRANCH_X)) + assert sql.startswith("SELECT count(*) AS count_1 FROM dkb.legal_entity WHERE ") + + +# --- проверка отдельной ГК --- + + +@pytest.mark.parametrize( + ("scope", "group_office", "visible"), + [ + (GLOBAL, OFFICE_Y, True), + (GLOBAL, None, True), + (BRANCH_X, OFFICE_X, True), + (BRANCH_X, OFFICE_Y, False), + (BRANCH_X, None, False), + ], +) +def test_company_group_is_visible(scope: VisibilityScope, group_office: uuid.UUID | None, visible: bool) -> None: + group = CompanyGroup(code="GK-000001", name="ГК", base_regional_office_id=group_office) + assert company_group_is_visible(group, scope) is visible + + +# --- проверка отдельного ЮЛ --- + + +@pytest.mark.parametrize( + ("scope", "in_group", "group_office", "entity_office", "visible"), + [ + # монозаёмщик своего / чужого / без филиала + (BRANCH_X, False, None, OFFICE_X, True), + (BRANCH_X, False, None, OFFICE_Y, False), + (BRANCH_X, False, None, None, False), + # ЮЛ в ГК своего / чужого филиала; филиал самого ЮЛ не важен — решает ГК + (BRANCH_X, True, OFFICE_X, OFFICE_Y, True), + (BRANCH_X, True, OFFICE_Y, OFFICE_X, False), + (BRANCH_X, True, None, OFFICE_X, False), + # ГО видит всё + (GLOBAL, False, None, OFFICE_Y, True), + (GLOBAL, True, OFFICE_Y, OFFICE_Y, True), + ], +) +def test_legal_entity_is_visible( + scope: VisibilityScope, + in_group: bool, + group_office: uuid.UUID | None, + entity_office: uuid.UUID | None, + visible: bool, +) -> None: + entity = LegalEntity( + inn="7700000000", + name="ООО Тест", + company_group_id=uuid.uuid4() if in_group else None, + regional_office_id=entity_office, + ) + assert legal_entity_is_visible(entity, group_office, scope) is visible + + +# --- филиал при создании ГК --- + + +@pytest.mark.parametrize("requested", [None, OFFICE_Y]) +def test_office_for_create_head_office_keeps_requested(requested: uuid.UUID | None) -> None: + assert office_for_create(GLOBAL, requested) == requested + + +def test_office_for_create_branch_defaults_to_own_office() -> None: + assert office_for_create(BRANCH_X, None) == OFFICE_X + + +def test_office_for_create_branch_accepts_own_office() -> None: + assert office_for_create(BRANCH_X, OFFICE_X) == OFFICE_X + + +def test_office_for_create_branch_rejects_foreign_office() -> None: + with pytest.raises(DomainValidationError) as exc_info: + office_for_create(BRANCH_X, OFFICE_Y) + assert exc_info.value.code == "OFFICE_NOT_ALLOWED" + assert exc_info.value.status == http_status.HTTP_422_UNPROCESSABLE_CONTENT + assert exc_info.value.message == "ГК можно закрепить только за своим филиалом" + + +# --- семантика SQL-фильтров на настоящей БД --- + + +@pytest.mark.integration +async def test_sql_filters_match_object_checks_on_real_db(db_session: AsyncSession) -> None: + office_x = await make_regional_office(db_session) + office_y = await make_regional_office(db_session) + group_x = await make_company_group(db_session, name="ГК X", office=office_x) + group_y = await make_company_group(db_session, name="ГК Y", office=office_y) + group_none = await make_company_group(db_session, name="ГК без филиала") + entities = { + "в ГК X": await make_legal_entity(db_session, group=group_x), + "в ГК X, филиал ЮЛ Y": await make_legal_entity(db_session, group=group_x, office=office_y), + "в ГК Y": await make_legal_entity(db_session, group=group_y), + "в ГК Y, филиал ЮЛ X": await make_legal_entity(db_session, group=group_y, office=office_x), + "в ГК без филиала, филиал ЮЛ X": await make_legal_entity(db_session, group=group_none, office=office_x), + "монозаёмщик X": await make_legal_entity(db_session, office=office_x), + "монозаёмщик Y": await make_legal_entity(db_session, office=office_y), + "монозаёмщик без филиала": await make_legal_entity(db_session), + } + names_by_id = {entity.id: name for name, entity in entities.items()} + groups = {group.id: group for group in (group_x, group_y, group_none)} + branch_x = scope_for(is_head_office=False, regional_office_id=office_x.id) + head = scope_for(is_head_office=True, regional_office_id=office_x.id) + + async def visible_entities(scope: VisibilityScope) -> set[str]: + stmt = legal_entities_visible(select(LegalEntity.id).where(LegalEntity.id.in_(names_by_id)), scope) + return {names_by_id[row] for row in (await db_session.scalars(stmt)).all()} + + async def visible_groups(scope: VisibilityScope) -> set[uuid.UUID]: + stmt = company_groups_visible(select(CompanyGroup.id).where(CompanyGroup.id.in_(groups)), scope) + return set((await db_session.scalars(stmt)).all()) + + assert await visible_entities(branch_x) == {"в ГК X", "в ГК X, филиал ЮЛ Y", "монозаёмщик X"} + assert await visible_entities(head) == set(entities) + assert await visible_groups(branch_x) == {group_x.id} + assert await visible_groups(head) == set(groups) + + # SQL-фильтр и проверка отдельного объекта согласованы + for scope in (branch_x, head): + expected_entities = { + name + for name, entity in entities.items() + if legal_entity_is_visible( + entity, + groups[entity.company_group_id].base_regional_office_id if entity.company_group_id else None, + scope, + ) + } + assert await visible_entities(scope) == expected_entities + assert await visible_groups(scope) == {gid for gid, g in groups.items() if company_group_is_visible(g, scope)} diff --git a/docs/superpowers/plans/2026-10-04-slice-1-directories-phase1.md b/docs/superpowers/plans/2026-10-04-slice-1-directories-phase1.md index 72cd514..cf5ae46 100644 --- a/docs/superpowers/plans/2026-10-04-slice-1-directories-phase1.md +++ b/docs/superpowers/plans/2026-10-04-slice-1-directories-phase1.md @@ -277,9 +277,9 @@ def office_for_create(scope: VisibilityScope, requested: uuid.UUID | None) -> uu ``` `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))`. +- [x] **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"`. +- [x] **Step 2:** падает → **Step 3:** реализация → **Step 4:** `pytest tests/test_domain_policy.py -v`, `mypy src` (strict для `src.domain.*`) → PASS. +- [x] **Step 5: Commit** `feat(domain): policy видимости справочников (§10.1, ASSUMPTION(OQ-86))`. --- diff --git a/openspec/changes/slice-1-directories/tasks.md b/openspec/changes/slice-1-directories/tasks.md index c7f3974..b41380a 100644 --- a/openspec/changes/slice-1-directories/tasks.md +++ b/openspec/changes/slice-1-directories/tasks.md @@ -26,7 +26,7 @@ test-integration`). Пути — от корня репозитория. Доп ## 2. Policy и справочники -- [ ] 2.1 **Policy.** `api/src/domain/policy.py` (strict mypy): `VisibilityScope`, `scope_for(principal)`, `apply_scope_company_groups(stmt, scope)`, +- [x] 2.1 **Policy.** `api/src/domain/policy.py` (strict mypy): `VisibilityScope`, `scope_for(principal)`, `apply_scope_company_groups(stmt, scope)`, `apply_scope_legal_entities(stmt, scope)`, `assert_office_allowed(scope, office_id)`; unit-тесты без БД на все ветки (ГО, филиал, монозаёмщик, чужой филиал → `OFFICE_NOT_ALLOWED`). - [ ] 2.2 **Филиалы и ГК.** `repositories/directories.py` (запросы с scope), `services/company_groups.py` (создание с `code` из