feat(crm): dominio backend (cuentas, contactos, prospectos, embudos, oportunidades, actividades)
- 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>
This commit is contained in:
77
backend/tests/test_opportunities.py
Normal file
77
backend/tests/test_opportunities.py
Normal file
@@ -0,0 +1,77 @@
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from api.v1.modules.crm.opportunities import 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 _pipeline_with_stages(db):
|
||||
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=10), T, C
|
||||
)
|
||||
s_won = pipelines_service.create_stage(
|
||||
db, StageCreate(pipeline_id=pipeline.id, name="Ganada", position=1, probability=100, is_won=True), T, C
|
||||
)
|
||||
s_lost = pipelines_service.create_stage(
|
||||
db, StageCreate(pipeline_id=pipeline.id, name="Perdida", position=2, is_lost=True), T, C
|
||||
)
|
||||
return pipeline, s_open, s_won, s_lost
|
||||
|
||||
|
||||
def test_create_opportunity(db):
|
||||
pipeline, s_open, _, _ = _pipeline_with_stages(db)
|
||||
opp = service.create_opportunity(
|
||||
db,
|
||||
OpportunityCreate(name="Licencia Aduanasoft", pipeline_id=pipeline.id, stage_id=s_open.id, amount=15000),
|
||||
T, C,
|
||||
)
|
||||
assert opp.status == "open"
|
||||
assert opp.currency == "MXN"
|
||||
|
||||
|
||||
def test_move_to_won_closes_and_sets_probability(db):
|
||||
pipeline, s_open, s_won, _ = _pipeline_with_stages(db)
|
||||
opp = service.create_opportunity(db, OpportunityCreate(name="Deal", pipeline_id=pipeline.id, stage_id=s_open.id), T, C)
|
||||
moved = service.move_opportunity(db, opp.id, s_won.id, T, C)
|
||||
assert moved.status == "won"
|
||||
assert moved.probability == 100
|
||||
assert moved.closed_at is not None
|
||||
assert moved.stage_id == s_won.id
|
||||
|
||||
|
||||
def test_move_to_lost(db):
|
||||
pipeline, s_open, _, s_lost = _pipeline_with_stages(db)
|
||||
opp = service.create_opportunity(db, OpportunityCreate(name="Deal", pipeline_id=pipeline.id, stage_id=s_open.id), T, C)
|
||||
moved = service.move_opportunity(db, opp.id, s_lost.id, T, C)
|
||||
assert moved.status == "lost"
|
||||
assert moved.probability == 0
|
||||
assert moved.closed_at is not None
|
||||
|
||||
|
||||
def test_move_back_to_open_reopens(db):
|
||||
pipeline, s_open, s_won, _ = _pipeline_with_stages(db)
|
||||
opp = service.create_opportunity(db, OpportunityCreate(name="Deal", pipeline_id=pipeline.id, stage_id=s_open.id), T, C)
|
||||
service.move_opportunity(db, opp.id, s_won.id, T, C)
|
||||
reopened = service.move_opportunity(db, opp.id, s_open.id, T, C)
|
||||
assert reopened.status == "open"
|
||||
assert reopened.probability == 10
|
||||
assert reopened.closed_at is None
|
||||
|
||||
|
||||
def test_create_rejects_unknown_account(db):
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
service.create_opportunity(db, OpportunityCreate(name="X", account_id=999), T, C)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
|
||||
def test_only_one_default_pipeline(db):
|
||||
pipelines_service.create_pipeline(db, PipelineCreate(name="P1", is_default=True), T, C)
|
||||
pipelines_service.create_pipeline(db, PipelineCreate(name="P2", is_default=True), T, C)
|
||||
pipelines = pipelines_service.get_pipelines(db, T, C)
|
||||
defaults = [p for p in pipelines if p.is_default]
|
||||
assert len(defaults) == 1 and defaults[0].name == "P2"
|
||||
Reference in New Issue
Block a user