088a8fc4df
- Nuevo schema `crm` con 7 tablas multi-tenant (TenantScopedMixin + soft delete) - Módulos FastAPI por dominio: models/dto/service/routes (patrón example) - Métricas del dashboard (KPIs + embudo por etapa) - Conversión de prospecto → cuenta/contacto/oportunidad (idempotente) - Movimiento de oportunidad entre etapas (Kanban) con estado/probabilidad derivados - 25 permisos registrados en PermissionRegistry - Migración Alembic con upgrade/downgrade completos - 24 tests de servicios (pytest) en verde Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
43 lines
2.0 KiB
Python
43 lines
2.0 KiB
Python
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.metrics import service as metrics_service
|
|
from api.v1.modules.crm.opportunities import service as opportunities_service
|
|
from api.v1.modules.crm.opportunities.dto import OpportunityCreate
|
|
from api.v1.modules.crm.pipelines import service as pipelines_service
|
|
from api.v1.modules.crm.pipelines.dto import PipelineCreate, StageCreate
|
|
|
|
T, C = 1, 1
|
|
|
|
|
|
def test_metrics_counts_and_pipeline(db):
|
|
accounts_service.create_account(db, AccountCreate(name="Cuenta 1"), T, C)
|
|
pipeline = pipelines_service.create_pipeline(db, PipelineCreate(name="Ventas", is_default=True), T, C)
|
|
s_open = pipelines_service.create_stage(
|
|
db, StageCreate(pipeline_id=pipeline.id, name="Prospecto", position=0, probability=20), T, C
|
|
)
|
|
s_won = pipelines_service.create_stage(
|
|
db, StageCreate(pipeline_id=pipeline.id, name="Ganada", position=1, is_won=True), T, C
|
|
)
|
|
|
|
opportunities_service.create_opportunity(
|
|
db, OpportunityCreate(name="Abierta", pipeline_id=pipeline.id, stage_id=s_open.id, amount=1000), T, C
|
|
)
|
|
won = opportunities_service.create_opportunity(
|
|
db, OpportunityCreate(name="Cerrada", pipeline_id=pipeline.id, stage_id=s_open.id, amount=2000), T, C
|
|
)
|
|
opportunities_service.move_opportunity(db, won.id, s_won.id, T, C)
|
|
|
|
metrics = metrics_service.get_metrics(db, T, C)
|
|
|
|
assert metrics["total_accounts"] == 1
|
|
assert metrics["open_opportunities"] == 1
|
|
assert float(metrics["open_pipeline_value"]) == 1000.0
|
|
assert metrics["won_opportunities"] == 1
|
|
assert float(metrics["won_value"]) == 2000.0
|
|
|
|
# El embudo reporta ambas etapas; la abierta tiene 1 oportunidad
|
|
by_stage = {row["stage_name"]: row for row in metrics["by_stage"]}
|
|
assert by_stage["Prospecto"]["count"] == 1
|
|
assert float(by_stage["Prospecto"]["value"]) == 1000.0
|
|
assert by_stage["Ganada"]["count"] == 0
|