"""Тесты HttpCheckboxClient. Сеть замокана через respx — реальных запросов не делает.""" from __future__ import annotations import asyncio import json import time import uuid import httpx import pytest import respx from httpx import Response from app.core.config import Settings from app.services.checkbox.client import ( CheckboxCredentials, CheckboxError, CheckboxRateLimitedError, CheckboxUnavailableError, ) from app.services.checkbox.http_client import HttpCheckboxClient BASE = "https://api.checkbox.test" CREDS = CheckboxCredentials(cash_register_id=uuid.uuid4(), license_key="lic-123", pin_code="1111") ETTN = { "id": "e-1", "status": "CREATED", "ettnNumber": "20450123456789", "totalSum": 120000, "description": "", "recipientPhone": "", "employee": {"id": "x", "dateCreated": "2026-09-23"}, "cashRegister": {"id": "y", "dateCreated": "2026-09-23"}, "lastCheckDate": "2026-09-23", "dateCreated": "2026-09-23", } def _client(interval_ms: int = 0) -> HttpCheckboxClient: settings = Settings( secret_key="test-secret-key", encryption_key="dGVzdC1lbmNyeXB0aW9uLWtleS0zMi1ieXRlcyEh", checkbox_base_url=BASE, checkbox_min_request_interval_ms=interval_ms, ) # type: ignore[arg-type] return HttpCheckboxClient(settings) def _signin(token: str = "tok-1") -> respx.Route: return respx.post(f"{BASE}/api/v1/cashier/signinPinCode").mock( return_value=Response(200, json={"access_token": token}) ) @respx.mock async def test_create_signs_in_and_sends_headers() -> None: signin = _signin() create = respx.post(f"{BASE}/api/v1/ettn").mock(return_value=Response(200, json=ETTN)) ettn = await _client().create_ettn(CREDS, {"provider": "novapost"}) assert ettn.id == "e-1" and ettn.ettn_number == "20450123456789" assert json.loads(signin.calls.last.request.content) == {"pin_code": "1111"} assert signin.calls.last.request.headers["X-License-Key"] == "lic-123" sent = create.calls.last.request assert sent.headers["Authorization"] == "Bearer tok-1" assert sent.headers["X-License-Key"] == "lic-123" assert sent.headers["X-Client-Name"] == "lux_fiscal" @respx.mock async def test_token_is_cached_between_calls() -> None: signin = _signin() respx.get(f"{BASE}/api/v1/ettn/e-1").mock(return_value=Response(200, json=ETTN)) client = _client() await client.get_ettn(CREDS, "e-1") await client.get_ettn(CREDS, "e-1") assert signin.call_count == 1 @respx.mock async def test_401_triggers_single_resign_in() -> None: signin = _signin("tok-2") route = respx.get(f"{BASE}/api/v1/ettn/e-1").mock( side_effect=[Response(401, json={"message": "expired"}), Response(200, json=ETTN)] ) ettn = await _client().get_ettn(CREDS, "e-1") assert ettn.status == "CREATED" assert signin.call_count == 2 assert route.calls.last.request.headers["Authorization"] == "Bearer tok-2" @respx.mock async def test_422_raises_checkbox_error_with_message() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock( return_value=Response( 422, json={ "message": "Validation error", "detail": [{"loc": ["body", "payments"], "msg": "bad", "type": "x"}], }, ) ) with pytest.raises(CheckboxError, match="body.payments: bad") as exc_info: await _client().create_ettn(CREDS, {}) assert not isinstance(exc_info.value, CheckboxUnavailableError) @respx.mock async def test_timeout_and_5xx_are_unavailable() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock(side_effect=httpx.ReadTimeout("slow")) with pytest.raises(CheckboxUnavailableError): await _client().create_ettn(CREDS, {}) respx.post(f"{BASE}/api/v1/ettn").mock(return_value=Response(503, text="down")) with pytest.raises(CheckboxUnavailableError): await _client().create_ettn(CREDS, {}) @respx.mock async def test_find_skips_cancelled_and_other_waybills() -> None: _signin() respx.get(f"{BASE}/api/v1/ettn").mock( return_value=Response( 200, json=[ {**ETTN, "id": "old", "status": "CANCELLED"}, {**ETTN, "id": "other", "ettnNumber": "111"}, {**ETTN, "id": "match"}, ], ) ) found = await _client().find_ettn(CREDS, "20450123456789") assert found is not None and found.id == "match" @respx.mock async def test_bad_pin_raises() -> None: respx.post(f"{BASE}/api/v1/cashier/signinPinCode").mock( return_value=Response(400, json={"message": "Невірний пін-код"}) ) with pytest.raises(CheckboxError, match="Невірний пін-код"): await _client().sign_in(CREDS) @respx.mock async def test_lowercase_status_from_live_api_is_normalized() -> None: _signin() respx.get(f"{BASE}/api/v1/ettn/e-1").mock( return_value=Response(200, json={**ETTN, "status": "done"}) ) ettn = await _client().get_ettn(CREDS, "e-1") assert ettn.status == "DONE" @respx.mock async def test_third_party_error_code_is_shown() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock( return_value=Response( 400, json={"code": "third_party.generic", "message": "Internal Server Error"} ) ) with pytest.raises(CheckboxError, match=r"third_party\.generic"): await _client().create_ettn(CREDS, {}) @respx.mock async def test_rate_limit_is_retryable() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock( return_value=Response(429, json={"message": "Занадто часто виконуються запити"}) ) with pytest.raises(CheckboxRateLimitedError): await _client().create_ettn(CREDS, {}) def _np_error(message: str, info: list[str], codes: list[str]) -> Response: """Ошибка Новой Почты в том виде, в каком её отдаёт боевой Checkbox.""" inner = { "type": "ettn", "message": message, "code": 0, "context": {"success": False, "errors": [message], "info": info, "errorCodes": codes}, } return Response(400, json={"code": "third_party.generic", "message": json.dumps(inner)}) @respx.mock async def test_np_too_many_requests_is_rate_limited() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock( return_value=_np_error("To many requests", ["Try again after 3 seconds"], ["20000401501"]) ) with pytest.raises(CheckboxRateLimitedError) as exc_info: await _client().create_ettn(CREDS, {}) assert exc_info.value.retry_after == 3 @respx.mock async def test_np_zero_retry_after_waits_at_least_a_second() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock( return_value=_np_error("To many requests", ["Try again after 0 seconds"], ["20000401501"]) ) with pytest.raises(CheckboxRateLimitedError) as exc_info: await _client().create_ettn(CREDS, {}) assert exc_info.value.retry_after == 1 @respx.mock async def test_np_timeout_is_unavailable_not_rate_limited() -> None: _signin() respx.post(f"{BASE}/api/v1/ettn").mock( return_value=_np_error("cURL error 28: SSL connection timeout", [], []) ) with pytest.raises(CheckboxUnavailableError) as exc_info: await _client().create_ettn(CREDS, {}) assert not isinstance(exc_info.value, CheckboxRateLimitedError) @respx.mock async def test_concurrent_calls_sign_in_once() -> None: signin = _signin() respx.get(f"{BASE}/api/v1/ettn/e-1").mock(return_value=Response(200, json=ETTN)) client = _client() await asyncio.gather(*(client.get_ettn(CREDS, "e-1") for _ in range(5))) assert signin.call_count == 1 @respx.mock async def test_requests_are_spaced_by_min_interval() -> None: _signin() sent_at: list[float] = [] def record(_: httpx.Request) -> Response: sent_at.append(time.monotonic()) return Response(200, json=ETTN) respx.get(f"{BASE}/api/v1/ettn/e-1").mock(side_effect=record) client = _client(interval_ms=50) await asyncio.gather(*(client.get_ettn(CREDS, "e-1") for _ in range(3))) gaps = [b - a for a, b in zip(sent_at, sent_at[1:], strict=False)] assert len(gaps) == 2 and all(gap >= 0.045 for gap in gaps)