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>
160 lines
5.5 KiB
Python
160 lines
5.5 KiB
Python
"""Тесты роутера заказов.
|
||
|
||
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
|