a196c44fae
Diagrama 1 (CRM comercial) y Diagrama 2 (Operaciones) del spec de agente de carga: - crm.service_requests (RFQ) + crm.rate_requests (solicitud de tarifas a proveedores) - crm.quotes + crm.quote_items: conceptos costo/venta/margen, totales automáticos, estados borrador→enviada→aceptada/rechazada - schema ops: ops.shipments (booking, Cut Off, ETD/ETA, naviera/agente aduanal/destino) y ops.shipment_documents (MBL/HBL, MAWB/HAWB, CMR…) - liberar-a-operaciones: crea el embarque desde la cotización aceptada - migración b3c4d5e6f7a8, routers/permisos, seed del flujo completo - 52 tests pytest en verde Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from api.v1.modules.crm.accounts import service as accounts_service
|
|
from api.v1.modules.crm.accounts.dto import AccountCreate
|
|
from api.v1.modules.crm.service_requests import service
|
|
from api.v1.modules.crm.service_requests.dto import (
|
|
RateRequestCreate,
|
|
ServiceRequestCreate,
|
|
ServiceRequestUpdate,
|
|
)
|
|
|
|
T, C = 1, 1
|
|
|
|
|
|
def test_create_service_request(db):
|
|
acc = accounts_service.create_account(db, AccountCreate(name="Cliente"), T, C)
|
|
sr = service.create_service_request(
|
|
db,
|
|
ServiceRequestCreate(
|
|
account_id=acc.id, operation_type="exportacion", transport_mode="maritimo",
|
|
service_type="puerta_puerta", origin="Manzanillo", destination="Long Beach",
|
|
incoterm="FOB", load_type="FCL",
|
|
),
|
|
T, C, user_id="dev",
|
|
)
|
|
assert sr.id is not None
|
|
assert sr.status == "nueva"
|
|
assert sr.created_by == "dev"
|
|
|
|
|
|
def test_service_request_rejects_unknown_account(db):
|
|
with pytest.raises(HTTPException) as exc:
|
|
service.create_service_request(db, ServiceRequestCreate(account_id=999, operation_type="importacion"), T, C)
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
def test_filter_by_operation_and_status(db):
|
|
service.create_service_request(db, ServiceRequestCreate(operation_type="exportacion"), T, C)
|
|
service.create_service_request(db, ServiceRequestCreate(operation_type="importacion"), T, C)
|
|
exp = service.get_service_requests(db, T, C, operation_type="exportacion")
|
|
assert len(exp) == 1 and exp[0].operation_type == "exportacion"
|
|
|
|
|
|
def test_rate_request_requires_service_request(db):
|
|
with pytest.raises(HTTPException) as exc:
|
|
service.create_rate_request(
|
|
db, RateRequestCreate(service_request_id=999, concept="flete_internacional"), T, C
|
|
)
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_rate_request_ok_and_listed(db):
|
|
sr = service.create_service_request(db, ServiceRequestCreate(operation_type="exportacion"), T, C)
|
|
service.create_rate_request(
|
|
db, RateRequestCreate(service_request_id=sr.id, concept="flete_internacional", rate_amount=1200, currency="USD"),
|
|
T, C,
|
|
)
|
|
rates = service.get_rate_requests(db, T, C, service_request_id=sr.id)
|
|
assert len(rates) == 1 and rates[0].concept == "flete_internacional"
|
|
|
|
|
|
def test_update_service_request_status(db):
|
|
sr = service.create_service_request(db, ServiceRequestCreate(operation_type="exportacion"), T, C)
|
|
upd = service.update_service_request(db, sr.id, ServiceRequestUpdate(status="en_analisis"), T, C)
|
|
assert upd.status == "en_analisis"
|