lauadminandClaude Opus 5.5
b8f2fe6b6e
- All frontend pages, labels, notices and errors; html lang=uk, uk-UA money format - Brand "Assistant System" in the top bar and page title - Backend error details returned to the UI (auth, orders, receipts, cash registers, Checkbox/CRM/NP errors) and CLI output - Tests updated for the new messages Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
74 lines
3.4 KiB
Python
74 lines
3.4 KiB
Python
"""Стаб Checkbox для тестов и локальной разработки.
|
|
|
|
ЕТТН-чеки на тестовой кассе Checkbox не работают, поэтому полный цикл без
|
|
боевой кассы прогоняется только через стаб. Состояние — в памяти процесса.
|
|
|
|
`auto_complete_after=N`: чек переходит в `DONE` на N-м вызове `get_ettn` —
|
|
имитирует получение посылки клиентом.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from app.schemas.checkbox import EttnOut, EttnStatus
|
|
from app.services.checkbox.client import CheckboxCredentials, CheckboxError
|
|
|
|
|
|
class StubCheckboxClient:
|
|
def __init__(self, *, auto_complete_after: int | None = None) -> None:
|
|
self.auto_complete_after = auto_complete_after
|
|
self.orders: dict[str, EttnOut] = {}
|
|
self.bodies: dict[str, dict[str, Any]] = {}
|
|
self._polls: dict[str, int] = {}
|
|
|
|
def set_status(self, ettn_id: str, status: str, *, raw_error: str | None = None) -> None:
|
|
ettn = self.orders[ettn_id]
|
|
receipt_id = str(uuid.uuid4()) if status.startswith(EttnStatus.DONE) else ettn.receipt_id
|
|
self.orders[ettn_id] = ettn.model_copy(
|
|
update={"status": status, "receipt_id": receipt_id, "raw_error": raw_error}
|
|
)
|
|
|
|
async def sign_in(self, creds: CheckboxCredentials) -> None:
|
|
if not creds.license_key or not creds.pin_code:
|
|
raise CheckboxError("Вхід касира не вдався: порожній ключ або PIN")
|
|
|
|
async def create_ettn(self, creds: CheckboxCredentials, body: dict[str, Any]) -> EttnOut:
|
|
waybill = body["receipt_body"]["payments"][0]["ettn"]
|
|
if await self.find_ettn(creds, waybill) is not None:
|
|
raise CheckboxError(f"ЕТТН {waybill} вже прив'язана до чека")
|
|
ettn = EttnOut(
|
|
id=str(uuid.uuid4()),
|
|
status=EttnStatus.CREATED,
|
|
ettn_number=waybill,
|
|
total_sum=body["receipt_body"]["payments"][0]["value"],
|
|
)
|
|
self.orders[ettn.id] = ettn
|
|
self.bodies[ettn.id] = body
|
|
return ettn
|
|
|
|
async def get_ettn(self, creds: CheckboxCredentials, ettn_id: str) -> EttnOut:
|
|
if ettn_id not in self.orders:
|
|
raise CheckboxError(f"ЕТТН-чек {ettn_id} не знайдено")
|
|
self._polls[ettn_id] = self._polls.get(ettn_id, 0) + 1
|
|
if (
|
|
self.auto_complete_after is not None
|
|
and self.orders[ettn_id].status == EttnStatus.CREATED
|
|
and self._polls[ettn_id] >= self.auto_complete_after
|
|
):
|
|
self.set_status(ettn_id, EttnStatus.DONE)
|
|
return self.orders[ettn_id]
|
|
|
|
async def find_ettn(self, creds: CheckboxCredentials, waybill_number: str) -> EttnOut | None:
|
|
for ettn in self.orders.values():
|
|
if ettn.ettn_number == waybill_number and ettn.status != EttnStatus.CANCELLED:
|
|
return ettn
|
|
return None
|
|
|
|
async def delete_ettn(self, creds: CheckboxCredentials, ettn_id: str) -> None:
|
|
# API и worker — разные процессы с разными стабами: неизвестный id
|
|
# считаем уже удалённым, иначе локально отмена не работала бы.
|
|
if ettn_id in self.orders:
|
|
self.set_status(ettn_id, EttnStatus.CANCELLED)
|