From 5526f90cb38abae22a9879f480f497a82c2b6797 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A0=D0=B0=D0=B8=D1=81=20=D0=AE=D1=81=D1=83=D0=BF=D0=B0?= =?UTF-8?q?=D0=BB=D0=B8=D0=B5=D0=B2?= Date: Sat, 23 May 2026 20:24:23 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=20=D0=BF=D0=BE=D0=BB=D0=BB=D0=B5=D1=80=20=D0=BD=D0=B0?= =?UTF-8?q?=D0=BA=D0=BB=D0=B0=D0=B4=D0=BD=D0=BE=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../20260524_034_add_cdek_polling_columns.py | 32 +++ .../delivery_providers/cdek/client.py | 90 ++++++ .../delivery_providers/cdek/order_mapper.py | 72 +++++ app/config.py | 6 + app/domain/cdek_polling.py | 11 + app/repositories/order/models.py | 5 + app/repositories/order/repository.py | 68 ++++- app/services/aggregator.py | 8 - app/services/waybill_poller.py | 169 ++++++++++++ app/workers/__init__.py | 0 app/workers/waybill_poller.py | 81 ++++++ config.example.yaml | 4 + docker-compose.yml | 12 + spec/index.md | 5 +- spec/tasks/033_add_waybill_polling_worker.md | 94 +++++++ .../cdek/test_order_info_client.py | 173 ++++++++++++ .../cdek/test_order_mapper.py | 75 +++++ tests/domain/test_cdek_polling.py | 19 ++ tests/repositories/order/test_repository.py | 252 ++++++++++++++--- tests/services/test_tbank_notifications.py | 19 +- tests/services/test_waybill_poller.py | 259 ++++++++++++++++++ tests/workers/__init__.py | 0 tests/workers/test_waybill_poller_main.py | 88 ++++++ 23 files changed, 1481 insertions(+), 61 deletions(-) create mode 100644 alembic/versions/20260524_034_add_cdek_polling_columns.py create mode 100644 app/domain/cdek_polling.py create mode 100644 app/services/waybill_poller.py create mode 100644 app/workers/__init__.py create mode 100644 app/workers/waybill_poller.py create mode 100644 spec/tasks/033_add_waybill_polling_worker.md create mode 100644 tests/adapters/delivery_providers/cdek/test_order_info_client.py create mode 100644 tests/domain/test_cdek_polling.py create mode 100644 tests/services/test_waybill_poller.py create mode 100644 tests/workers/__init__.py create mode 100644 tests/workers/test_waybill_poller_main.py diff --git a/alembic/versions/20260524_034_add_cdek_polling_columns.py b/alembic/versions/20260524_034_add_cdek_polling_columns.py new file mode 100644 index 0000000..d4322b3 --- /dev/null +++ b/alembic/versions/20260524_034_add_cdek_polling_columns.py @@ -0,0 +1,32 @@ +"""Add CDEK polling columns to orders table.""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260524_034" +down_revision: str | None = "20260523_033" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "orders", + sa.Column("cdek_order_status", sa.String(length=64), nullable=True), + ) + op.add_column( + "orders", + sa.Column( + "cdek_polled_at", + sa.DateTime(timezone=True), + nullable=True, + ), + ) + + +def downgrade() -> None: + op.drop_column("orders", "cdek_polled_at") + op.drop_column("orders", "cdek_order_status") diff --git a/app/adapters/delivery_providers/cdek/client.py b/app/adapters/delivery_providers/cdek/client.py index 43e00fd..0ceeb5b 100644 --- a/app/adapters/delivery_providers/cdek/client.py +++ b/app/adapters/delivery_providers/cdek/client.py @@ -19,13 +19,17 @@ from app.adapters.delivery_providers.cdek.mapper import ( map_cdek_response_for_tariff_code, ) from app.adapters.delivery_providers.cdek.order_mapper import ( + CDEKOrderInfo, CDEKOrderMappingError, CDEKOrderRegistrationResult, + CDEKWaybillInfo, centimeters_string_to_int, kilograms_string_to_grams, map_cdek_existing_order_response, + map_cdek_order_info_response, map_cdek_order_request, map_cdek_order_response, + map_cdek_waybill_info_response, resolve_cdek_city_code, ) from app.cities import cities_map @@ -61,6 +65,7 @@ class CDEKClient: normalized_base_url = base_url.rstrip("/") self._tariff_url = f"{normalized_base_url}/calculator/tarifflist" self._orders_url = f"{normalized_base_url}/orders" + self._print_orders_url = f"{normalized_base_url}/print/orders" self._timeout_seconds = timeout_seconds self._retry_attempts = retry_attempts self._retry_backoff_seconds = retry_backoff_seconds @@ -222,6 +227,85 @@ class CDEKClient: raise CDEKClientError("CDEK order registration failed unexpectedly.") + async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo: + url = f"{self._orders_url}/{cdek_order_uuid}" + raw_payload = await self._get_json( + url, + failure_message="CDEK get order failed", + ) + try: + return map_cdek_order_info_response(raw_payload) + except CDEKOrderMappingError as exc: + raise CDEKClientError( + "CDEK get order response payload is invalid." + ) from exc + + async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo: + url = f"{self._print_orders_url}/{cdek_waybill_uuid}" + raw_payload = await self._get_json( + url, + failure_message="CDEK get waybill failed", + ) + try: + return map_cdek_waybill_info_response(raw_payload) + except CDEKOrderMappingError as exc: + raise CDEKClientError( + "CDEK get waybill response payload is invalid." + ) from exc + + async def _get_json(self, url: str, *, failure_message: str) -> dict[str, Any]: + for attempt in range(self._retry_attempts + 1): + try: + token = await self._auth_client.get_access_token() + response = await self._http_client.get( + url, + headers={"Authorization": f"Bearer {token}"}, + timeout=self._timeout_seconds, + ) + except (httpx.TimeoutException, httpx.TransportError) as exc: + if attempt < self._retry_attempts: + await self._sleep(self._retry_delay(attempt)) + continue + raise CDEKClientError( + f"{failure_message} after retry attempts." + ) from exc + + if self._should_retry(response.status_code): + if attempt < self._retry_attempts: + await self._sleep(self._retry_delay(attempt)) + continue + raise CDEKClientError( + f"{failure_message} with retriable status " + f"{response.status_code}." + ) + + if 400 <= response.status_code < 500: + log.warning( + "cdek_get_request_rejected", + url=url, + status_code=response.status_code, + response_body=_response_text_or_none(response), + ) + raise CDEKRequestError( + f"{failure_message} with status {response.status_code}." + ) + + try: + response.raise_for_status() + raw_payload = response.json() + except (httpx.HTTPError, TypeError, ValueError) as exc: + raise CDEKClientError( + f"{failure_message}: invalid response payload." + ) from exc + + if not isinstance(raw_payload, dict): + raise CDEKClientError( + f"{failure_message}: payload must be a JSON object." + ) + return raw_payload + + raise CDEKClientError(f"{failure_message} unexpectedly.") + def _retry_delay(self, attempt: int) -> float: return self._retry_backoff_seconds * (attempt + 1) @@ -361,6 +445,12 @@ class CDEKProvider(DeliveryProvider): ) -> CDEKOrderRegistrationResult: return await self._client.register_order(request) + async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo: + return await self._client.get_order(cdek_order_uuid) + + async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo: + return await self._client.get_waybill(cdek_waybill_uuid) + def _response_json_or_none(response: httpx.Response) -> object | None: try: diff --git a/app/adapters/delivery_providers/cdek/order_mapper.py b/app/adapters/delivery_providers/cdek/order_mapper.py index 1872a90..376099b 100644 --- a/app/adapters/delivery_providers/cdek/order_mapper.py +++ b/app/adapters/delivery_providers/cdek/order_mapper.py @@ -29,6 +29,19 @@ class CDEKOrderRegistrationResult: waybill_url: str | None +@dataclass(frozen=True) +class CDEKOrderInfo: + order_uuid: str + status_code: str | None + waybill_uuid: str | None + + +@dataclass(frozen=True) +class CDEKWaybillInfo: + waybill_uuid: str + url: str | None + + def map_cdek_order_request(request: InitPaymentRequest) -> dict[str, Any]: payload: dict[str, Any] = { "number": request.order_uuid, @@ -89,6 +102,65 @@ def map_cdek_existing_order_response( ) +def map_cdek_order_info_response(payload: dict[str, Any]) -> CDEKOrderInfo: + entity = payload.get("entity") + if not isinstance(entity, dict): + raise CDEKOrderMappingError( + "CDEK order info response must include entity object." + ) + + order_uuid = entity.get("uuid") + if not isinstance(order_uuid, str) or not order_uuid: + raise CDEKOrderMappingError( + "CDEK order info response must include entity.uuid." + ) + + status_code = _latest_status_code(entity.get("statuses")) + waybill_uuid, _ = _extract_waybill(payload) + return CDEKOrderInfo( + order_uuid=order_uuid, + status_code=status_code, + waybill_uuid=waybill_uuid, + ) + + +def map_cdek_waybill_info_response(payload: dict[str, Any]) -> CDEKWaybillInfo: + entity = payload.get("entity") + if not isinstance(entity, dict): + raise CDEKOrderMappingError( + "CDEK waybill info response must include entity object." + ) + + waybill_uuid = entity.get("uuid") + if not isinstance(waybill_uuid, str) or not waybill_uuid: + raise CDEKOrderMappingError( + "CDEK waybill info response must include entity.uuid." + ) + + url_value = entity.get("url") + waybill_url = url_value if isinstance(url_value, str) and url_value else None + return CDEKWaybillInfo(waybill_uuid=waybill_uuid, url=waybill_url) + + +def _latest_status_code(statuses: object) -> str | None: + if not isinstance(statuses, list) or not statuses: + return None + dated: list[tuple[str, str]] = [] + for entry in statuses: + if not isinstance(entry, dict): + continue + code = entry.get("code") + if not isinstance(code, str) or not code: + continue + date_time = entry.get("date_time") + date_time_key = date_time if isinstance(date_time, str) else "" + dated.append((date_time_key, code)) + if not dated: + return None + dated.sort(key=lambda item: item[0]) + return dated[-1][1] + + def _extract_waybill(payload: dict[str, Any]) -> tuple[str | None, str | None]: related_entities = payload.get("related_entities") if not isinstance(related_entities, list): diff --git a/app/config.py b/app/config.py index b44801d..e8b3c0c 100644 --- a/app/config.py +++ b/app/config.py @@ -116,6 +116,11 @@ class ObservabilityConfig(BaseModel): otlp_insecure: bool = True +class WaybillPollerConfig(BaseModel): + interval_seconds: float = Field(default=30.0, gt=0) + batch_size: int = Field(default=50, gt=0) + + class Settings(BaseSettings): model_config = SettingsConfigDict( extra="ignore", @@ -132,6 +137,7 @@ class Settings(BaseSettings): default_factory=AddressSuggestionsConfig ) observability: ObservabilityConfig + waybill_poller: WaybillPollerConfig = Field(default_factory=WaybillPollerConfig) @classmethod def settings_customise_sources( diff --git a/app/domain/cdek_polling.py b/app/domain/cdek_polling.py new file mode 100644 index 0000000..66f08fa --- /dev/null +++ b/app/domain/cdek_polling.py @@ -0,0 +1,11 @@ +"""Pure rules for CDEK order polling lifecycle.""" + +TERMINAL_ORDER_STATUSES: frozenset[str] = frozenset( + {"INVALID", "DELIVERED", "NOT_DELIVERED", "CANCELLED"} +) + + +def is_terminal_order_status(code: str | None) -> bool: + if code is None: + return False + return code in TERMINAL_ORDER_STATUSES diff --git a/app/repositories/order/models.py b/app/repositories/order/models.py index 7317da9..2ba464d 100644 --- a/app/repositories/order/models.py +++ b/app/repositories/order/models.py @@ -41,8 +41,13 @@ class Order(Base): payment_status: Mapped[str | None] = mapped_column(String(64), nullable=True) tbank_payment_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) cdek_order_uuid: Mapped[str | None] = mapped_column(String(128), nullable=True) + cdek_order_status: Mapped[str | None] = mapped_column(String(64), nullable=True) cdek_waybill_uuid: Mapped[str | None] = mapped_column(String(128), nullable=True) cdek_waybill_url: Mapped[str | None] = mapped_column(String(2048), nullable=True) + cdek_polled_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), diff --git a/app/repositories/order/repository.py b/app/repositories/order/repository.py index fc2162c..473085c 100644 --- a/app/repositories/order/repository.py +++ b/app/repositories/order/repository.py @@ -1,12 +1,15 @@ """PostgreSQL order repository.""" +from collections.abc import Sequence from contextlib import AbstractAsyncContextManager from dataclasses import dataclass +from datetime import datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from app.domain.cdek_polling import TERMINAL_ORDER_STATUSES from app.repositories.order.models import Order @@ -72,15 +75,72 @@ class OrderRepository: session: AsyncSession, order_uuid: str, cdek_order_uuid: str, - cdek_waybill_uuid: str | None = None, - cdek_waybill_url: str | None = None, ) -> Order | None: order = await self.get_order_by_order_uuid(session, order_uuid) if order is None: return None order.cdek_order_uuid = cdek_order_uuid - order.cdek_waybill_uuid = cdek_waybill_uuid - order.cdek_waybill_url = cdek_waybill_url + await session.flush() + return order + + async def list_orders_pending_waybill( + self, + session: AsyncSession, + *, + limit: int, + ) -> Sequence[Order]: + statement = ( + select(Order) + .where( + Order.cdek_order_uuid.is_not(None), + Order.cdek_waybill_url.is_(None), + ( + Order.cdek_order_status.is_(None) + | Order.cdek_order_status.not_in(TERMINAL_ORDER_STATUSES) + ), + ) + .order_by(Order.cdek_polled_at.asc().nulls_first()) + .limit(limit) + .with_for_update(skip_locked=True) + ) + result = await session.execute(statement) + return result.scalars().all() + + async def record_order_poll( + self, + session: AsyncSession, + *, + order_uuid: str, + order_status: str | None, + waybill_uuid: str | None, + polled_at: datetime, + ) -> Order | None: + order = await self.get_order_by_order_uuid(session, order_uuid) + if order is None: + return None + + order.cdek_order_status = order_status + if waybill_uuid is not None and order.cdek_waybill_uuid is None: + order.cdek_waybill_uuid = waybill_uuid + order.cdek_polled_at = polled_at + await session.flush() + return order + + async def record_waybill_poll( + self, + session: AsyncSession, + *, + order_uuid: str, + waybill_url: str | None, + polled_at: datetime, + ) -> Order | None: + order = await self.get_order_by_order_uuid(session, order_uuid) + if order is None: + return None + + if waybill_url is not None and order.cdek_waybill_url is None: + order.cdek_waybill_url = waybill_url + order.cdek_polled_at = polled_at await session.flush() return order diff --git a/app/services/aggregator.py b/app/services/aggregator.py index 8241ece..2599a7e 100644 --- a/app/services/aggregator.py +++ b/app/services/aggregator.py @@ -141,8 +141,6 @@ class OrderRepositoryProtocol(Protocol): session: object, order_uuid: str, cdek_order_uuid: str, - cdek_waybill_uuid: str | None = None, - cdek_waybill_url: str | None = None, ) -> object | None: ... @@ -411,8 +409,6 @@ class AggregatorService: await self._save_cdek_order_uuid( order_uuid=notification.OrderId, cdek_order_uuid=registration_result.order_uuid, - cdek_waybill_uuid=registration_result.waybill_uuid, - cdek_waybill_url=registration_result.waybill_url, ) return "OK" @@ -492,8 +488,6 @@ class AggregatorService: *, order_uuid: str, cdek_order_uuid: str, - cdek_waybill_uuid: str | None, - cdek_waybill_url: str | None, ) -> None: if self._order_repository is None: raise TBankPaymentNotificationProcessingError( @@ -506,8 +500,6 @@ class AggregatorService: session, order_uuid, cdek_order_uuid, - cdek_waybill_uuid, - cdek_waybill_url, ) if order is None: logger.warning( diff --git a/app/services/waybill_poller.py b/app/services/waybill_poller.py new file mode 100644 index 0000000..7ce03f7 --- /dev/null +++ b/app/services/waybill_poller.py @@ -0,0 +1,169 @@ +"""Background service that polls CDEK for waybill updates.""" + +import asyncio +from collections.abc import Callable, Sequence +from contextlib import AbstractAsyncContextManager +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Protocol + +import structlog + +from app.adapters.delivery_providers.cdek.order_mapper import ( + CDEKOrderInfo, + CDEKWaybillInfo, +) + +logger = structlog.get_logger(__name__) + + +class CDEKOrderInfoAdapterProtocol(Protocol): + async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo: ... + + +class CDEKWaybillInfoAdapterProtocol(Protocol): + async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo: ... + + +class OrderRecord(Protocol): + order_uuid: str + cdek_order_uuid: str | None + cdek_waybill_uuid: str | None + + +class WaybillPollerRepositoryProtocol(Protocol): + def session(self) -> AbstractAsyncContextManager[object]: ... + + async def list_orders_pending_waybill( + self, session: object, *, limit: int + ) -> Sequence[OrderRecord]: ... + + async def record_order_poll( + self, + session: object, + *, + order_uuid: str, + order_status: str | None, + waybill_uuid: str | None, + polled_at: datetime, + ) -> object | None: ... + + async def record_waybill_poll( + self, + session: object, + *, + order_uuid: str, + waybill_url: str | None, + polled_at: datetime, + ) -> object | None: ... + + +@dataclass(frozen=True) +class PollBatchSummary: + processed: int + succeeded: int + failed: int + + +class WaybillPollerService: + def __init__( + self, + *, + order_repository: WaybillPollerRepositoryProtocol, + order_info_adapter: CDEKOrderInfoAdapterProtocol, + waybill_info_adapter: CDEKWaybillInfoAdapterProtocol, + batch_size: int, + datetime_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), + ) -> None: + self._repository = order_repository + self._order_info_adapter = order_info_adapter + self._waybill_info_adapter = waybill_info_adapter + self._batch_size = batch_size + self._datetime_now = datetime_now + + async def poll_once(self) -> PollBatchSummary: + async with self._repository.session() as session: + orders = await self._repository.list_orders_pending_waybill( + session, limit=self._batch_size + ) + succeeded = 0 + failed = 0 + for order in orders: + try: + await self._handle_order(session, order) + succeeded += 1 + except Exception: + failed += 1 + logger.exception( + "waybill_poll_order_failed", + order_uuid=order.order_uuid, + cdek_order_uuid=order.cdek_order_uuid, + cdek_waybill_uuid=order.cdek_waybill_uuid, + ) + return PollBatchSummary( + processed=len(orders), + succeeded=succeeded, + failed=failed, + ) + + async def run_forever( + self, + *, + interval_seconds: float, + stop_event: asyncio.Event, + ) -> None: + while not stop_event.is_set(): + try: + summary = await self.poll_once() + logger.info( + "waybill_poll_tick", + processed=summary.processed, + succeeded=summary.succeeded, + failed=summary.failed, + ) + except Exception: + logger.exception("waybill_poll_tick_failed") + + try: + await asyncio.wait_for( + stop_event.wait(), timeout=interval_seconds + ) + except asyncio.TimeoutError: + continue + + async def _handle_order(self, session: object, order: OrderRecord) -> None: + polled_at = self._datetime_now() + if order.cdek_waybill_uuid is None: + cdek_order_uuid = order.cdek_order_uuid + if cdek_order_uuid is None: + return + info = await self._order_info_adapter.get_order(cdek_order_uuid) + await self._repository.record_order_poll( + session, + order_uuid=order.order_uuid, + order_status=info.status_code, + waybill_uuid=info.waybill_uuid, + polled_at=polled_at, + ) + logger.info( + "waybill_poll_order_result", + order_uuid=order.order_uuid, + cdek_order_uuid=cdek_order_uuid, + cdek_order_status=info.status_code, + cdek_waybill_uuid=info.waybill_uuid, + ) + return + + waybill = await self._waybill_info_adapter.get_waybill(order.cdek_waybill_uuid) + await self._repository.record_waybill_poll( + session, + order_uuid=order.order_uuid, + waybill_url=waybill.url, + polled_at=polled_at, + ) + logger.info( + "waybill_poll_waybill_result", + order_uuid=order.order_uuid, + cdek_waybill_uuid=order.cdek_waybill_uuid, + cdek_waybill_url=waybill.url, + ) diff --git a/app/workers/__init__.py b/app/workers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/workers/waybill_poller.py b/app/workers/waybill_poller.py new file mode 100644 index 0000000..5ef65d5 --- /dev/null +++ b/app/workers/waybill_poller.py @@ -0,0 +1,81 @@ +"""Background worker that polls CDEK for waybill updates.""" + +import asyncio +import signal + +import httpx +import structlog + +from app.adapters.delivery_providers.cdek.auth import CDEKAuthClient +from app.adapters.delivery_providers.cdek.client import CDEKClient +from app.adapters.postgres.engine import ( + create_postgres_engine, + create_postgres_session_factory, +) +from app.config import Settings, get_settings +from app.repositories.order import OrderRepository +from app.runtime.logging import configure_logging +from app.services.waybill_poller import WaybillPollerService + +logger = structlog.get_logger(__name__) + + +async def _run(settings: Settings, stop_event: asyncio.Event) -> None: + http_client = httpx.AsyncClient(timeout=settings.adapter.cdek_timeout_seconds) + auth_client = CDEKAuthClient( + http_client=http_client, + base_url=settings.adapter.cdek_base_url, + client_id=settings.adapter.cdek_client_id, + client_secret=settings.adapter.cdek_client_secret, + timeout_seconds=settings.adapter.cdek_timeout_seconds, + ) + cdek_client = CDEKClient( + http_client=http_client, + auth_client=auth_client, + base_url=settings.adapter.cdek_base_url, + timeout_seconds=settings.adapter.cdek_timeout_seconds, + retry_attempts=settings.adapter.cdek_retry_attempts, + retry_backoff_seconds=settings.adapter.cdek_retry_backoff_seconds, + ) + engine = create_postgres_engine(settings.postgres) + session_factory = create_postgres_session_factory(engine) + repository = OrderRepository(session_factory=session_factory) + service = WaybillPollerService( + order_repository=repository, + order_info_adapter=cdek_client, + waybill_info_adapter=cdek_client, + batch_size=settings.waybill_poller.batch_size, + ) + logger.info( + "waybill_poller_started", + interval_seconds=settings.waybill_poller.interval_seconds, + batch_size=settings.waybill_poller.batch_size, + ) + try: + await service.run_forever( + interval_seconds=settings.waybill_poller.interval_seconds, + stop_event=stop_event, + ) + finally: + await http_client.aclose() + await engine.dispose() + logger.info("waybill_poller_stopped") + + +async def main() -> None: + configure_logging() + settings = get_settings() + stop_event = asyncio.Event() + + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + try: + loop.add_signal_handler(sig, stop_event.set) + except NotImplementedError: + pass + + await _run(settings, stop_event) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/config.example.yaml b/config.example.yaml index 5032924..e7572b1 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -63,3 +63,7 @@ observability: service_name: "g2s-aggregator" otlp_endpoint: "http://localhost:4317" otlp_insecure: true + +waybill_poller: + interval_seconds: 30 + batch_size: 50 diff --git a/docker-compose.yml b/docker-compose.yml index beff946..0fc5113 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -45,5 +45,17 @@ services: - ./config.yaml:/config.yaml command: ["poetry", "run", "alembic", "upgrade", "head"] restart: "no" + waybill-poller: + image: yusupal1ev/g2s-aggregator:0.0.2 + container_name: g2s-aggregator-waybill-poller + depends_on: + postgres: + condition: service_healthy + migrations: + condition: service_completed_successfully + volumes: + - ./config.yaml:/config.yaml + command: ["poetry", "run", "python", "-m", "app.workers.waybill_poller"] + restart: unless-stopped volumes: postgres: diff --git a/spec/index.md b/spec/index.md index a1a554e..600253c 100644 --- a/spec/index.md +++ b/spec/index.md @@ -39,9 +39,10 @@ | 030 | DONE | 2026-04-18 | Add TBank payment notification webhook and CDEK order creation | `spec/tasks/030_add_tbank_payment_notification_webhook.md` | | 031 | TODO | 2026-04-18 | Validate init-payment price with CDEK tariff | `spec/tasks/031_validate_init_payment_price_with_cdek.md` | | 032 | TODO | 2026-05-13 | Rework init-payment contract to camelCase and structured address/contact | `spec/tasks/032_rework_init_payment_contract_camelcase.md` | +| 033 | DONE | 2026-05-23 | Add CDEK waybill polling worker | `spec/tasks/033_add_waybill_polling_worker.md` | ## Summary -- Total: **33** +- Total: **34** - TODO: **2** -- DONE: **31** +- DONE: **32** diff --git a/spec/tasks/033_add_waybill_polling_worker.md b/spec/tasks/033_add_waybill_polling_worker.md new file mode 100644 index 0000000..0ba066f --- /dev/null +++ b/spec/tasks/033_add_waybill_polling_worker.md @@ -0,0 +1,94 @@ +--- +id: 033 +title: Add CDEK waybill polling worker +status: DONE +created: 2026-05-23 +--- + +## Context +`POST /v2/orders` в CDEK асинхронный. Когда `entity.related_entities[type=waybill].uuid` +синхронно приходит в ответе, поле `url` практически всегда пустое — PDF генерится +позже и его url доступен только при `GET /v2/print/orders/{waybill_uuid}` после +перехода накладной в `READY`. Бывают и случаи, когда waybill_uuid тоже не приходит +синхронно, или заказ уходит в терминальный `INVALID` без генерации накладной. +Ранее `_save_cdek_order_uuid` (`app/services/aggregator.py`) писал waybill-поля из +синхронного ответа POST, из-за чего запись прыгала между двумя источниками и +требовала покрытия граничных случаев в двух местах. + +## Goal +Перевести запись waybill-данных под единоличную ответственность отдельного +фонового worker-а. При регистрации заказа сохранять только `cdek_order_uuid`, +а waybill_uuid / waybill_url пополнять поллингом до тех пор, пока заказ не +дойдёт до waybill_url IS NOT NULL либо до терминального статуса. + +## Constraints +- Запуск — отдельный процесс `python -m app.workers.waybill_poller`, отдельный + сервис в `docker-compose.yml`, одна реплика. +- Никаких записей waybill-полей при регистрации заказа: `_save_cdek_order_uuid` + должен сохранять только `cdek_order_uuid`. +- Поллер — единственный writer полей `cdek_order_status`, `cdek_waybill_uuid`, + `cdek_waybill_url`, `cdek_polled_at`. +- На терминальных статусах заказа (`INVALID`, `DELIVERED`, `NOT_DELIVERED`, + `CANCELLED`) опросы по этому заказу прекращаются. +- Бизнес-логика терминальности — pure функция в `app/domain/cdek_polling.py`, + без зависимостей на репозиторий/адаптер. +- Соблюсти слои AGENTS.md: новые HTTP-вызовы CDEK живут только в адаптере, + работа с БД — только в репозитории, оркестрация — в сервисе. + +## Acceptance criteria +- При успешной регистрации заказа в БД заполняется только `cdek_order_uuid`; + `cdek_waybill_uuid` и `cdek_waybill_url` остаются `NULL`. +- Worker раз в `waybill_poller.interval_seconds` секунд выбирает заказы с + `cdek_order_uuid IS NOT NULL AND cdek_waybill_url IS NULL` и нетерминальным + `cdek_order_status` (с упорядочиванием `cdek_polled_at NULLS FIRST` и лимитом + `waybill_poller.batch_size`). +- Если у заказа `cdek_waybill_uuid IS NULL`, worker делает `GET /v2/orders/{uuid}` + и сохраняет последний статус заказа + waybill_uuid из `related_entities`. +- Если `cdek_waybill_uuid IS NOT NULL` и url пуст, worker делает + `GET /v2/print/orders/{waybill_uuid}` и сохраняет `entity.url`. +- Ошибка CDEK по одному заказу логируется и не валит весь батч. +- Worker корректно завершает работу по SIGTERM/SIGINT: закрывает HTTP-клиент и + async engine. + +## Definition of Done +- [ ] Миграция `20260524_034_add_cdek_polling_columns.py` добавляет + `cdek_order_status` и `cdek_polled_at`; модель синхронизирована. +- [ ] `_save_cdek_order_uuid` и `OrderRepository.mark_cdek_order_registered` + больше не принимают waybill-поля. +- [ ] `CDEKClient.get_order` / `get_waybill` и соответствующие мапперы + (`map_cdek_order_info_response`, `map_cdek_waybill_info_response`, + dataclasses `CDEKOrderInfo`, `CDEKWaybillInfo`) реализованы; `CDEKProvider` + предоставляет обёртки. +- [ ] `app/domain/cdek_polling.py` содержит `TERMINAL_ORDER_STATUSES` и + `is_terminal_order_status`. +- [ ] `OrderRepository.list_orders_pending_waybill`, `record_order_poll`, + `record_waybill_poll` реализованы. +- [ ] `WaybillPollerService.poll_once` и `run_forever` реализованы. +- [ ] `app/workers/waybill_poller.py` поднимает зависимости, поддерживает + graceful shutdown по сигналу. +- [ ] `config.example.yaml` содержит секцию `waybill_poller`, `app/config.py` — + `WaybillPollerConfig`. +- [ ] `docker-compose.yml` содержит сервис `waybill-poller`. + +## Tests +- `tests/domain/test_cdek_polling.py` — терминальные/нетерминальные статусы. +- `tests/adapters/delivery_providers/cdek/test_order_mapper.py` — мапперы + ответов GET /v2/orders/{uuid} и GET /v2/print/orders/{uuid}. +- `tests/adapters/delivery_providers/cdek/test_order_info_client.py` — + `CDEKClient.get_order` и `get_waybill`: 200/4xx/5xx, ретраи, парсинг. +- `tests/repositories/order/test_repository.py` — `list_orders_pending_waybill`, + `record_order_poll`, `record_waybill_poll` (фильтр, упорядочивание, защита + waybill_uuid/waybill_url от перезаписи). +- `tests/services/test_waybill_poller.py` — `poll_once` на стабах (два пути, + терминальный статус, изоляция ошибок), `run_forever` завершается по + stop_event. +- `tests/services/test_tbank_notifications.py` — регистрация заказа больше не + сохраняет waybill-поля. +- `tests/workers/test_waybill_poller_main.py` — `_run` собирает зависимости и + завершается по stop_event. + +## Commands +- `poetry run pytest -q` +- `poetry run alembic upgrade head` +- `poetry run python -m app.workers.waybill_poller` +- `python3 spec/gen_spec_index.py --check` diff --git a/tests/adapters/delivery_providers/cdek/test_order_info_client.py b/tests/adapters/delivery_providers/cdek/test_order_info_client.py new file mode 100644 index 0000000..f5c2107 --- /dev/null +++ b/tests/adapters/delivery_providers/cdek/test_order_info_client.py @@ -0,0 +1,173 @@ +import asyncio +from typing import Any + +import httpx +import pytest + +from app.adapters.delivery_providers.cdek.client import ( + CDEKClient, + CDEKClientError, + CDEKRequestError, +) +from app.adapters.delivery_providers.cdek.order_mapper import ( + CDEKOrderInfo, + CDEKWaybillInfo, +) + + +class StubAuthClient: + async def get_access_token(self) -> str: + return "test-token" + + +class SequenceHTTPClient: + def __init__(self, results: list[Any]) -> None: + self._results = results + self.calls: list[dict[str, Any]] = [] + + def _next_result(self) -> Any: + return self._results[len(self.calls) - 1] + + async def get( + self, + url: str, + *, + headers: dict[str, str] | None = None, + timeout: float | None = None, + ) -> httpx.Response: + self.calls.append( + {"method": "GET", "url": url, "headers": headers, "timeout": timeout} + ) + result = self._next_result() + if isinstance(result, Exception): + raise result + return result + + +def _make_client(http_client: SequenceHTTPClient, **kwargs: Any) -> CDEKClient: + kwargs.setdefault("retry_attempts", 0) + return CDEKClient( + http_client=http_client, # type: ignore[arg-type] + auth_client=StubAuthClient(), # type: ignore[arg-type] + base_url="https://api.cdek.test/v2", + timeout_seconds=7.5, + **kwargs, + ) + + +def test_get_order_parses_status_and_waybill_uuid() -> None: + response = httpx.Response( + 200, + json={ + "entity": { + "uuid": "cdek-order-uuid", + "statuses": [ + {"code": "ACCEPTED", "date_time": "2026-05-24T10:00:00+0000"} + ], + }, + "related_entities": [ + {"type": "waybill", "uuid": "waybill-uuid-1"} + ], + }, + request=httpx.Request("GET", "https://api.cdek.test/v2/orders/cdek-order-uuid"), + ) + http_client = SequenceHTTPClient([response]) + client = _make_client(http_client) + + result = asyncio.run(client.get_order("cdek-order-uuid")) + + assert result == CDEKOrderInfo( + order_uuid="cdek-order-uuid", + status_code="ACCEPTED", + waybill_uuid="waybill-uuid-1", + ) + assert http_client.calls[0]["url"] == ( + "https://api.cdek.test/v2/orders/cdek-order-uuid" + ) + assert http_client.calls[0]["headers"] == {"Authorization": "Bearer test-token"} + + +def test_get_order_retries_on_5xx_and_succeeds() -> None: + flaky = httpx.Response( + 503, + json={"message": "temporary"}, + request=httpx.Request("GET", "https://api.cdek.test/v2/orders/x"), + ) + success = httpx.Response( + 200, + json={"entity": {"uuid": "x", "statuses": []}}, + request=httpx.Request("GET", "https://api.cdek.test/v2/orders/x"), + ) + http_client = SequenceHTTPClient([flaky, success]) + sleep_calls: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleep_calls.append(seconds) + + client = _make_client( + http_client, + retry_attempts=1, + retry_backoff_seconds=0.1, + sleep=fake_sleep, + ) + + result = asyncio.run(client.get_order("x")) + + assert result.order_uuid == "x" + assert len(http_client.calls) == 2 + assert sleep_calls == [0.1] + + +def test_get_order_raises_request_error_on_4xx() -> None: + response = httpx.Response( + 404, + json={"requests": [{"errors": [{"code": "v2_not_found"}]}]}, + request=httpx.Request("GET", "https://api.cdek.test/v2/orders/missing"), + ) + http_client = SequenceHTTPClient([response]) + client = _make_client(http_client) + + with pytest.raises(CDEKRequestError, match="status 404"): + asyncio.run(client.get_order("missing")) + + +def test_get_waybill_returns_url_when_present() -> None: + response = httpx.Response( + 200, + json={ + "entity": { + "uuid": "waybill-uuid-1", + "url": "https://cdek.test/waybill/1.pdf", + } + }, + request=httpx.Request( + "GET", "https://api.cdek.test/v2/print/orders/waybill-uuid-1" + ), + ) + http_client = SequenceHTTPClient([response]) + client = _make_client(http_client) + + result = asyncio.run(client.get_waybill("waybill-uuid-1")) + + assert result == CDEKWaybillInfo( + waybill_uuid="waybill-uuid-1", + url="https://cdek.test/waybill/1.pdf", + ) + assert http_client.calls[0]["url"] == ( + "https://api.cdek.test/v2/print/orders/waybill-uuid-1" + ) + + +def test_get_waybill_raises_client_error_for_invalid_payload() -> None: + response = httpx.Response( + 200, + json={"entity": {}}, + request=httpx.Request( + "GET", "https://api.cdek.test/v2/print/orders/waybill-uuid-1" + ), + ) + http_client = SequenceHTTPClient([response]) + client = _make_client(http_client) + + with pytest.raises(CDEKClientError, match="response payload is invalid"): + asyncio.run(client.get_waybill("waybill-uuid-1")) diff --git a/tests/adapters/delivery_providers/cdek/test_order_mapper.py b/tests/adapters/delivery_providers/cdek/test_order_mapper.py index e91b7c6..a9fc032 100644 --- a/tests/adapters/delivery_providers/cdek/test_order_mapper.py +++ b/tests/adapters/delivery_providers/cdek/test_order_mapper.py @@ -1,10 +1,16 @@ from app.adapters.delivery_providers.cdek.order_mapper import ( + CDEKOrderInfo, + CDEKOrderMappingError, CDEKOrderRegistrationResult, + CDEKWaybillInfo, compose_address_line, map_cdek_existing_order_response, + map_cdek_order_info_response, map_cdek_order_request, map_cdek_order_response, + map_cdek_waybill_info_response, ) +import pytest from app.schemas.payment import Address from tests.payment_fixtures import make_init_payment_request @@ -207,3 +213,72 @@ def test_compose_address_line_handles_empty_apartment() -> None: ) assert compose_address_line(address) == "Москва, Lenina, 1" + + +def test_map_cdek_order_info_response_picks_latest_status_by_date_time() -> None: + result = map_cdek_order_info_response( + { + "entity": { + "uuid": "cdek-order-uuid", + "statuses": [ + {"code": "ACCEPTED", "date_time": "2026-05-24T10:00:00+0000"}, + {"code": "INVALID", "date_time": "2026-05-24T10:00:05+0000"}, + ], + }, + "related_entities": [ + {"type": "waybill", "uuid": "waybill-uuid-1"} + ], + } + ) + + assert result == CDEKOrderInfo( + order_uuid="cdek-order-uuid", + status_code="INVALID", + waybill_uuid="waybill-uuid-1", + ) + + +def test_map_cdek_order_info_response_returns_none_status_when_statuses_empty() -> None: + result = map_cdek_order_info_response( + {"entity": {"uuid": "cdek-order-uuid", "statuses": []}} + ) + + assert result == CDEKOrderInfo( + order_uuid="cdek-order-uuid", + status_code=None, + waybill_uuid=None, + ) + + +def test_map_cdek_order_info_response_raises_for_missing_entity() -> None: + with pytest.raises(CDEKOrderMappingError): + map_cdek_order_info_response({"requests": []}) + + +def test_map_cdek_waybill_info_response_returns_url_when_present() -> None: + result = map_cdek_waybill_info_response( + { + "entity": { + "uuid": "waybill-uuid-1", + "url": "https://cdek.test/waybill/1.pdf", + } + } + ) + + assert result == CDEKWaybillInfo( + waybill_uuid="waybill-uuid-1", + url="https://cdek.test/waybill/1.pdf", + ) + + +def test_map_cdek_waybill_info_response_returns_none_url_when_missing() -> None: + result = map_cdek_waybill_info_response( + {"entity": {"uuid": "waybill-uuid-1"}} + ) + + assert result == CDEKWaybillInfo(waybill_uuid="waybill-uuid-1", url=None) + + +def test_map_cdek_waybill_info_response_raises_for_missing_uuid() -> None: + with pytest.raises(CDEKOrderMappingError): + map_cdek_waybill_info_response({"entity": {}}) diff --git a/tests/domain/test_cdek_polling.py b/tests/domain/test_cdek_polling.py new file mode 100644 index 0000000..d4f94d0 --- /dev/null +++ b/tests/domain/test_cdek_polling.py @@ -0,0 +1,19 @@ +import pytest + +from app.domain.cdek_polling import TERMINAL_ORDER_STATUSES, is_terminal_order_status + + +@pytest.mark.parametrize( + "code", sorted(TERMINAL_ORDER_STATUSES) +) +def test_terminal_statuses_are_terminal(code: str) -> None: + assert is_terminal_order_status(code) is True + + +@pytest.mark.parametrize("code", ["ACCEPTED", "CREATED", "RECEIVED_AT_SHIPMENT_WAREHOUSE", ""]) +def test_non_terminal_statuses_are_not_terminal(code: str) -> None: + assert is_terminal_order_status(code) is False + + +def test_none_status_is_not_terminal() -> None: + assert is_terminal_order_status(None) is False diff --git a/tests/repositories/order/test_repository.py b/tests/repositories/order/test_repository.py index f5b6f8f..e34676f 100644 --- a/tests/repositories/order/test_repository.py +++ b/tests/repositories/order/test_repository.py @@ -1,5 +1,6 @@ import asyncio from collections.abc import Awaitable, Callable +from datetime import datetime, timezone from typing import Any import pytest @@ -173,7 +174,7 @@ def test_mark_payment_status_returns_none_for_missing_order() -> None: asyncio.run(_with_repository(run)) -def test_mark_cdek_order_registered_persists_cdek_order_and_waybill() -> None: +def test_mark_cdek_order_registered_persists_cdek_order_uuid_only() -> None: async def run( repository: OrderRepository, session_factory: async_sessionmaker[AsyncSession], @@ -186,8 +187,6 @@ def test_mark_cdek_order_registered_persists_cdek_order_and_waybill() -> None: session, "order-uuid-1", "cdek-order-uuid-1", - "waybill-uuid-1", - "https://cdek.test/waybill/1.pdf", ) async with session_factory() as session: @@ -197,34 +196,6 @@ def test_mark_cdek_order_registered_persists_cdek_order_and_waybill() -> None: persisted_order = result.scalar_one() assert order is not None - assert persisted_order.cdek_order_uuid == "cdek-order-uuid-1" - assert persisted_order.cdek_waybill_uuid == "waybill-uuid-1" - assert persisted_order.cdek_waybill_url == "https://cdek.test/waybill/1.pdf" - - asyncio.run(_with_repository(run)) - - -def test_mark_cdek_order_registered_allows_missing_waybill_fields() -> None: - async def run( - repository: OrderRepository, - session_factory: async_sessionmaker[AsyncSession], - ) -> None: - async with repository.session() as session: - await repository.create_order(session, _make_order_data()) - - async with repository.session() as session: - await repository.mark_cdek_order_registered( - session, - "order-uuid-1", - "cdek-order-uuid-1", - ) - - async with session_factory() as session: - result = await session.execute( - select(Order).where(Order.order_uuid == "order-uuid-1") - ) - persisted_order = result.scalar_one() - assert persisted_order.cdek_order_uuid == "cdek-order-uuid-1" assert persisted_order.cdek_waybill_uuid is None assert persisted_order.cdek_waybill_url is None @@ -279,3 +250,222 @@ def test_mark_cdek_order_registered_returns_none_for_missing_order() -> None: assert order is None asyncio.run(_with_repository(run)) + + +async def _seed_order( + repository: OrderRepository, + *, + order_uuid: str, + cdek_order_uuid: str | None, + cdek_order_status: str | None = None, + cdek_waybill_uuid: str | None = None, + cdek_waybill_url: str | None = None, + cdek_polled_at: datetime | None = None, +) -> None: + async with repository.session() as session: + await repository.create_order( + session, _make_order_data(order_uuid=order_uuid) + ) + if cdek_order_uuid is not None: + await repository.mark_cdek_order_registered( + session, order_uuid, cdek_order_uuid + ) + if ( + cdek_order_status is not None + or cdek_waybill_uuid is not None + or cdek_polled_at is not None + ): + order = await repository.get_order_by_order_uuid(session, order_uuid) + assert order is not None + if cdek_order_status is not None: + order.cdek_order_status = cdek_order_status + if cdek_waybill_uuid is not None: + order.cdek_waybill_uuid = cdek_waybill_uuid + if cdek_polled_at is not None: + order.cdek_polled_at = cdek_polled_at + if cdek_waybill_url is not None: + order = await repository.get_order_by_order_uuid(session, order_uuid) + assert order is not None + order.cdek_waybill_url = cdek_waybill_url + + +def test_list_orders_pending_waybill_returns_orders_without_url() -> None: + async def run( + repository: OrderRepository, + _session_factory: async_sessionmaker[AsyncSession], + ) -> None: + await _seed_order(repository, order_uuid="pending", cdek_order_uuid="o1") + await _seed_order( + repository, + order_uuid="done", + cdek_order_uuid="o2", + cdek_waybill_uuid="w2", + cdek_waybill_url="https://cdek.test/2.pdf", + ) + await _seed_order( + repository, + order_uuid="invalid", + cdek_order_uuid="o3", + cdek_order_status="INVALID", + ) + await _seed_order(repository, order_uuid="no-cdek", cdek_order_uuid=None) + + async with repository.session() as session: + orders = await repository.list_orders_pending_waybill(session, limit=10) + + assert [order.order_uuid for order in orders] == ["pending"] + + asyncio.run(_with_repository(run)) + + +def test_list_orders_pending_waybill_orders_polled_at_nulls_first() -> None: + earlier = datetime(2026, 5, 24, 10, 0, tzinfo=timezone.utc) + later = datetime(2026, 5, 24, 11, 0, tzinfo=timezone.utc) + + async def run( + repository: OrderRepository, + _session_factory: async_sessionmaker[AsyncSession], + ) -> None: + await _seed_order( + repository, + order_uuid="late", + cdek_order_uuid="o-late", + cdek_polled_at=later, + ) + await _seed_order( + repository, + order_uuid="early", + cdek_order_uuid="o-early", + cdek_polled_at=earlier, + ) + await _seed_order(repository, order_uuid="never", cdek_order_uuid="o-never") + + async with repository.session() as session: + orders = await repository.list_orders_pending_waybill(session, limit=10) + + assert [order.order_uuid for order in orders] == ["never", "early", "late"] + + asyncio.run(_with_repository(run)) + + +def test_record_order_poll_sets_status_and_waybill_uuid() -> None: + polled = datetime(2026, 5, 24, 12, 0, tzinfo=timezone.utc) + + async def run( + repository: OrderRepository, + _session_factory: async_sessionmaker[AsyncSession], + ) -> None: + await _seed_order(repository, order_uuid="o", cdek_order_uuid="cdek-o") + + async with repository.session() as session: + order = await repository.record_order_poll( + session, + order_uuid="o", + order_status="ACCEPTED", + waybill_uuid="waybill-1", + polled_at=polled, + ) + + assert order is not None + assert order.cdek_order_status == "ACCEPTED" + assert order.cdek_waybill_uuid == "waybill-1" + assert order.cdek_polled_at == polled + + asyncio.run(_with_repository(run)) + + +def test_record_order_poll_does_not_overwrite_existing_waybill_uuid() -> None: + polled = datetime(2026, 5, 24, 12, 0, tzinfo=timezone.utc) + + async def run( + repository: OrderRepository, + _session_factory: async_sessionmaker[AsyncSession], + ) -> None: + await _seed_order( + repository, + order_uuid="o", + cdek_order_uuid="cdek-o", + cdek_waybill_uuid="existing-waybill", + ) + + async with repository.session() as session: + order = await repository.record_order_poll( + session, + order_uuid="o", + order_status="ACCEPTED", + waybill_uuid="new-waybill", + polled_at=polled, + ) + + assert order is not None + assert order.cdek_waybill_uuid == "existing-waybill" + + asyncio.run(_with_repository(run)) + + +def test_record_waybill_poll_sets_url_only_when_previously_null() -> None: + polled = datetime(2026, 5, 24, 13, 0, tzinfo=timezone.utc) + + async def run( + repository: OrderRepository, + _session_factory: async_sessionmaker[AsyncSession], + ) -> None: + await _seed_order( + repository, + order_uuid="o", + cdek_order_uuid="cdek-o", + cdek_waybill_uuid="waybill-1", + ) + + async with repository.session() as session: + order = await repository.record_waybill_poll( + session, + order_uuid="o", + waybill_url="https://cdek.test/1.pdf", + polled_at=polled, + ) + + assert order is not None + assert order.cdek_waybill_url == "https://cdek.test/1.pdf" + assert order.cdek_polled_at == polled + + async with repository.session() as session: + order = await repository.record_waybill_poll( + session, + order_uuid="o", + waybill_url="https://cdek.test/REPLACED.pdf", + polled_at=polled, + ) + assert order is not None + assert order.cdek_waybill_url == "https://cdek.test/1.pdf" + + asyncio.run(_with_repository(run)) + + +def test_record_waybill_poll_updates_polled_at_when_url_is_none() -> None: + polled = datetime(2026, 5, 24, 13, 0, tzinfo=timezone.utc) + + async def run( + repository: OrderRepository, + _session_factory: async_sessionmaker[AsyncSession], + ) -> None: + await _seed_order( + repository, + order_uuid="o", + cdek_order_uuid="cdek-o", + cdek_waybill_uuid="waybill-1", + ) + + async with repository.session() as session: + order = await repository.record_waybill_poll( + session, + order_uuid="o", + waybill_url=None, + polled_at=polled, + ) + + assert order is not None + assert order.cdek_waybill_url is None + assert order.cdek_polled_at == polled + + asyncio.run(_with_repository(run)) diff --git a/tests/services/test_tbank_notifications.py b/tests/services/test_tbank_notifications.py index 8bc5a92..ab62c09 100644 --- a/tests/services/test_tbank_notifications.py +++ b/tests/services/test_tbank_notifications.py @@ -113,20 +113,9 @@ class StubOrderRepository: session: object, order_uuid: str, cdek_order_uuid: str, - cdek_waybill_uuid: str | None = None, - cdek_waybill_url: str | None = None, ) -> StoredOrder | None: self.calls.append( - ( - "mark_cdek_order_registered", - ( - session, - order_uuid, - cdek_order_uuid, - cdek_waybill_uuid, - cdek_waybill_url, - ), - ) + ("mark_cdek_order_registered", (session, order_uuid, cdek_order_uuid)) ) if self._mark_cdek_errors: error = self._mark_cdek_errors.pop(0) @@ -137,8 +126,6 @@ class StubOrderRepository: if order is None: return None order.cdek_order_uuid = cdek_order_uuid - order.cdek_waybill_uuid = cdek_waybill_uuid - order.cdek_waybill_url = cdek_waybill_url return order @@ -211,8 +198,8 @@ def test_confirmed_notification_registers_cdek_order_and_saves_uuid() -> None: assert order.payment_status == "CONFIRMED" assert order.tbank_payment_id == 8347568144 assert order.cdek_order_uuid == "cdek-order-uuid-1" - assert order.cdek_waybill_uuid == "waybill-uuid-1" - assert order.cdek_waybill_url == "https://cdek.test/waybill/1.pdf" + assert order.cdek_waybill_uuid is None + assert order.cdek_waybill_url is None assert len(cdek_adapter.calls) == 1 assert cdek_adapter.calls[0].order_uuid == "order-uuid-1" diff --git a/tests/services/test_waybill_poller.py b/tests/services/test_waybill_poller.py new file mode 100644 index 0000000..14fdcc8 --- /dev/null +++ b/tests/services/test_waybill_poller.py @@ -0,0 +1,259 @@ +import asyncio +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from app.adapters.delivery_providers.cdek.order_mapper import ( + CDEKOrderInfo, + CDEKWaybillInfo, +) +from app.services.waybill_poller import WaybillPollerService + + +@dataclass +class StoredOrder: + order_uuid: str + cdek_order_uuid: str | None = None + cdek_order_status: str | None = None + cdek_waybill_uuid: str | None = None + cdek_waybill_url: str | None = None + cdek_polled_at: datetime | None = None + + +class StubSessionContext: + def __init__(self, session: object) -> None: + self._session = session + + async def __aenter__(self) -> object: + return self._session + + async def __aexit__(self, exc_type, exc, traceback) -> None: + return None + + +class StubRepository: + def __init__(self, orders: list[StoredOrder]) -> None: + self._orders = {order.order_uuid: order for order in orders} + self.session_value = object() + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def session(self) -> StubSessionContext: + return StubSessionContext(self.session_value) + + async def list_orders_pending_waybill( + self, session: object, *, limit: int + ) -> list[StoredOrder]: + self.calls.append(("list", {"session": session, "limit": limit})) + return list(self._orders.values()) + + async def record_order_poll( + self, + session: object, + *, + order_uuid: str, + order_status: str | None, + waybill_uuid: str | None, + polled_at: datetime, + ) -> StoredOrder | None: + self.calls.append( + ( + "record_order_poll", + { + "order_uuid": order_uuid, + "order_status": order_status, + "waybill_uuid": waybill_uuid, + "polled_at": polled_at, + }, + ) + ) + order = self._orders.get(order_uuid) + if order is None: + return None + order.cdek_order_status = order_status + if waybill_uuid is not None and order.cdek_waybill_uuid is None: + order.cdek_waybill_uuid = waybill_uuid + order.cdek_polled_at = polled_at + return order + + async def record_waybill_poll( + self, + session: object, + *, + order_uuid: str, + waybill_url: str | None, + polled_at: datetime, + ) -> StoredOrder | None: + self.calls.append( + ( + "record_waybill_poll", + { + "order_uuid": order_uuid, + "waybill_url": waybill_url, + "polled_at": polled_at, + }, + ) + ) + order = self._orders.get(order_uuid) + if order is None: + return None + if waybill_url is not None and order.cdek_waybill_url is None: + order.cdek_waybill_url = waybill_url + order.cdek_polled_at = polled_at + return order + + +class StubOrderInfoAdapter: + def __init__(self, results: dict[str, CDEKOrderInfo | Exception]) -> None: + self._results = results + self.calls: list[str] = [] + + async def get_order(self, cdek_order_uuid: str) -> CDEKOrderInfo: + self.calls.append(cdek_order_uuid) + result = self._results[cdek_order_uuid] + if isinstance(result, Exception): + raise result + return result + + +class StubWaybillInfoAdapter: + def __init__(self, results: dict[str, CDEKWaybillInfo | Exception]) -> None: + self._results = results + self.calls: list[str] = [] + + async def get_waybill(self, cdek_waybill_uuid: str) -> CDEKWaybillInfo: + self.calls.append(cdek_waybill_uuid) + result = self._results[cdek_waybill_uuid] + if isinstance(result, Exception): + raise result + return result + + +_POLLED_AT = datetime(2026, 5, 24, 12, 0, tzinfo=timezone.utc) + + +def _make_service( + *, + repository: StubRepository, + order_info: StubOrderInfoAdapter | None = None, + waybill_info: StubWaybillInfoAdapter | None = None, +) -> WaybillPollerService: + return WaybillPollerService( + order_repository=repository, + order_info_adapter=order_info or StubOrderInfoAdapter({}), + waybill_info_adapter=waybill_info or StubWaybillInfoAdapter({}), + batch_size=10, + datetime_now=lambda: _POLLED_AT, + ) + + +def test_poll_once_fetches_order_info_when_waybill_uuid_is_missing() -> None: + order = StoredOrder(order_uuid="o", cdek_order_uuid="cdek-o") + repo = StubRepository([order]) + order_info = StubOrderInfoAdapter( + { + "cdek-o": CDEKOrderInfo( + order_uuid="cdek-o", + status_code="ACCEPTED", + waybill_uuid="waybill-1", + ) + } + ) + service = _make_service(repository=repo, order_info=order_info) + + summary = asyncio.run(service.poll_once()) + + assert summary.processed == 1 and summary.succeeded == 1 and summary.failed == 0 + assert order_info.calls == ["cdek-o"] + assert order.cdek_order_status == "ACCEPTED" + assert order.cdek_waybill_uuid == "waybill-1" + assert order.cdek_polled_at == _POLLED_AT + + +def test_poll_once_fetches_waybill_info_when_waybill_uuid_is_present() -> None: + order = StoredOrder( + order_uuid="o", + cdek_order_uuid="cdek-o", + cdek_waybill_uuid="waybill-1", + ) + repo = StubRepository([order]) + waybill_info = StubWaybillInfoAdapter( + { + "waybill-1": CDEKWaybillInfo( + waybill_uuid="waybill-1", + url="https://cdek.test/1.pdf", + ) + } + ) + service = _make_service(repository=repo, waybill_info=waybill_info) + + summary = asyncio.run(service.poll_once()) + + assert summary.processed == 1 and summary.succeeded == 1 and summary.failed == 0 + assert waybill_info.calls == ["waybill-1"] + assert order.cdek_waybill_url == "https://cdek.test/1.pdf" + assert order.cdek_polled_at == _POLLED_AT + + +def test_poll_once_records_terminal_status_without_waybill() -> None: + order = StoredOrder(order_uuid="o", cdek_order_uuid="cdek-o") + repo = StubRepository([order]) + order_info = StubOrderInfoAdapter( + { + "cdek-o": CDEKOrderInfo( + order_uuid="cdek-o", + status_code="INVALID", + waybill_uuid=None, + ) + } + ) + service = _make_service(repository=repo, order_info=order_info) + + summary = asyncio.run(service.poll_once()) + + assert summary.succeeded == 1 + assert order.cdek_order_status == "INVALID" + assert order.cdek_waybill_uuid is None + + +def test_poll_once_failure_on_one_order_does_not_break_batch() -> None: + failing = StoredOrder(order_uuid="bad", cdek_order_uuid="cdek-bad") + good = StoredOrder( + order_uuid="good", + cdek_order_uuid="cdek-good", + cdek_waybill_uuid="waybill-good", + ) + repo = StubRepository([failing, good]) + order_info = StubOrderInfoAdapter({"cdek-bad": RuntimeError("cdek down")}) + waybill_info = StubWaybillInfoAdapter( + { + "waybill-good": CDEKWaybillInfo( + waybill_uuid="waybill-good", + url="https://cdek.test/good.pdf", + ) + } + ) + service = _make_service( + repository=repo, order_info=order_info, waybill_info=waybill_info + ) + + summary = asyncio.run(service.poll_once()) + + assert summary.processed == 2 + assert summary.succeeded == 1 + assert summary.failed == 1 + assert good.cdek_waybill_url == "https://cdek.test/good.pdf" + + +def test_run_forever_exits_when_stop_event_is_set() -> None: + repo = StubRepository([]) + service = _make_service(repository=repo) + + async def run() -> None: + stop_event = asyncio.Event() + stop_event.set() + await asyncio.wait_for( + service.run_forever(interval_seconds=0.01, stop_event=stop_event), + timeout=1.0, + ) + + asyncio.run(run()) diff --git a/tests/workers/__init__.py b/tests/workers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/workers/test_waybill_poller_main.py b/tests/workers/test_waybill_poller_main.py new file mode 100644 index 0000000..c4ce015 --- /dev/null +++ b/tests/workers/test_waybill_poller_main.py @@ -0,0 +1,88 @@ +"""Smoke test for the waybill poller worker entrypoint. + +The test patches all external dependencies (HTTP client, DB engine, CDEK, +service) and verifies that `_run` wires the components together and exits +cleanly when the stop event is set. +""" + +import asyncio +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from app.config import ( + AdapterConfig, + ObservabilityConfig, + PostgresConfig, + Settings, + TBankPaymentAuthConfig, + TBankPaymentConfig, + WaybillPollerConfig, +) +from app.workers.waybill_poller import _run + + +def _make_settings() -> Settings: + return Settings( + adapter=AdapterConfig( + cdek_base_url="https://api.cdek.test/v2", + cdek_client_id="id", + cdek_client_secret="secret", + ), + tbank_payment=TBankPaymentConfig( + init_url="https://pay.test/init", + notification_url="https://pay.test/notify", + success_url="https://pay.test/success", + auth=TBankPaymentAuthConfig(terminal_key="t", password="p"), + ), + postgres=PostgresConfig(dsn="postgresql+asyncpg://u:p@h/db"), + observability=ObservabilityConfig( + service_name="svc", otlp_endpoint="http://otlp" + ), + waybill_poller=WaybillPollerConfig(interval_seconds=0.01, batch_size=5), + business_logic={"provider_price_multiplier": "1.0"}, + ) + + +def test_run_exits_when_stop_event_is_set() -> None: + settings = _make_settings() + + http_client_instance = MagicMock() + http_client_instance.aclose = AsyncMock() + engine_instance = MagicMock() + engine_instance.dispose = AsyncMock() + + run_forever_mock = AsyncMock() + + with ( + patch( + "app.workers.waybill_poller.httpx.AsyncClient", + return_value=http_client_instance, + ), + patch("app.workers.waybill_poller.CDEKAuthClient"), + patch("app.workers.waybill_poller.CDEKClient"), + patch( + "app.workers.waybill_poller.create_postgres_engine", + return_value=engine_instance, + ), + patch("app.workers.waybill_poller.create_postgres_session_factory"), + patch("app.workers.waybill_poller.OrderRepository"), + patch( + "app.workers.waybill_poller.WaybillPollerService" + ) as service_cls_mock, + ): + service_instance = MagicMock() + service_instance.run_forever = run_forever_mock + service_cls_mock.return_value = service_instance + + async def runner() -> None: + stop_event = asyncio.Event() + stop_event.set() + await asyncio.wait_for(_run(settings, stop_event), timeout=1.0) + + asyncio.run(runner()) + + assert run_forever_mock.await_count == 1 + call_kwargs: dict[str, Any] = run_forever_mock.await_args.kwargs + assert call_kwargs["interval_seconds"] == 0.01 + assert http_client_instance.aclose.await_count == 1 + assert engine_instance.dispose.await_count == 1