lauadminandClaude Opus 5.5
518a99197f
Cashier creates an ETTN receipt in Checkbox bound to the TTN with payment control; Checkbox fiscalizes it itself when the parcel is paid for. - cash_registers (Fernet-encrypted license key / PIN) and receipts tables - Checkbox HTTP client + stub (ETTN does not work on test registers) - two-phase create via ARQ job, timeout reconciliation, cron status polling - /receipts and /cash-registers API, audit records - dashboard: per-order and bulk create, prepayment, cancel; cash registers page Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
290 lines
11 KiB
Python
290 lines
11 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.schemas.checkbox import EttnStatus
|
||
from app.services import receipts as svc
|
||
from app.services.checkbox.client import 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"] == [{"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}
|
||
]
|
||
assert [(d["type"], d["value"]) for d in rb["discounts"]] == [
|
||
("DISCOUNT", 5000),
|
||
("PRE_PAYMENT", 20000),
|
||
]
|
||
assert rb["payments"][0]["value"] == 85000
|
||
|
||
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 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 # второго чека на ту же ТТН нет
|
||
|
||
|
||
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]
|
||
)
|