import asyncio from decimal import Decimal from typing import Any import httpx import pytest from app.adapters.delivery_providers.cse import ( CSEClient, CSEClientError, CSEProvider, CSERequestError, ) from app.adapters.delivery_providers.cse import order_mapper from app.adapters.delivery_providers.cse.order_mapper import ( CSEOrderRegistrationParams, ) from app.adapters.delivery_providers.cse.soap import ( build_envelope, parse_response, ) from app.schemas.request import DeliveryCalculationRequest, DeliveryEntity from tests.payment_fixtures import make_init_payment_request _CALC_RESPONSE = """ Calc Destination Tariff tariff-guid-1 Total1114.92 CurrencyNameRUR ServiceРоссия доставка Urgencyurg-std MinPeriod3 MaxPeriod5 Tariff tariff-guid-2 Total2000.00 CurrencyNameRUR ServiceЭкспресс Urgencyurg-exp MinPeriod1 MaxPeriod2 """ _SAVE_RESPONSE = """ SaveDocuments Order NumberCSE-000123 """ _TRACKING_RESPONSE = """ Tracking CSE-000123 Order NumberCSE-000123 Заказ принят, идет обработка заказа. DateTime2026-06-01T10:00:00 Накладная оформлена. DateTime2026-06-01T10:02:00 Waybills 496-AA-1676378 DocumentTypeWaybill Number496-AA-1676378 """ _FORM_RESPONSE = """ GetPrintForms 496-AA-1676378 FormFormatPDF JVBERg== """ _CALC_RESPONSE_WITH_ADDITIONAL_SERVICE_FIRST = """ Calc Destination Tariff additional-service-guid Total100.00 CurrencyNameRUR ServiceДоп. услуга Urgencyurg-exp AdditionalServicetrue Tariff delivery-guid Total793.00 CurrencyNameRUR ServiceЭкспресс Urgencyurg-exp AdditionalServicefalse MinPeriod1 MaxPeriod2 """ _CALC_RESPONSE_WITH_BUYOUT_SERVICE_FIRST = """ Calc Destination Tariff buyout-service-guid Total100.00 CurrencyNameRUR ServiceЧастичный выкуп Urgencyurg-exp Tariff delivery-guid Total793.00 CurrencyNameRUR ServiceЭкспресс Urgencyurg-exp AdditionalServicefalse MinPeriod1 MaxPeriod2 """ _ERROR_RESPONSE = """ Calc Error Description1001 """ _DELIVERY_TYPES_RESPONSE = """ deliverytype NameOfdeliverytype ДоставкаДоДверей InformationДверь-Дверь NameOfdeliverytype СкладДверь InformationСклад-Дверь """ class SequenceHTTPClient: def __init__(self, results: list[Any]) -> None: self._results = results self.calls: list[dict[str, Any]] = [] async def post( self, url: str, *, content: bytes | None = None, headers: dict[str, str] | None = None, timeout: float | None = None, ) -> httpx.Response: self.calls.append({"url": url, "content": content, "headers": headers}) result = self._results[len(self.calls) - 1] if isinstance(result, Exception): raise result return result def _params() -> CSEOrderRegistrationParams: return CSEOrderRegistrationParams( payer="payer-1", payment_method="pm-1", shipping_method="sm-1", ) def _build_client(results: list[Any]) -> tuple[CSEClient, SequenceHTTPClient]: http_client = SequenceHTTPClient(results) client = CSEClient( http_client=http_client, # type: ignore[arg-type] base_url="http://lk-test.cse.ru/1c/ws/web1c.1cws", login="test", password="2016", registration_params=_params(), retry_attempts=0, ) return client, http_client @pytest.fixture(autouse=True) def _patch_references(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( order_mapper, "resolve_cse_geography", lambda city_id: f"geo-{city_id}" ) monkeypatch.setattr( order_mapper, "resolve_type_of_cargo", lambda parcel_type: "cargo-guid" ) def _calc_request() -> DeliveryCalculationRequest: return DeliveryCalculationRequest( entity=DeliveryEntity.INDIVIDUAL, from_city=1, to_city=2, weight_kg=1.0, length_cm=10.0, width_cm=10.0, height_cm=10.0, ) def test_build_envelope_contains_credentials_and_payload() -> None: body = order_mapper.build_calc_body_for_calculation(_calc_request()) envelope = build_envelope("Calc", login="test", password="2016", body=body) text = envelope.decode("utf-8") assert "Calc" in text assert "test" in text or ">test<" in text assert "geo-1" in text assert "geo-2" in text def test_save_order_envelope_serializes_properties_before_fields() -> None: body = order_mapper.map_cse_save_order_request( make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 200000, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|tariff-guid-2|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": {"length": "10", "width": "10", "height": "10"}, } ), "order-uuid-1", _params(), ) text = build_envelope( "SaveDocuments", login="test", password="2016", body=body ).decode("utf-8") order_start = text.index("Order") properties_index = text.index("", order_start) fields_index = text.index("", order_start) assert properties_index < fields_index def test_parse_calc_response_round_trip() -> None: root = parse_response(_CALC_RESPONSE, "Calc") assert root.key == "Calc" assert root.items[0].items[0].field_value("Total") == "1114.92" def test_provider_get_prices_maps_all_tariffs() -> None: client, http_client = _build_client( [ httpx.Response(200, text=_DELIVERY_TYPES_RESPONSE), httpx.Response(200, text=_CALC_RESPONSE), httpx.Response(200, text=_CALC_RESPONSE), ] ) provider = CSEProvider( client, delivery_service_guids=("tariff-guid-1", "tariff-guid-2"), ) prices = asyncio.run(provider.get_prices(_calc_request())) # PVZ-requiring scheme (Склад-Дверь) is excluded: only delivery types + one # Calc per configured CSE delivery service are requested. assert len(http_client.calls) == 3 assert [price.provider for price in prices] == ["cse", "cse"] assert [price.tariff_code for price in prices] == [ "ДоставкаДоДверей|tariff-guid-1|urg-std", "ДоставкаДоДверей|tariff-guid-2|urg-exp", ] assert prices[0].price == Decimal("1114.92") assert prices[0].currency == "RUB" assert prices[0].delivery_days_min == 3 assert prices[0].delivery_days_max == 5 assert prices[0].service_name == "Россия доставка — Дверь-Дверь" def test_provider_get_payment_price_selects_matching_tariff() -> None: response = httpx.Response(200, text=_CALC_RESPONSE) client, http_client = _build_client([response]) provider = CSEProvider(client) request = make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 200000, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|tariff-guid-2|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": {"length": "10", "width": "10", "height": "10"}, } ) price = asyncio.run(provider.get_payment_price(request)) assert price is not None assert price.tariff_code == "ДоставкаДоДверей|tariff-guid-2|urg-exp" assert price.price == Decimal("2000.00") content = http_client.calls[0]["content"].decode("utf-8") assert "Service" in content assert "tariff-guid-2" in content def test_provider_get_payment_price_ignores_matching_additional_service() -> None: response = httpx.Response(200, text=_CALC_RESPONSE_WITH_ADDITIONAL_SERVICE_FIRST) client, _ = _build_client([response]) provider = CSEProvider(client) request = make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 79300, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|delivery-guid|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": {"length": "10", "width": "10", "height": "10"}, } ) price = asyncio.run(provider.get_payment_price(request)) assert price is not None assert price.tariff_code == "ДоставкаДоДверей|delivery-guid|urg-exp" assert price.price == Decimal("793.00") def test_provider_get_payment_price_selects_matching_service_guid() -> None: response = httpx.Response(200, text=_CALC_RESPONSE_WITH_BUYOUT_SERVICE_FIRST) client, _ = _build_client([response]) provider = CSEProvider(client) request = make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 79300, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|delivery-guid|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": {"length": "10", "width": "10", "height": "10"}, } ) price = asyncio.run(provider.get_payment_price(request)) assert price is not None assert price.tariff_code == "ДоставкаДоДверей|delivery-guid|urg-exp" assert price.price == Decimal("793.00") def test_provider_get_prices_excludes_additional_service_tariffs() -> None: client, _ = _build_client( [ httpx.Response(200, text=_DELIVERY_TYPES_RESPONSE), httpx.Response(200, text=_CALC_RESPONSE_WITH_ADDITIONAL_SERVICE_FIRST), ] ) provider = CSEProvider(client, delivery_service_guids=("delivery-guid",)) prices = asyncio.run(provider.get_prices(_calc_request())) assert [price.price for price in prices] == [Decimal("793.00")] assert [price.service_name for price in prices] == [ "Экспресс — Дверь-Дверь" ] def test_provider_get_prices_uses_configured_service_guid_filter() -> None: client, http_client = _build_client( [ httpx.Response(200, text=_DELIVERY_TYPES_RESPONSE), httpx.Response(200, text=_CALC_RESPONSE_WITH_BUYOUT_SERVICE_FIRST), ] ) provider = CSEProvider(client, delivery_service_guids=("delivery-guid",)) prices = asyncio.run(provider.get_prices(_calc_request())) assert [price.price for price in prices] == [Decimal("793.00")] assert [price.service_name for price in prices] == [ "Экспресс — Дверь-Дверь" ] content = http_client.calls[1]["content"].decode("utf-8") assert "Service" in content assert "delivery-guid" in content def test_provider_get_prices_requires_delivery_service_guid_configuration() -> None: client, _ = _build_client([httpx.Response(200, text=_DELIVERY_TYPES_RESPONSE)]) provider = CSEProvider(client) with pytest.raises(CSERequestError): asyncio.run(provider.get_prices(_calc_request())) def test_provider_register_order_returns_document_number() -> None: response = httpx.Response(200, text=_SAVE_RESPONSE) client, http_client = _build_client([response]) provider = CSEProvider(client) result = asyncio.run( provider.register_order( make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 200000, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|tariff-guid-2|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": {"length": "10", "width": "10", "height": "10"}, } ), "order-uuid-1", ) ) assert result.order_number == "CSE-000123" assert http_client.calls[0]["url"] == "http://lk-test.cse.ru/1c/ws/web1c.1cws" def test_provider_get_order_extracts_waybill_number_from_tracking() -> None: response = httpx.Response(200, text=_TRACKING_RESPONSE) client, http_client = _build_client([response]) provider = CSEProvider(client) result = asyncio.run(provider.get_order("CSE-000123")) assert result.order_number == "CSE-000123" assert result.status_code == "Накладная оформлена." assert result.waybill_number == "496-AA-1676378" content = http_client.calls[0]["content"].decode("utf-8") assert "Tracking" in content assert "DocumentType" in content assert "Order" in content def test_provider_download_waybill_pdf_uses_print_form() -> None: response = httpx.Response(200, text=_FORM_RESPONSE) client, http_client = _build_client([response]) provider = CSEProvider(client) pdf = asyncio.run(provider.download_waybill_pdf("496-AA-1676378")) assert pdf == b"%PDF" content = http_client.calls[0]["content"].decode("utf-8") assert "GetFormsForDocuments" in content assert "DocumentType" in content assert "waybill" in content assert "Type" in content assert "print" in content assert "Format" in content assert "pdf" in content def test_save_order_request_uses_selected_tariff_urgency() -> None: request = make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 200000, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|tariff-guid-2|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": {"length": "10", "width": "10", "height": "10"}, } ) body = order_mapper.map_cse_save_order_request(request, "order-uuid-1", _params()) order = body["data"].items[0] assert order.field_value("Urgency") == "urg-exp" assert order.field_value("DeliveryOfCargo") == "ДоставкаДоДверей" assert order.field_value("Payer") == "payer-1" assert order.field_value("ShippingMethod") == "sm-1" assert order.field_value("CargoCost") == "50000" def test_application_error_in_response_raises_request_error() -> None: client, _ = _build_client( [ httpx.Response(200, text=_DELIVERY_TYPES_RESPONSE), httpx.Response(200, text=_ERROR_RESPONSE), ] ) provider = CSEProvider(client, delivery_service_guids=("delivery-guid",)) with pytest.raises(CSERequestError): asyncio.run(provider.get_prices(_calc_request())) def test_server_error_logs_response_excerpt(monkeypatch: pytest.MonkeyPatch) -> None: warning_calls: list[dict[str, Any]] = [] def capture_warning(event: str, **kwargs: Any) -> None: warning_calls.append({"event": event, **kwargs}) monkeypatch.setattr( "app.adapters.delivery_providers.cse.client.log.warning", capture_warning, ) response = httpx.Response( 500, text="\n CSE rejected SaveDocuments because field X is invalid \n", ) client, _ = _build_client([response]) provider = CSEProvider(client) with pytest.raises(CSEClientError): asyncio.run( provider.register_order( make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 200000, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|tariff-guid-2|urg-exp", }, "parcelType": "parcel", "weight": "1.0", "dimensions": { "length": "10", "width": "10", "height": "10", }, } ), "order-uuid-1", ) ) assert warning_calls == [ { "event": "cse_request_server_error", "operation": "SaveDocuments", "status_code": 500, "response_excerpt": ( " CSE rejected SaveDocuments because field X is invalid " "" ), } ]