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>
458 lines
18 KiB
Python
458 lines
18 KiB
Python
"""Тесты сервиса ЕТТН-чеков: суммы, валидация, машина состояний.
|
||
|
||
БД заменена минимальной фейковой сессией (`get` по словарю + счётчик commit),
|
||
Checkbox — `StubCheckboxClient`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.core import crypto
|
||
from app.db.models.cash_register import CashRegister
|
||
from app.db.models.order import Order
|
||
from app.db.models.receipt import Receipt, ReceiptStatus
|
||
from app.db.models.user import User
|
||
from app.schemas.checkbox import EttnStatus
|
||
from app.services import receipts as svc
|
||
from app.services.checkbox.client import CheckboxRateLimitedError, CheckboxUnavailableError
|
||
from app.services.checkbox.stub_client import StubCheckboxClient
|
||
|
||
|
||
def _good(**overrides: Any) -> dict[str, Any]:
|
||
good = {
|
||
"id": "1",
|
||
"sku": "SKU-1",
|
||
"name": "Товар 1",
|
||
"price": "600.00",
|
||
"quantity": "2.000",
|
||
"discount_amount": "0.00",
|
||
"discount_percent": "0.00",
|
||
"amount": "1200.00",
|
||
}
|
||
return {**good, **overrides}
|
||
|
||
|
||
def _order(**overrides: Any) -> Order:
|
||
fields: dict[str, Any] = {
|
||
"id": "100",
|
||
"create_date_time": datetime(2026, 9, 20, tzinfo=UTC),
|
||
"recipient_name": "Тест",
|
||
"recipient_phone": "0501112233",
|
||
"recipient_email": "a@b.ua",
|
||
"waybill_number": "20450123456789",
|
||
"notes": None,
|
||
"total_amount_kopecks": 120000,
|
||
"goods": [_good()],
|
||
"is_deleted": False,
|
||
"receipt_created_at": None,
|
||
"np_status": "В дорозі",
|
||
"np_status_code": "5",
|
||
"np_cod_amount_kopecks": 120000,
|
||
}
|
||
fields.update(overrides)
|
||
return Order(**fields)
|
||
|
||
|
||
def _register(tax_codes: list[Any] | None = None) -> CashRegister:
|
||
return CashRegister(
|
||
id=uuid.uuid4(),
|
||
name="Каса",
|
||
license_key_enc=crypto.encrypt("lic"),
|
||
cashier_pin_enc=crypto.encrypt("1111"),
|
||
tax_codes=tax_codes or [],
|
||
is_active=True,
|
||
is_default=True,
|
||
)
|
||
|
||
|
||
class TestResolveAmounts:
|
||
def test_default_prepayment_is_total_minus_cod(self) -> None:
|
||
amounts = svc.resolve_amounts(_order(np_cod_amount_kopecks=100000), None)
|
||
assert amounts.prepayment_kopecks == 20000
|
||
assert amounts.cod_kopecks == 100000
|
||
|
||
def test_explicit_prepayment_must_match_cod(self) -> None:
|
||
with pytest.raises(svc.ReceiptValidationError, match="≠ післяплата"):
|
||
svc.resolve_amounts(_order(np_cod_amount_kopecks=100000), 10000)
|
||
|
||
@pytest.mark.parametrize(
|
||
("overrides", "message"),
|
||
[
|
||
({"waybill_number": None}, "немає ТТН"),
|
||
({"np_cod_amount_kopecks": None}, "контролю оплати"),
|
||
({"np_status_code": "9"}, "не в дорозі"),
|
||
({"np_cod_amount_kopecks": 130000}, "більша за суму замовлення"),
|
||
({"is_deleted": True}, "видалено"),
|
||
],
|
||
)
|
||
def test_rejects(self, overrides: dict[str, Any], message: str) -> None:
|
||
with pytest.raises(svc.ReceiptValidationError, match=message):
|
||
svc.resolve_amounts(_order(**overrides), None)
|
||
|
||
|
||
class TestBuildBody:
|
||
def test_goods_payments_delivery(self) -> None:
|
||
order = _order()
|
||
body = svc.build_ettn_body(order, _register(), svc.resolve_amounts(order, None))
|
||
|
||
assert body["provider"] == "novapost"
|
||
rb = body["receipt_body"]
|
||
assert rb["goods"] == [
|
||
{
|
||
"good": {"code": "SKU-1", "name": "Товар 1", "price": 60000},
|
||
"quantity": 2000,
|
||
"is_return": False,
|
||
}
|
||
]
|
||
assert rb["payments"] == [
|
||
{
|
||
"type": "ETTN",
|
||
"label": "Експрес-накладна",
|
||
"value": 120000,
|
||
"ettn": "20450123456789",
|
||
}
|
||
]
|
||
assert "discounts" not in rb
|
||
assert rb["delivery"] == {"phone": "380501112233", "emails": ["a@b.ua"]}
|
||
|
||
def test_prepayment_line_and_order_discounts_and_tax(self) -> None:
|
||
order = _order(
|
||
goods=[_good(amount="1100.00")], # скидка по строке 100 ₴
|
||
total_amount_kopecks=105000, # ещё 50 ₴ скидки на заказ
|
||
np_cod_amount_kopecks=85000, # 200 ₴ предоплаты
|
||
)
|
||
body = svc.build_ettn_body(order, _register([8]), svc.resolve_amounts(order, 20000))
|
||
rb = body["receipt_body"]
|
||
|
||
assert rb["goods"][0]["good"]["tax"] == [8]
|
||
assert rb["goods"][0]["discounts"] == [
|
||
{"type": "DISCOUNT", "mode": "VALUE", "value": 10000}
|
||
]
|
||
# Скидка заказа 50 ₴ + предоплата 200 ₴ — одной «Знижкой», без PRE_PAYMENT.
|
||
assert [(d["type"], d["value"]) for d in rb["discounts"]] == [("DISCOUNT", 25000)]
|
||
assert rb["payments"][0]["value"] == 85000
|
||
|
||
def test_all_goods_with_prices_quantities_and_sums(self) -> None:
|
||
order = _order(
|
||
goods=[
|
||
_good(id="1", sku="A", name="Сукня", price="600.00", amount="1200.00"),
|
||
_good(id="2", sku="", name="Пояс", price="150.50", quantity="1.000",
|
||
amount="150.50"),
|
||
_good(id="3", sku="C", name="Тканина", price="99.99", quantity="1.500",
|
||
amount="149.99"),
|
||
],
|
||
total_amount_kopecks=150049,
|
||
np_cod_amount_kopecks=130049, # 200 ₴ предоплаты
|
||
)
|
||
body = svc.build_ettn_body(order, _register(), svc.resolve_amounts(order, 20000))
|
||
rb = body["receipt_body"]
|
||
|
||
assert [
|
||
(g["good"]["code"], g["good"]["name"], g["good"]["price"], g["quantity"])
|
||
for g in rb["goods"]
|
||
] == [
|
||
("A", "Сукня", 60000, 2000),
|
||
("2", "Пояс", 15050, 1000), # пустой SKU → ID товара
|
||
("C", "Тканина", 9999, 1500),
|
||
]
|
||
# 99.99 × 1.5 = 149.985 → CRM округлила до 149.99, скидки по строке нет.
|
||
assert all("discounts" not in g for g in rb["goods"])
|
||
goods_total = sum(svc._line_sum(g["good"]["price"], g["quantity"]) for g in rb["goods"])
|
||
assert goods_total - 20000 == rb["payments"][0]["value"] == 130049
|
||
assert rb["discounts"] == [
|
||
{"type": "DISCOUNT", "mode": "VALUE", "value": 20000, "name": "Знижка"}
|
||
]
|
||
|
||
def test_fractional_quantity(self) -> None:
|
||
order = _order(
|
||
goods=[_good(price="100.00", quantity="2.250", amount="225.00")],
|
||
total_amount_kopecks=22500,
|
||
np_cod_amount_kopecks=22500,
|
||
)
|
||
body = svc.build_ettn_body(order, _register(), svc.resolve_amounts(order, None))
|
||
assert body["receipt_body"]["goods"][0]["quantity"] == 2250
|
||
|
||
def test_bad_phone_is_skipped(self) -> None:
|
||
order = _order(recipient_phone="12345", recipient_email=None)
|
||
body = svc.build_ettn_body(order, _register(), svc.resolve_amounts(order, None))
|
||
assert "delivery" not in body["receipt_body"]
|
||
|
||
|
||
class TestRequestReceipts:
|
||
"""Чек создаётся от кассы, к которой опрос НП привязал заказ."""
|
||
|
||
class Session:
|
||
def __init__(self, registers: list[CashRegister], orders: list[Order]) -> None:
|
||
self.registers, self.orders, self.added = registers, orders, []
|
||
|
||
async def scalars(self, statement: Any) -> list[Any]:
|
||
entity = statement.column_descriptions[0]["entity"]
|
||
if entity is CashRegister:
|
||
return self.registers
|
||
if entity is Order:
|
||
return self.orders
|
||
return [] # активные чеки заказов
|
||
|
||
def add(self, obj: Any) -> None:
|
||
self.added.append(obj)
|
||
|
||
@staticmethod
|
||
def _user() -> User:
|
||
return User(id=uuid.uuid4(), email="c@x.ua")
|
||
|
||
async def test_uses_order_register(self) -> None:
|
||
default, other = _register(), _register()
|
||
order = _order(cash_register_id=other.id)
|
||
session = self.Session([default, other], [order])
|
||
|
||
result = await svc.request_receipts(
|
||
session, [(order.id, None)], user=self._user() # type: ignore[arg-type]
|
||
)
|
||
|
||
assert result.errors == {}
|
||
assert result.created[0].cash_register_id == other.id
|
||
|
||
async def test_unbound_order_is_rejected(self) -> None:
|
||
order = _order(cash_register_id=None)
|
||
session = self.Session([_register()], [order])
|
||
|
||
result = await svc.request_receipts(
|
||
session, [(order.id, None)], user=self._user() # type: ignore[arg-type]
|
||
)
|
||
|
||
assert result.created == []
|
||
assert "Касу не визначено" in result.errors[order.id]
|
||
|
||
|
||
class FakeSession:
|
||
def __init__(self, *objects: Any) -> None:
|
||
self.objects = {(type(o), o.id): o for o in objects}
|
||
self.commits = 0
|
||
|
||
async def get(self, model: type, key: Any, **_: Any) -> Any:
|
||
return self.objects.get((model, key))
|
||
|
||
async def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
|
||
def _pending(order: Order, register: CashRegister) -> Receipt:
|
||
amounts = svc.resolve_amounts(order, None)
|
||
return Receipt(
|
||
id=uuid.uuid4(),
|
||
order_id=order.id,
|
||
cash_register_id=register.id,
|
||
waybill_number=order.waybill_number,
|
||
total_kopecks=amounts.total_kopecks,
|
||
prepayment_kopecks=amounts.prepayment_kopecks,
|
||
cod_kopecks=amounts.cod_kopecks,
|
||
status=ReceiptStatus.PENDING,
|
||
request_body=svc.build_ettn_body(order, register, amounts),
|
||
)
|
||
|
||
|
||
class TestCreateEttn:
|
||
async def test_pending_becomes_created(self) -> None:
|
||
order, register = _order(receipt_created_at=datetime.now(UTC)), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
|
||
await svc.create_ettn_for_receipt(FakeSession(order, register, receipt), client, receipt.id)
|
||
|
||
assert receipt.status == ReceiptStatus.CREATED
|
||
assert receipt.checkbox_ettn_id in client.orders
|
||
assert order.receipt_created_at is not None
|
||
|
||
async def test_second_call_is_noop(self) -> None:
|
||
order, register = _order(), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
session = FakeSession(order, register, receipt)
|
||
|
||
await svc.create_ettn_for_receipt(session, client, receipt.id)
|
||
await svc.create_ettn_for_receipt(session, client, receipt.id)
|
||
|
||
assert len(client.orders) == 1
|
||
|
||
async def test_rejection_fails_and_releases_order(self) -> None:
|
||
order, register = _order(receipt_created_at=datetime.now(UTC)), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
# Та же ТТН уже привязана — стаб отвечает ошибкой, как Checkbox.
|
||
await client.create_ettn(svc.credentials(register), receipt.request_body)
|
||
|
||
await svc.create_ettn_for_receipt(FakeSession(order, register, receipt), client, receipt.id)
|
||
|
||
assert receipt.status == ReceiptStatus.FAILED
|
||
assert "вже прив'язана" in (receipt.error or "")
|
||
assert order.receipt_created_at is None
|
||
|
||
async def test_unavailable_keeps_pending_then_reconciles(self) -> None:
|
||
order, register = _order(), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
session = FakeSession(order, register, receipt)
|
||
real_create = client.create_ettn
|
||
|
||
async def create_then_timeout(creds: Any, body: dict[str, Any]) -> Any:
|
||
await real_create(creds, body) # запрос дошёл до Checkbox…
|
||
raise CheckboxUnavailableError("timeout") # …но ответа мы не получили
|
||
|
||
client.create_ettn = create_then_timeout # type: ignore[method-assign]
|
||
await svc.create_ettn_for_receipt(session, client, receipt.id)
|
||
assert receipt.status == ReceiptStatus.PENDING and receipt.error
|
||
|
||
client.create_ettn = real_create # type: ignore[method-assign]
|
||
await svc.create_ettn_for_receipt(session, client, receipt.id)
|
||
|
||
assert receipt.status == ReceiptStatus.CREATED
|
||
assert len(client.orders) == 1 # второго чека на ту же ТТН нет
|
||
|
||
async def test_rate_limit_keeps_pending_without_unknown_outcome(self) -> None:
|
||
order, register = _order(receipt_created_at=datetime.now(UTC)), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
session = FakeSession(order, register, receipt)
|
||
|
||
async def rate_limited(creds: Any, body: dict[str, Any]) -> Any:
|
||
raise CheckboxRateLimitedError("To many requests", retry_after=1)
|
||
|
||
client.create_ettn = rate_limited # type: ignore[method-assign]
|
||
with pytest.raises(CheckboxRateLimitedError):
|
||
await svc.create_ettn_for_receipt(session, client, receipt.id)
|
||
|
||
assert receipt.status == ReceiptStatus.PENDING
|
||
assert receipt.error is None # повтор не пойдёт через find_ettn
|
||
assert order.receipt_created_at is not None # заказ не вернулся в очередь
|
||
assert session.commits == 1 # блокировка строки снята
|
||
|
||
|
||
class TestApplyEttn:
|
||
@pytest.mark.parametrize(
|
||
("checkbox_status", "expected"),
|
||
[
|
||
(EttnStatus.CREATED, ReceiptStatus.CREATED),
|
||
(EttnStatus.DONE, ReceiptStatus.DONE),
|
||
(EttnStatus.DONE_WITHOUT_SMS, ReceiptStatus.DONE),
|
||
(EttnStatus.RETURNED, ReceiptStatus.RETURNED),
|
||
(EttnStatus.RECEIPT_ERROR, ReceiptStatus.RECEIPT_ERROR),
|
||
(EttnStatus.CANCELLED, ReceiptStatus.CANCELLED),
|
||
],
|
||
)
|
||
async def test_status_mapping(self, checkbox_status: str, expected: ReceiptStatus) -> None:
|
||
order, register = _order(), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
await svc.create_ettn_for_receipt(FakeSession(order, register, receipt), client, receipt.id)
|
||
|
||
client.set_status(receipt.checkbox_ettn_id or "", checkbox_status, raw_error="boom")
|
||
ettn = await client.get_ettn(svc.credentials(register), receipt.checkbox_ettn_id or "")
|
||
svc._apply_ettn(receipt, ettn)
|
||
|
||
assert receipt.status == expected
|
||
if expected == ReceiptStatus.DONE:
|
||
assert receipt.checkbox_receipt_id
|
||
if expected == ReceiptStatus.RECEIPT_ERROR:
|
||
assert receipt.error == "boom"
|
||
|
||
|
||
class TestCancel:
|
||
async def test_cancel_created(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
order, register = _order(receipt_created_at=datetime.now(UTC)), _register()
|
||
receipt = _pending(order, register)
|
||
client = StubCheckboxClient()
|
||
session = FakeSession(order, register, receipt)
|
||
session.add = lambda obj: None # type: ignore[attr-defined]
|
||
await svc.create_ettn_for_receipt(session, client, receipt.id)
|
||
order.receipt_created_at = datetime.now(UTC)
|
||
|
||
await svc.cancel_receipt(session, client, receipt.id, user=None) # type: ignore[arg-type]
|
||
|
||
assert receipt.status == ReceiptStatus.CANCELLED
|
||
assert client.orders[receipt.checkbox_ettn_id or ""].status == EttnStatus.CANCELLED
|
||
assert order.receipt_created_at is None
|
||
|
||
async def test_cannot_cancel_done(self) -> None:
|
||
order, register = _order(), _register()
|
||
receipt = _pending(order, register)
|
||
receipt.status = ReceiptStatus.DONE
|
||
with pytest.raises(svc.ReceiptStateError):
|
||
await svc.cancel_receipt(
|
||
FakeSession(order, register, receipt), # type: ignore[arg-type]
|
||
StubCheckboxClient(),
|
||
receipt.id,
|
||
user=None, # type: ignore[arg-type]
|
||
)
|
||
|
||
|
||
class CrmSession:
|
||
"""Фейковая сессия для sync_crm_statuses: scalars → id, scalar → строка с «блокировкой»."""
|
||
|
||
def __init__(self, *receipts: Receipt) -> None:
|
||
self.receipts = {r.id: r for r in receipts}
|
||
self.added: list[Any] = []
|
||
self.commits = 0
|
||
self.rollbacks = 0
|
||
|
||
async def scalars(self, _: Any) -> list[uuid.UUID]:
|
||
return [r.id for r in self.receipts.values() if r.crm_status_set_at is None]
|
||
|
||
async def scalar(self, query: Any) -> Receipt | None:
|
||
receipt_id = query.whereclause.clauses[0].right.value
|
||
receipt = self.receipts[receipt_id]
|
||
return receipt if receipt.crm_status_set_at is None else None
|
||
|
||
def add(self, obj: Any) -> None:
|
||
self.added.append(obj)
|
||
|
||
async def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
async def rollback(self) -> None:
|
||
self.rollbacks += 1
|
||
|
||
|
||
class TestSyncCrmStatuses:
|
||
async def test_sets_packed_once(self) -> None:
|
||
from app.services.crm.stub_client import StubCrmClient
|
||
|
||
order, register = _order(), _register()
|
||
receipt = _pending(order, register)
|
||
receipt.status = ReceiptStatus.CREATED
|
||
session = CrmSession(receipt)
|
||
crm = StubCrmClient()
|
||
|
||
await svc.sync_crm_statuses(session, crm) # type: ignore[arg-type]
|
||
await svc.sync_crm_statuses(session, crm) # type: ignore[arg-type]
|
||
|
||
assert crm.statuses == {"100": "PACKED"}
|
||
assert receipt.crm_status_set_at is not None
|
||
assert session.commits == 1 # второй проход ничего не делает
|
||
|
||
async def test_crm_failure_is_retried_later(self) -> None:
|
||
from app.services.crm.client import CrmError
|
||
from app.services.crm.stub_client import StubCrmClient
|
||
|
||
order, register = _order(), _register()
|
||
receipt = _pending(order, register)
|
||
receipt.status = ReceiptStatus.CREATED
|
||
session = CrmSession(receipt)
|
||
crm = StubCrmClient()
|
||
|
||
async def broken(**_: Any) -> None:
|
||
raise CrmError("CRM вернула ошибку")
|
||
|
||
real_set_status = crm.set_status
|
||
crm.set_status = broken # type: ignore[method-assign]
|
||
await svc.sync_crm_statuses(session, crm) # type: ignore[arg-type]
|
||
assert receipt.crm_status_set_at is None and session.rollbacks == 1
|
||
|
||
crm.set_status = real_set_status # type: ignore[method-assign]
|
||
await svc.sync_crm_statuses(session, crm) # type: ignore[arg-type]
|
||
assert crm.statuses == {"100": "PACKED"}
|