test(import): вынести committed_client в conftest, добиты проверки dry_run, безопасное имя файла (T10.1–T10.4, T10.6)

- T10.6: committed_sessionmaker/committed_app/committed_client/committed_storage и FakeStorage — в conftest.py,
  приложение собирается общей _build_application (как и app).
- T10.1: dry_run не пишет в S3 (objects == {}), боевой прогон — ровно один объект.
- T10.2: dry_run на реальном xlsx РБ (21 строка, таблицы не изменились).
- T10.3: dry_run с on_existing=update не оставляет истории/аудита, имена прежние.
- T10.4: safe_filename отбрасывает управляющие символы и ограничивает длину (200) с сохранением расширения.

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

api/src/services/directory_import.py

@@ -15,6 +15,7 @@
 import asyncio
 import datetime
 import hashlib
+import unicodedata
 import uuid
 from collections import defaultdict
 from pathlib import PurePosixPath
@@ -69,6 +70,8 @@ CONTENT_TYPES = {
     "csv": "text/csv",
 }
 MEGABYTE = 1024 * 1024
+MAX_FILENAME_LENGTH = 200  # символов: ключ S3 `uploads/imports/{uuid}/{имя}` остаётся в пределах 1024 байт
+MAX_SUFFIX_LENGTH = 16
 
 
 class ObjectStorage(Protocol):
@@ -86,9 +89,20 @@ def check_size(size: int, limit: int) -> None:
 
 
 def safe_filename(filename: str | None) -> str:
-    """Имя файла без пути (клиент может прислать `C:\\…\\x.xlsx` или `../x.csv`): для формата, аудита и ключа S3."""
+    """Имя файла без пути (клиент может прислать `C:\\…\\x.xlsx` или `../x.csv`): для формата, аудита и ключа S3.
-    name = PurePosixPath((filename or "").replace("\\", "/")).name.strip()
+
-    return name if name not in {"", ".", ".."} else DEFAULT_FILENAME
+    Управляющие символы (в т. ч. NUL — его не принимает PostgreSQL) отбрасываются; длина ограничена
+    `MAX_FILENAME_LENGTH` с сохранением расширения (T10.4), чтобы имя не давало 500 и ключ S3 не выходил за 1024 байта.
+    """
+    cleaned = "".join(char for char in (filename or "").replace("\\", "/") if unicodedata.category(char) != "Cc")
+    name = PurePosixPath(cleaned).name.strip()
+    if name in {"", ".", ".."}:
+        return DEFAULT_FILENAME
+    if len(name) > MAX_FILENAME_LENGTH:
+        suffix = PurePosixPath(name).suffix
+        suffix = suffix if len(suffix) <= MAX_SUFFIX_LENGTH else ""
+        name = name[: MAX_FILENAME_LENGTH - len(suffix)] + suffix
+    return name
 
 
 class DirectoryImportService:

api/tests/conftest.py

@@ -4,6 +4,12 @@
 трогается (`make check` не требует БД). Тестовая БД сессии — отдельная свежая `armcore_pytest_<hex>` на сервере из
 `.env.test`: мигрируется alembic один раз на сессию и удаляется в конце; рабочая БД `armka` не используется.
 Изоляция тестов — внешняя транзакция с SAVEPOINT: `commit()` внутри теста ничего не фиксирует.
+
+`committed_sessionmaker` / `committed_app` / `committed_client` / `committed_storage` — для тестов, которым нужна сессия
+запроса «как в проде»: отдельная временная БД с ЗАФИКСИРОВАННЫМ seed и сессия `async with session.begin()` без
+SAVEPOINT-обёртки. Нужны там, где сервис делает явный `session.rollback()` (`dry_run` импорта): внутри `db_session`
+такой откат снял бы корневой SAVEPOINT вместе с seed. Приложение собирается той же `_build_application`, что и `app`;
+S3 — `FakeStorage` (объекты в словаре).
 """
 
 import uuid
@@ -11,13 +17,20 @@ from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AsyncExitStack
 from pathlib import Path
 from types import SimpleNamespace
+from typing import Any
 
 import asyncpg
 import httpx
 import pytest
 import pytest_asyncio
 from fastapi import FastAPI
-from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession, create_async_engine
+from sqlalchemy.ext.asyncio import (
+    AsyncConnection,
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
 
 from scripts.seed import run_seed
 from src.api.router import build_api_router
@@ -113,12 +126,10 @@ async def seeded(db_session: AsyncSession) -> AsyncSession:
     return db_session
 
 
-@pytest.fixture
+def _build_application(
-def app(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
+    session_dependency: Callable[[], AsyncIterator[AsyncSession]], monkeypatch: pytest.MonkeyPatch
-    """Приложение на `build_api_router` с dev-эмитентом (`DEBUG` и `AUTH_DEV_ISSUER`) поверх `db_session`.
+) -> FastAPI:
-
+    """Приложение на `build_api_router` с dev-эмитентом (`DEBUG` и `AUTH_DEV_ISSUER`) и переопределённой сессией БД.
-    Override сессии повторяет транзакцию запроса: `begin_nested()` — ошибка в запросе откатывает лишь до SAVEPOINT,
-    данные теста сохраняются; без lifespan, `storage` отсутствует.
 
     Предупреждение «Включён dev-эмитент…» из `build_verifier` (`src.core.auth.logger`) подавлено (T3.1).
     """
