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
This commit is contained in:
Andrey Epifancev
parent
58075e9445
commit
c793b42fba
@ -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."""
|
||||
name = PurePosixPath((filename or "").replace("\\", "/")).name.strip()
|
||||
return name if name not in {"", ".", ".."} else DEFAULT_FILENAME
|
||||
"""Имя файла без пути (клиент может прислать `C:\\…\\x.xlsx` или `../x.csv`): для формата, аудита и ключа S3.
|
||||
|
||||
Управляющие символы (в т. ч. 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:
|
||||
|
||||
@ -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 app(db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
|
||||
"""Приложение на `build_api_router` с dev-эмитентом (`DEBUG` и `AUTH_DEV_ISSUER`) поверх `db_session`.
|
||||
|
||||
Override сессии повторяет транзакцию запроса: `begin_nested()` — ошибка в запросе откатывает лишь до SAVEPOINT,
|
||||
данные теста сохраняются; без lifespan, `storage` отсутствует.
|
||||
def _build_application(
|
||||
session_dependency: Callable[[], AsyncIterator[AsyncSession]], monkeypatch: pytest.MonkeyPatch
|
||||
) -> FastAPI:
|
||||
"""Приложение на `build_api_router` с dev-эмитентом (`DEBUG` и `AUTH_DEV_ISSUER`) и переопределённой сессией БД.
|
||||
|
||||
Предупреждение «Включён 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
|
||||
application.state.token_verifier = build_verifier(settings)
|
||||
application.state.storage = None
|
||||
return application
|
||||
return _build_application(override_session, monkeypatch)
|
||||
|
||||
|
||||
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)
|
||||
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)
|
||||
return await _client_factory(app, stack, email)
|
||||
|
||||
yield factory
|
||||
|
||||
|
||||
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_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
|
||||
|
||||
@ -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-обёртки `db_session` такой откат снимает корневую точку сохранения сессии теста вместе с seed — поведение
|
||||
отличалось бы от прода (там откатывается только транзакция запроса). Поэтому `dry_run` гоняется так же, как в проде:
|
||||
сессия запроса `async with session.begin()`, данные seed зафиксированы.
|
||||
временной БД без SAVEPOINT-обёртки (`committed_client`, `committed_sessionmaker`, `committed_storage` из `conftest.py`):
|
||||
сервис делает явный `session.rollback()`, а внутри SAVEPOINT-обёртки `db_session` такой откат снимает корневую точку
|
||||
сохранения сессии теста вместе с 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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user