Files
lux_fiscal/backend/tests/test_orders_router.py
T
lauadminandClaude Sonnet 5 f4072be451 Add CRM order queue: live sync, view modal, receipt tabs, delete
Wires the CRM (exoCRM GetOrders) into the dashboard as a locally
persisted order queue instead of the previous static mockup:

- CrmClient Protocol + ExoCrmClient/StubCrmClient for the CRM's
  signed JSON-RPC API
- Order model + migration, synced from CRM on each queue view;
  soft-deleted orders stay hidden across re-syncs
- GET/DELETE /api/v1/orders with "no receipt"/"receipt issued" tabs
  (the latter is empty until Checkbox fiscalization lands)
- Dashboard: real order list, item-detail modal, tab switcher,
  one-click delete

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-22 21:39:03 +03:00

160 lines
5.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Тесты роутера заказов.
Auth подменена через dependency_overrides, сервисный слой (`app.services.orders`)
— через monkeypatch. `SessionDep` создаёт `AsyncSession`, но соединение с БД
открывается лениво только при первом запросе — раз сервисные функции его не
трогают, реальная БД не нужна.
"""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
import pytest
from fastapi.testclient import TestClient
from app.api.deps import get_crm_client, get_current_user
from app.db.models.order import Order
from app.db.models.user import User, UserRole
from app.main import app
from app.services.crm.stub_client import StubCrmClient
pytestmark = pytest.mark.usefixtures("_patch_orders_service")
def _user(role: UserRole) -> User:
return User(
id=uuid.uuid4(),
email="test@example.com",
password_hash="x",
full_name="Тест",
role=role,
is_active=True,
)
def _order(order_id: str = "1") -> Order:
return Order(
id=order_id,
create_date_time=datetime(2026, 9, 20, 10, 0, 0, tzinfo=UTC),
recipient_name="Тест Тестов",
recipient_phone="+380501112233",
recipient_email=None,
waybill_number="20450123456789",
notes=None,
total_amount_kopecks=120000,
goods=[
{
"id": "1",
"sku": "SKU-1",
"name": "Товар 1",
"price": "1200.00",
"quantity": "1.000",
"discount_amount": "0.00",
"discount_percent": "0.00",
"amount": "1200.00",
}
],
is_deleted=False,
deleted_at=None,
receipt_created_at=None,
)
@pytest.fixture
def _patch_orders_service(monkeypatch: pytest.MonkeyPatch) -> None:
import app.api.v1.orders as orders_router
async def fake_sync(session: object, crm: object) -> None:
return None
async def fake_list(session: object, *, has_receipt: bool) -> list[Order]:
return [] if has_receipt else [_order()]
async def fake_delete(session: object, order_id: str) -> Order | None:
return _order(order_id) if order_id == "1" else None
monkeypatch.setattr(orders_router.orders_service, "sync_orders_from_crm", fake_sync)
monkeypatch.setattr(orders_router.orders_service, "list_orders", fake_list)
monkeypatch.setattr(orders_router.orders_service, "delete_order", fake_delete)
async def fake_record(session: object, **kwargs: object) -> None:
return None
monkeypatch.setattr(orders_router.audit, "record", fake_record)
# session.commit() вызывается напрямую роутером после audit.record — сама
# сессия не трогает БД до первого execute/commit, поэтому патчим только
# commit, а не всю SessionDep.
from sqlalchemy.ext.asyncio import AsyncSession
async def fake_commit(self: AsyncSession) -> None:
return None
monkeypatch.setattr(AsyncSession, "commit", fake_commit)
@pytest.fixture
def client() -> TestClient:
app.dependency_overrides[get_crm_client] = lambda: StubCrmClient()
try:
yield TestClient(app)
finally:
app.dependency_overrides.pop(get_crm_client, None)
app.dependency_overrides.pop(get_current_user, None)
class TestListOrders:
@pytest.mark.parametrize("role", [UserRole.ADMIN, UserRole.CASHIER, UserRole.VIEWER])
def test_any_authenticated_role_can_list_no_receipt_orders(
self, client: TestClient, role: UserRole
) -> None:
app.dependency_overrides[get_current_user] = lambda: _user(role)
response = client.get("/api/v1/orders", params={"has_receipt": "false"})
assert response.status_code == 200
body = response.json()
assert len(body) == 1
assert body[0]["id"] == "1"
assert body[0]["total_amount"] == "1200.00"
assert body[0]["has_receipt"] is False
assert body[0]["goods"][0]["sku"] == "SKU-1"
def test_has_receipt_tab_is_empty_for_now(self, client: TestClient) -> None:
app.dependency_overrides[get_current_user] = lambda: _user(UserRole.CASHIER)
response = client.get("/api/v1/orders", params={"has_receipt": "true"})
assert response.status_code == 200
assert response.json() == []
def test_requires_authentication(self, client: TestClient) -> None:
response = client.get("/api/v1/orders")
assert response.status_code == 401
class TestDeleteOrder:
@pytest.mark.parametrize("role", [UserRole.ADMIN, UserRole.CASHIER])
def test_admin_and_cashier_can_delete(self, client: TestClient, role: UserRole) -> None:
app.dependency_overrides[get_current_user] = lambda: _user(role)
response = client.delete("/api/v1/orders/1")
assert response.status_code == 204
def test_viewer_cannot_delete(self, client: TestClient) -> None:
app.dependency_overrides[get_current_user] = lambda: _user(UserRole.VIEWER)
response = client.delete("/api/v1/orders/1")
assert response.status_code == 403
def test_deleting_unknown_order_is_404(self, client: TestClient) -> None:
app.dependency_overrides[get_current_user] = lambda: _user(UserRole.CASHIER)
response = client.delete("/api/v1/orders/does-not-exist")
assert response.status_code == 404