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