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>
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
"""Тесты роутера чеков: роли, постановка задач после 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"],))]
|