Add Checkbox ETTN receipts for Nova Poshta COD waybills
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>
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
"""Тесты роутера чеков: роли, постановка задач после commit, ответ с ошибками по заказам."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.deps import get_current_user
|
||||
from app.db.models.receipt import Receipt, ReceiptStatus
|
||||
from app.db.models.user import User, UserRole
|
||||
from app.main import app
|
||||
from app.services import receipts as receipts_service
|
||||
from app.services.task_queue import get_task_queue
|
||||
|
||||
|
||||
class FakeQueue:
|
||||
def __init__(self) -> None:
|
||||
self.jobs: list[tuple[str, tuple[Any, ...]]] = []
|
||||
|
||||
async def enqueue(self, function: str, *args: Any) -> None:
|
||||
self.jobs.append((function, args))
|
||||
|
||||
|
||||
def _user(role: UserRole) -> User:
|
||||
return User(id=uuid.uuid4(), email="t@e.ua", password_hash="x", full_name="Т", role=role)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def queue(monkeypatch: pytest.MonkeyPatch) -> FakeQueue:
|
||||
fake = FakeQueue()
|
||||
receipt = Receipt(
|
||||
id=uuid.uuid4(),
|
||||
order_id="1",
|
||||
waybill_number="20450123456789",
|
||||
total_kopecks=120000,
|
||||
prepayment_kopecks=0,
|
||||
cod_kopecks=120000,
|
||||
status=ReceiptStatus.PENDING,
|
||||
created_at=datetime.now(UTC),
|
||||
)
|
||||
|
||||
async def fake_request(session: Any, items: Any, *, user: Any, request: Any) -> Any:
|
||||
return receipts_service.RequestResult(created=[receipt], errors={"2": "У заказа нет ТТН"})
|
||||
|
||||
async def fake_commit(self: AsyncSession) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(receipts_service, "request_receipts", fake_request)
|
||||
monkeypatch.setattr(AsyncSession, "commit", fake_commit)
|
||||
app.dependency_overrides[get_task_queue] = lambda: fake
|
||||
yield fake
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def test_viewer_cannot_create(queue: FakeQueue) -> None:
|
||||
app.dependency_overrides[get_current_user] = lambda: _user(UserRole.VIEWER)
|
||||
response = TestClient(app).post("/api/v1/receipts", json={"items": [{"order_id": "1"}]})
|
||||
assert response.status_code == 403
|
||||
assert queue.jobs == []
|
||||
|
||||
|
||||
def test_cashier_creates_and_enqueues(queue: FakeQueue) -> None:
|
||||
app.dependency_overrides[get_current_user] = lambda: _user(UserRole.CASHIER)
|
||||
|
||||
response = TestClient(app).post(
|
||||
"/api/v1/receipts", json={"items": [{"order_id": "1"}, {"order_id": "2"}]}
|
||||
)
|
||||
|
||||
assert response.status_code == 202
|
||||
body = response.json()
|
||||
assert [r["order_id"] for r in body["created"]] == ["1"]
|
||||
assert body["errors"] == {"2": "У заказа нет ТТН"}
|
||||
assert queue.jobs == [("create_ettn_receipt", (body["created"][0]["id"],))]
|
||||
Reference in New Issue
Block a user