lauadminandClaude Opus 5.5
8b3c2b6d63
Each cash register stores its own encrypted NP API key. Status polling uses register keys and binds an order to the register whose key sees the TTN as its own (PhoneSender present); ETTN receipts are created from that register. Migration 0008 moves the old NOVA_POSHTA_API_KEY into the default register. Closes #3 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"}
|