import asyncio from decimal import Decimal from typing import Any import httpx import pytest from app.adapters.delivery_providers.cse import ( CSEClient, 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 """ _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_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), ] ) provider = CSEProvider(client) prices = asyncio.run(provider.get_prices(_calc_request())) # PVZ-requiring scheme (Склад-Дверь) is excluded: only delivery types + one # Calc (for door-to-door) are requested. assert len(http_client.calls) == 2 assert [price.provider for price in prices] == ["cse", "cse"] assert [price.tariff_code for price in prices] == [ "ДоставкаДоДверей|urg-std", "ДоставкаДоДверей|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, _ = _build_client([response]) provider = CSEProvider(client) request = make_init_payment_request( systemData={ "tariff": { "provider": "cse", "serviceName": "Экспресс", "price": 200000, "deliveryDaysMin": 1, "deliveryDaysMax": 2, "tariffCode": "ДоставкаДоДверей|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 == "ДоставкаДоДверей|urg-exp" assert price.price == Decimal("2000.00") 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(), "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_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": "ДоставкаДоДверей|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" 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) with pytest.raises(CSERequestError): asyncio.run(provider.get_prices(_calc_request()))