@@ -127,15 +138,38 @@ def app(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
     application = FastAPI()
     register_exception_handlers(application)
     application.include_router(build_api_router(settings))
+    application.dependency_overrides[get_async_session] = session_dependency
+    application.state.token_verifier = build_verifier(settings)
+    application.state.storage = None
+    return application
+
+
+@pytest.fixture
+def app(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
+    """Приложение поверх `db_session` (см. `_build_application`); без lifespan, `storage` отсутствует.
+
+    Override сессии повторяет транзакцию запроса: `begin_nested()` — ошибка в запросе откатывает лишь до SAVEPOINT,
+    данные теста сохраняются.
+    """
 
     async def override_session() -> AsyncIterator[AsyncSession]:
         async with db_session.begin_nested():  # как транзакция запроса: ошибка → откат до SAVEPOINT
             yield db_session
 
-    application.dependency_overrides[get_async_session] = override_session
+    return _build_application(override_session, monkeypatch)
-    application.state.token_verifier = build_verifier(settings)
+
-    application.state.storage = None
+
-    return application
+async def _client_factory(application: FastAPI, stack: AsyncExitStack, email: str) -> httpx.AsyncClient:
+    """Клиент с dev-токеном пользователя `email`, `base_url=.../api/v1`; закрывается вместе со `stack`."""
+    transport = httpx.ASGITransport(app=application)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as anonymous:
+        response = await anonymous.post("/api/v1/dev/token", json={"email": email})
+    assert response.status_code == 200, response.text  # noqa: PLR2004,S101
+    token = response.json()["access_token"]
+    client = httpx.AsyncClient(
+        transport=transport, base_url="http://test/api/v1", headers={"Authorization": f"Bearer {token}"}
+    )
+    return await stack.enter_async_context(client)
 
 
 @pytest_asyncio.fixture
@@ -148,14 +182,66 @@ async def auth_client(app: FastAPI, seeded: AsyncSession) -> AsyncIterator[AuthC
     async with AsyncExitStack() as stack:
 
         async def factory(email: str) -> httpx.AsyncClient:
-            transport = httpx.ASGITransport(app=app)
+            return await _client_factory(app, stack, email)
-            async with httpx.AsyncClient(transport=transport, base_url="http://test") as anonymous:
+
-                response = await anonymous.post("/api/v1/dev/token", json={"email": email})
+        yield factory
-            assert response.status_code == 200, response.text  # noqa: PLR2004,S101
+
-            token = response.json()["access_token"]
+
-            client = httpx.AsyncClient(
+class FakeStorage:
-                transport=transport, base_url="http://test/api/v1", headers={"Authorization": f"Bearer {token}"}
+    """Заглушка `S3Storage`: тот же интерфейс `bucket`/`put_object(obj_name, file_data)`, объекты — в словаре."""
-            )
+
-            return await stack.enter_async_context(client)
+    def __init__(self, bucket: str = "arm-ka") -> None:
+        self.bucket = bucket
+        self.objects: dict[str, bytes] = {}
+
+    async def put_object(self, obj_name: str, file_data: bytes) -> None:
+        self.objects[obj_name] = file_data
+
+
+@pytest_asyncio.fixture
+async def committed_sessionmaker() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """Отдельная временная БД: `alembic upgrade head` и seed (dev) ЗАФИКСИРОВАНЫ; удаляется после теста."""
+    name = f"armcore_import_{uuid.uuid4().hex[:12]}"
+    url = (await create_scratch_db(name)).render_as_string(hide_password=False)
+    engine = create_async_engine(url, connect_args={"server_settings": {"search_path": "dkb"}})
+    try:
+        await alembic_async(url, "upgrade", "head")
+        maker = async_sessionmaker(engine, expire_on_commit=False)
+        async with maker() as session, session.begin():
+            await run_seed(session, dev=True)
+        yield maker
+    finally:
+        await engine.dispose()
+        await drop_scratch_db(name)
+
+
+@pytest.fixture
+def committed_app(committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch) -> FastAPI:
+    """Приложение с сессией запроса как в проде (`session.begin()` без SAVEPOINT); `storage` — `FakeStorage`."""
+
+    async def production_like_session() -> AsyncIterator[AsyncSession]:
+        async with committed_sessionmaker() as session, session.begin():
+            yield session
+
+    application = _build_application(production_like_session, monkeypatch)
+    application.state.storage = FakeStorage()
+    return application
+
+
+@pytest.fixture
+def committed_storage(committed_app: FastAPI) -> FakeStorage:
+    """`FakeStorage` приложения `committed_app`: что реально «записано в S3»."""
+    storage: Any = committed_app.state.storage
+    assert isinstance(storage, FakeStorage)  # noqa: S101
+    return storage
+
+
+@pytest_asyncio.fixture
+async def committed_client(committed_app: FastAPI) -> AsyncIterator[AuthClientFactory]:
+    """Фабрика клиентов `committed_app` (как `auth_client`, но на БД с зафиксированным seed)."""
+    async with AsyncExitStack() as stack:
+
+        async def factory(email: str) -> httpx.AsyncClient:
+            return await _client_factory(committed_app, stack, email)
 
         yield factory

api/tests/test_api_directory_import.py

@@ -1,33 +1,24 @@
 """API импорта справочников из выгрузки ЕЦП (spec directory-import, audit-trail; `05-domain-rules.md` §13.3).
 
 Тесты боевого импорта идут поверх фикстуры `app` (транзакция теста с SAVEPOINT). `dry_run` проверяется на отдельной
-временной БД без SAVEPOINT-обёртки (`committed_app`): сервис делает явный `session.rollback()`, а внутри
+временной БД без SAVEPOINT-обёртки (`committed_client`, `committed_sessionmaker`, `committed_storage` из `conftest.py`):
-SAVEPOINT-обёртки `db_session` такой откат снимает корневую точку сохранения сессии теста вместе с seed — поведение
+сервис делает явный `session.rollback()`, а внутри SAVEPOINT-обёртки `db_session` такой откат снимает корневую точку
-отличалось бы от прода (там откатывается только транзакция запроса). Поэтому `dry_run` гоняется так же, как в проде:
+сохранения сессии теста вместе с seed — поведение отличалось бы от прода (там откатывается только транзакция запроса).
-сессия запроса `async with session.begin()`, данные seed зафиксированы.
+Поэтому `dry_run` гоняется так же, как в проде: сессия запроса `async with session.begin()`, данные seed зафиксированы.
 """
 
 import re
 import uuid
-from collections.abc import AsyncIterator, Awaitable, Callable
-from contextlib import AsyncExitStack
 from pathlib import Path
-from types import SimpleNamespace
 from typing import Any
 
 import httpx
 import pytest
-import pytest_asyncio
 from fastapi import FastAPI
 from sqlalchemy import func, select
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from scripts.seed import run_seed
-from src.api.router import build_api_router
-from src.core.auth import build_verifier
 from src.core.config import get_settings
-from src.core.exception_handlers import register_exception_handlers
-from src.db.db import get_async_session
 from src.models.access import AppUser
 from src.models.audit_ex import AuditEvent
 from src.models.directories import (
@@ -41,9 +32,9 @@ from src.models.directories import (
 )
 from src.models.enums import AuditAction, AuditSection, EntityType, LeHistoryEvent, RelatedGroupKind, S3BucketRole
 from src.models.files import S3Object
-from tests.conftest import AuthClientFactory
+from src.services.directory_import import MAX_FILENAME_LENGTH, safe_filename
+from tests.conftest import AuthClientFactory, FakeStorage
 from tests.factories import make_company_group, make_legal_entity
-from tests.pg_helpers import alembic_async, create_scratch_db, drop_scratch_db
 
 HTTP_OK = 200
 HTTP_FORBIDDEN = 403
@@ -101,17 +92,6 @@ HEADER = ["ИНН", "iD ГК", "Базовый РФ ГК в CRM", "Группа
 KRASNODAR = "Краснодарский региональный филиал"
 
 
-class FakeStorage:
-    """Заглушка `S3Storage`: тот же интерфейс `bucket`/`put_object(obj_name, file_data)`, объекты — в словаре."""
-
-    def __init__(self, bucket: str = "arm-ka") -> None:
-        self.bucket = bucket
-        self.objects: dict[str, bytes] = {}
-
-    async def put_object(self, obj_name: str, file_data: bytes) -> None:
-        self.objects[obj_name] = file_data
-
-
 @pytest.fixture
 def storage(app: FastAPI) -> FakeStorage:
     fake = FakeStorage()
@@ -450,65 +430,11 @@ async def test_import_requires_directory_admin(auth_client: AuthClientFactory, s
 # --------------------------------------------------------------------------- dry_run (как в проде, без SAVEPOINT)
 
 
-ClientFactory = Callable[[str], Awaitable[httpx.AsyncClient]]
-
-
-@pytest_asyncio.fixture
-async def committed_sessionmaker() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
-    """Отдельная временная БД: `alembic upgrade head` и seed (dev) ЗАФИКСИРОВАНЫ; удаляется после теста."""
-    name = f"armcore_import_{uuid.uuid4().hex[:12]}"
-    url = (await create_scratch_db(name)).render_as_string(hide_password=False)
-    engine = create_async_engine(url, connect_args={"server_settings": {"search_path": "dkb"}})
-    try:
-        await alembic_async(url, "upgrade", "head")
-        maker = async_sessionmaker(engine, expire_on_commit=False)
-        async with maker() as session, session.begin():
-            await run_seed(session, dev=True)
-        yield maker
-    finally:
-        await engine.dispose()
-        await drop_scratch_db(name)
-
-
-@pytest_asyncio.fixture
-async def committed_client(
-    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
-) -> AsyncIterator[ClientFactory]:
-    """Клиенты приложения, сессия запроса которого устроена как в проде (`src.db.db.get_async_session`)."""
-    monkeypatch.setattr("src.core.auth.logger", SimpleNamespace(warning=lambda *args, **kwargs: None))
-    settings = get_settings().model_copy(update={"DEBUG": True, "AUTH_DEV_ISSUER": True})
-    application = FastAPI()
-    register_exception_handlers(application)
-    application.include_router(build_api_router(settings))
-
-    async def production_like_session() -> AsyncIterator[AsyncSession]:
-        async with committed_sessionmaker() as session, session.begin():
-            yield session
-
-    application.dependency_overrides[get_async_session] = production_like_session
-    application.state.token_verifier = build_verifier(settings)
-    application.state.storage = FakeStorage()
-
-    async with AsyncExitStack() as stack:
-
-        async def factory(email: str) -> httpx.AsyncClient:
-            transport = httpx.ASGITransport(app=application)
-            async with httpx.AsyncClient(transport=transport, base_url="http://test") as anonymous:
-                response = await anonymous.post("/api/v1/dev/token", json={"email": email})
-            assert response.status_code == HTTP_OK, response.text
-            token = response.json()["access_token"]
-            client = httpx.AsyncClient(
-                transport=transport, base_url="http://test/api/v1", headers={"Authorization": f"Bearer {token}"}
-            )
-            return await stack.enter_async_context(client)
-
-        yield factory
-        assert isinstance(application.state.storage, FakeStorage)
-
-
 @pytest.mark.integration
 async def test_dry_run_leaves_no_trace_and_repeats_identically(
-    committed_client: ClientFactory, committed_sessionmaker: async_sessionmaker[AsyncSession]
+    committed_client: AuthClientFactory,
+    committed_sessionmaker: async_sessionmaker[AsyncSession],
+    committed_storage: FakeStorage,
 ) -> None:
     """`dry_run` (по умолчанию): отчёт как у боевого, ни строк справочника, ни аудита, ни `s3_object`; повтор `dry_run`
     даёт тот же отчёт (Review Focus 4: коды из последовательностей расходуются, но не видны и не ломают прогон)."""
@@ -529,6 +455,7 @@ async def test_dry_run_leaves_no_trace_and_repeats_identically(
     assert await counts() == before
     second = await _report(client, "dry.csv", content, dry_run=True)
     assert await counts() == before
+    assert committed_storage.objects == {}  # T10.1: dry_run не пишет файл в S3
     assert second == first
     assert {key: first[key] for key in COUNTER_KEYS} == {
         **dict.fromkeys(COUNTER_KEYS, 0),
@@ -543,6 +470,7 @@ async def test_dry_run_leaves_no_trace_and_repeats_identically(
 
     real = await _report(client, "dry.csv", content, dry_run=False)
 
+    assert len(committed_storage.objects) == 1  # T10.1: боевой прогон — ровно один объект
     assert real == first  # отчёт dry_run идентичен боевому (§13.3)
     after = await counts()
     assert {table: after[table] - before[table] for table in after} == {
@@ -556,3 +484,103 @@ async def test_dry_run_leaves_no_trace_and_repeats_identically(
         "audit_event": 1,
         "s3_object": 1,
     }
+
+
+@pytest.mark.integration
+@pytest.mark.fixtures
+async def test_dry_run_on_russkiy_biskvit_file_leaves_no_trace(
+    committed_client: AuthClientFactory,
+    committed_sessionmaker: async_sessionmaker[AsyncSession],
+    committed_storage: FakeStorage,
+    fixtures_dir: Path,
+) -> None:
+    """T10.2: `dry_run` на реальном xlsx (не только csv): 200, 21 строка, таблицы и S3 без изменений."""
+    client = await committed_client(ADMIN)
+    content = (fixtures_dir / "directories" / RB_FILE).read_bytes()
+    async with committed_sessionmaker() as session:
+        before = await _counts(session)
+
+    report = await _report(client, RB_FILE, content, dry_run=True, content_type=XLSX_TYPE)
+
+    assert report["rows"] == RB_ROWS
+    assert (report["groups_created"], report["entities_created"]) == (1, RB_ENTITIES)
+    async with committed_sessionmaker() as session:
+        assert await _counts(session) == before
+    assert committed_storage.objects == {}
+
+
+@pytest.mark.integration
+async def test_dry_run_with_update_leaves_history_and_audit_untouched(
+    committed_client: AuthClientFactory, committed_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """T10.3: путь `update` пишет снимки истории и аудит ДО отката — после `dry_run` их нет, имена прежние."""
+    client = await committed_client(ADMIN)
+    await _report(client, "first.csv", _csv(_row("7708000001", "1-DRYUP", "ГК Старая", "ООО Старое")), dry_run=False)
+    changed = _csv(_row("7708000001", "1-DRYUP", "ГК Новая", "ООО Новое"))
+
+    async def state() -> tuple[dict[str, int], list[str]]:
+        async with committed_sessionmaker() as session:
+            names = [
+                *(await session.execute(select(CompanyGroup.name).where(CompanyGroup.crm_id == "1-DRYUP"))).scalars(),
+                *(await session.execute(select(LegalEntity.name).where(LegalEntity.inn == "7708000001"))).scalars(),
+            ]
+            return await _counts(session), names
+
+    before, names_before = await state()
+    assert names_before == ["ГК Старая", "ООО Старое"]
+
+    report = await _report(client, "changed.csv", changed, dry_run=True, on_existing="update")
+
+    assert _finals(report) == {2: "LE_UPDATED"}  # план применялся: история и аудит изменения были записаны и откатились
+    assert (report["groups_updated"], report["entities_updated"]) == (1, 1)
+    after, names_after = await state()
+    assert after == before
+    assert after["company_group_history"] == 0
+    assert names_after == names_before
+
+
+# --------------------------------------------------------------------------- имя файла (T10.4)
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("a\x00b.csv", "ab.csv"),
+        ("\x00", "import"),
+        ("../../etc/x\x07y.xlsx", "xy.xlsx"),
+        ("C:\\dir\\f.xlsx", "f.xlsx"),
+        ("..", "import"),
+        (None, "import"),
+        ("тест.csv", "тест.csv"),
+    ],
+)
+def test_safe_filename_drops_control_characters(raw: str | None, expected: str) -> None:
+    assert safe_filename(raw) == expected
+
+
+def test_safe_filename_caps_length_keeping_extension() -> None:
+    long_name = safe_filename("я" * 1500 + ".xlsx")
+    assert len(long_name) == MAX_FILENAME_LENGTH
+    assert long_name.endswith(".xlsx")
+    # гигантское «расширение» отбрасывается
+    assert len(safe_filename("x" * 1500 + "." + "e" * 500)) == MAX_FILENAME_LENGTH
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("raw_name", ["a\x00b.csv", "я" * 1500 + ".csv"], ids=["nul", "over-1024-bytes"])
+async def test_hostile_filename_does_not_produce_500(
+    auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage, raw_name: str
+) -> None:
+    """T10.4: NUL и имя длиннее 1024 байт в `filename` multipart — не 500: имя очищено, ключ S3 в пределах лимита."""
+    client = await auth_client(ADMIN)
+
+    report = await _report(client, raw_name, _csv(_row("7709000001", "1-NAME", "ГК Имя", "ООО Имя")), dry_run=False)
+
+    assert report["entities_created"] == 1
+    (event,) = await _import_events(seeded)
+    assert event.description is not None
+    assert "\x00" not in event.description
+    assert len(event.description) <= MAX_FILENAME_LENGTH
+    assert event.description.endswith(".csv")
+    (key,) = storage.objects
+    assert len(key.encode()) <= 1024  # noqa: PLR2004 - предел длины ключа S3