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>
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
"""Опрос статусов НП ключами касс и привязка заказов к кассам (`sync_np_statuses`)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from app.core import crypto
|
||
from app.db.models.cash_register import CashRegister
|
||
from app.db.models.order import Order
|
||
from app.services.nova_poshta.stub_client import StubNovaPoshtaClient
|
||
from app.services.orders import sync_np_statuses
|
||
|
||
|
||
def _register(name: str, np_key: str | None, *, is_default: bool = False) -> CashRegister:
|
||
return CashRegister(
|
||
id=uuid.uuid4(),
|
||
name=name,
|
||
license_key_enc=crypto.encrypt("lic"),
|
||
cashier_pin_enc=crypto.encrypt("1111"),
|
||
np_api_key_enc=crypto.encrypt(np_key) if np_key else None,
|
||
tax_codes=[],
|
||
is_active=True,
|
||
is_default=is_default,
|
||
created_at=datetime(2026, 9, 1, tzinfo=UTC),
|
||
)
|
||
|
||
|
||
def _order(waybill: str, register_id: uuid.UUID | None = None) -> Order:
|
||
return Order(
|
||
id=waybill[-4:],
|
||
create_date_time=datetime(2026, 9, 20, tzinfo=UTC),
|
||
waybill_number=waybill,
|
||
total_amount_kopecks=120000,
|
||
goods=[],
|
||
is_deleted=False,
|
||
cash_register_id=register_id,
|
||
)
|
||
|
||
|
||
def _status(waybill: str, *, own: bool) -> dict[str, Any]:
|
||
status = {"Number": waybill, "Status": "В дорозі", "StatusCode": "5"}
|
||
if own:
|
||
status |= {"PhoneSender": "380961112233", "AfterpaymentOnGoodsCost": 1200}
|
||
return status
|
||
|
||
|
||
class FakeSession:
|
||
"""`scalars` отдаёт кассы (с ключом НП, активные) или заказы — по сущности запроса."""
|
||
|
||
def __init__(self, registers: list[CashRegister], orders: list[Order]) -> None:
|
||
self.registers = registers
|
||
self.orders = orders
|
||
self.commits = 0
|
||
|
||
async def scalars(self, statement: Any) -> list[Any]:
|
||
entity = statement.column_descriptions[0]["entity"]
|
||
if entity is CashRegister:
|
||
return [r for r in self.registers if r.np_api_key_enc and r.is_active]
|
||
return self.orders
|
||
|
||
async def commit(self) -> None:
|
||
self.commits += 1
|
||
|
||
|
||
async def test_unbound_order_is_claimed_by_register_that_sees_it_as_own() -> None:
|
||
first, second = _register("A", "key-a", is_default=True), _register("B", "key-b")
|
||
order = _order("20450000000001")
|
||
np = StubNovaPoshtaClient(
|
||
by_key={
|
||
"key-a": {order.waybill_number: _status(order.waybill_number, own=False)},
|
||
"key-b": {order.waybill_number: _status(order.waybill_number, own=True)},
|
||
}
|
||
)
|
||
session = FakeSession([first, second], [order])
|
||
|
||
await sync_np_statuses(session, np) # type: ignore[arg-type]
|
||
|
||
assert order.cash_register_id == second.id
|
||
assert order.np_cod_amount_kopecks == 120000
|
||
assert session.commits == 1
|
||
|
||
|
||
async def test_bound_order_is_polled_only_with_its_register_key() -> None:
|
||
first, second = _register("A", "key-a", is_default=True), _register("B", "key-b")
|
||
order = _order("20450000000002", register_id=second.id)
|
||
np = StubNovaPoshtaClient(
|
||
by_key={"key-b": {order.waybill_number: _status(order.waybill_number, own=True)}}
|
||
)
|
||
|
||
await sync_np_statuses(FakeSession([first, second], [order]), np) # type: ignore[arg-type]
|
||
|
||
assert [key for key, _ in np.calls] == ["key-b"]
|
||
assert order.cash_register_id == second.id
|
||
|
||
|
||
async def test_foreign_waybill_keeps_status_but_no_register_or_cod() -> None:
|
||
register = _register("A", "key-a", is_default=True)
|
||
order = _order("20450000000003", register_id=register.id)
|
||
np = StubNovaPoshtaClient(
|
||
by_key={"key-a": {order.waybill_number: _status(order.waybill_number, own=False)}}
|
||
)
|
||
|
||
await sync_np_statuses(FakeSession([register], [order]), np) # type: ignore[arg-type]
|
||
|
||
assert order.cash_register_id is None
|
||
assert order.np_status_code == "5"
|
||
assert order.np_cod_amount_kopecks is None
|
||
|
||
|
||
async def test_no_registers_with_np_key_does_nothing() -> None:
|
||
order = _order("20450000000004")
|
||
np = StubNovaPoshtaClient()
|
||
session = FakeSession([_register("A", None)], [order])
|
||
|
||
await sync_np_statuses(session, np) # type: ignore[arg-type]
|
||
|
||
assert np.calls == []
|
||
assert session.commits == 0
|