064d67f099
Add the Sales CRM service through collaboration, sync architecture/registry docs, expose crm.torbatyar.ir, and include a production deploy script. Co-authored-by: Cursor <cursoragent@cursor.com>
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""Phase 6.1 model / validation / audit / service tests."""
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from app.core.database import engine
|
|
from app.models.types import CustomFieldType, LeadStatus
|
|
from app.schemas.foundation import (
|
|
AttachmentCreate,
|
|
LeadCreate,
|
|
NoteCreate,
|
|
)
|
|
from app.services.audit_service import AuditService
|
|
from app.services.lead_service import LeadService
|
|
from app.services.supporting_services import AttachmentService, NoteService
|
|
from app.tests.conftest import TENANT_A
|
|
from app.validators import (
|
|
validate_custom_field_value,
|
|
validate_email,
|
|
validate_lead_status,
|
|
validate_owner,
|
|
validate_phone,
|
|
)
|
|
from shared.exceptions import AppError
|
|
from shared.security import CurrentUser
|
|
|
|
|
|
def test_email_validation():
|
|
assert validate_email("A@B.com") == "a@b.com"
|
|
with pytest.raises(AppError):
|
|
validate_email("bad-email")
|
|
|
|
|
|
def test_phone_validation():
|
|
assert validate_phone("+98 912 123 4567") is not None
|
|
with pytest.raises(AppError):
|
|
validate_phone("12")
|
|
|
|
|
|
def test_owner_required():
|
|
assert validate_owner("u-1") == "u-1"
|
|
with pytest.raises(AppError):
|
|
validate_owner("")
|
|
|
|
|
|
def test_lead_status_validation():
|
|
assert validate_lead_status("qualified") == LeadStatus.QUALIFIED
|
|
with pytest.raises(AppError):
|
|
validate_lead_status("nope")
|
|
|
|
|
|
def test_custom_field_select_validation():
|
|
validate_custom_field_value(
|
|
field_type=CustomFieldType.SELECT,
|
|
is_required=True,
|
|
options=["a", "b"],
|
|
value_text="a",
|
|
)
|
|
with pytest.raises(AppError):
|
|
validate_custom_field_value(
|
|
field_type=CustomFieldType.SELECT,
|
|
is_required=True,
|
|
options=["a", "b"],
|
|
value_text="z",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lead_service_and_audit(db_setup):
|
|
Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with Session() as session:
|
|
svc = LeadService(session)
|
|
actor = CurrentUser(user_id="actor-1", username="actor", roles=["tenant_admin"])
|
|
lead = await svc.create(
|
|
TENANT_A,
|
|
LeadCreate(
|
|
first_name="Nima",
|
|
last_name="Ahmadi",
|
|
email="nima@example.com",
|
|
owner_user_id="owner-1",
|
|
estimated_value=Decimal("1000"),
|
|
),
|
|
actor=actor,
|
|
)
|
|
assert lead.lead_number
|
|
assert lead.full_name == "Nima Ahmadi"
|
|
audits = await AuditService(session).list_for_entity(TENANT_A, "lead", lead.id)
|
|
assert any(a.action.value == "create" for a in audits)
|
|
|
|
note = await NoteService(session).create(
|
|
TENANT_A,
|
|
NoteCreate(entity_type="lead", entity_id=lead.id, body="hello"),
|
|
actor=actor,
|
|
)
|
|
assert note.version == 1
|
|
|
|
att = await AttachmentService(session).create(
|
|
TENANT_A,
|
|
AttachmentCreate(
|
|
entity_type="lead",
|
|
entity_id=lead.id,
|
|
file_storage_id="file-1",
|
|
file_name="a.pdf",
|
|
),
|
|
actor=actor,
|
|
)
|
|
assert att.file_storage_id == "file-1"
|