morteza/TorbatYar
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
9e1fcd9176
TorbatYar/backend/services/crm/app/tests/test_api.py
Mortezakoohjani 064d67f099 Ship CRM Core Platform (phases 6.0–6.3) with docs and prod deploy. 
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>
2026-07-24 20:32:00 +03:30

326 lines
10 KiB
Python
Raw Blame History

"""CRM API and event contract tests — Phase 6.0 + 6.1."""
from __future__ import annotations
import pytest
from app.events.publisher import get_event_publisher
from app.events.types import CrmEventType
from app.tests.conftest import TENANT_A, TENANT_B, tenant_headers
@pytest.mark.asyncio
async def test_health(client):
resp = await client.get("/health")
assert resp.status_code == 200
body = resp.json()
assert body["service"] == "crm-service"
assert body["status"] == "ok"
@pytest.mark.asyncio
async def test_create_lead_publishes_event(client):
resp = await client.post(
"/api/v1/leads",
json={
"title": "New Lead",
"first_name": "Ali",
"last_name": "Rezaei",
"email": "lead@example.com",
"mobile": "+989121234567",
"source": "web",
"owner_user_id": "owner-1",
},
headers=tenant_headers(TENANT_A),
)
assert resp.status_code == 201, resp.text
data = resp.json()
assert data["full_name"]
assert data["lead_number"]
assert data["tenant_id"] == str(TENANT_A)
assert data["owner_user_id"] == "owner-1"
events = get_event_publisher().published
assert any(e.event_type == CrmEventType.LEAD_CREATED.value for e in events)
@pytest.mark.asyncio
async def test_contact_and_organization_flow(client):
org = await client.post(
"/api/v1/organizations",
json={
"name": "Buyer Co",
"industry": "retail",
"kind": "company",
"legal_name": "Buyer Co Ltd",
},
headers=tenant_headers(TENANT_A),
)
assert org.status_code == 201, org.text
org_id = org.json()["id"]
contact = await client.post(
"/api/v1/contacts",
json={
"first_name": "Ada",
"last_name": "Lovelace",
"email": "ada@example.com",
"phones": ["+1-555-0100", "+1-555-0101"],
"emails": ["ada@example.com", "ada.work@example.com"],
"kind": "business",
"organization_id": org_id,
},
headers=tenant_headers(TENANT_A),
)
assert contact.status_code == 201, contact.text
assert contact.json()["organization_id"] == org_id
assert contact.json()["full_name"] == "Ada Lovelace"
@pytest.mark.asyncio
async def test_pipeline_opportunity_win_flow(client):
pipeline = await client.post(
"/api/v1/pipelines",
json={"name": "Default", "code": "DEFAULT", "is_default": True},
headers=tenant_headers(TENANT_A),
)
assert pipeline.status_code == 201
pipeline_id = pipeline.json()["id"]
stage = await client.post(
"/api/v1/pipelines/stages",
json={
"pipeline_id": pipeline_id,
"name": "Proposal",
"code": "PROPOSAL",
"position": 1,
"win_probability": 60,
},
headers=tenant_headers(TENANT_A),
)
assert stage.status_code == 201
stage_id = stage.json()["id"]
opportunity = await client.post(
"/api/v1/opportunities",
json={
"name": "Big Deal",
"pipeline_id": pipeline_id,
"stage_id": stage_id,
"amount": "150000.0000",
"currency_code": "IRR",
},
headers=tenant_headers(TENANT_A),
)
assert opportunity.status_code == 201
opportunity_id = opportunity.json()["id"]
won = await client.post(
f"/api/v1/opportunities/{opportunity_id}/win",
headers=tenant_headers(TENANT_A),
)
assert won.status_code == 200
assert won.json()["status"] == "won"
events = get_event_publisher().published
assert any(e.event_type == CrmEventType.OPPORTUNITY_WON.value for e in events)
@pytest.mark.asyncio
async def test_activity_complete_and_quote_create(client):
activity = await client.post(
"/api/v1/activities",
json={"subject": "Discovery call", "activity_type": "call"},
headers=tenant_headers(TENANT_A),
)
assert activity.status_code == 201
activity_id = activity.json()["id"]
completed = await client.post(
f"/api/v1/activities/{activity_id}/complete",
headers=tenant_headers(TENANT_A),
)
assert completed.status_code == 200
assert completed.json()["status"] == "completed"
quote = await client.post(
"/api/v1/quotes",
json={
"title": "Q1 Offer",
"quote_number": "Q-1001",
"amount": "25000.0000",
},
headers=tenant_headers(TENANT_A),
)
assert quote.status_code == 201
events = get_event_publisher().published
assert any(e.event_type == CrmEventType.ACTIVITY_COMPLETED.value for e in events)
assert any(e.event_type == CrmEventType.QUOTE_CREATED.value for e in events)
@pytest.mark.asyncio
async def test_lead_update_publishes_event(client):
created = await client.post(
"/api/v1/leads",
json={"title": "Update me", "owner_user_id": "owner-1"},
headers=tenant_headers(TENANT_A),
)
assert created.status_code == 201, created.text
lead_id = created.json()["id"]
updated = await client.patch(
f"/api/v1/leads/{lead_id}",
json={"status": "contacted"},
headers=tenant_headers(TENANT_A),
)
assert updated.status_code == 200
assert updated.json()["status"] == "contacted"
events = get_event_publisher().published
assert any(e.event_type == CrmEventType.LEAD_UPDATED.value for e in events)
@pytest.mark.asyncio
async def test_phase61_entities_notes_attachments_tags_custom_fields(client):
lead = await client.post(
"/api/v1/leads",
json={
"first_name": "Sara",
"last_name": "Karimi",
"owner_user_id": "owner-1",
"email": "sara@example.com",
},
headers=tenant_headers(TENANT_A),
)
assert lead.status_code == 201, lead.text
lead_id = lead.json()["id"]
source = await client.post(
"/api/v1/lookups/lead-sources",
json={"code": "WEB", "name": "Website"},
headers=tenant_headers(TENANT_A),
)
assert source.status_code == 201
status_def = await client.post(
"/api/v1/lookups/lead-statuses",
json={"code": "NEW", "name": "New", "position": 1},
headers=tenant_headers(TENANT_A),
)
assert status_def.status_code == 201
tag = await client.post(
"/api/v1/tags",
json={"name": "hot", "color": "#ff0000", "group_name": "priority"},
headers=tenant_headers(TENANT_A),
)
assert tag.status_code == 201
tag_id = tag.json()["id"]
assigned = await client.post(
f"/api/v1/tags/leads/{lead_id}",
json={"tag_id": tag_id},
headers=tenant_headers(TENANT_A),
)
assert assigned.status_code == 200
address = await client.post(
"/api/v1/addresses",
json={
"entity_type": "lead",
"entity_id": lead_id,
"address_type": "office",
"line1": "Valiasr St 1",
"city": "Tehran",
"country": "IR",
"latitude": "35.6892000",
"longitude": "51.3890000",
"is_default": True,
},
headers=tenant_headers(TENANT_A),
)
assert address.status_code == 201, address.text
note = await client.post(
"/api/v1/notes",
json={
"entity_type": "lead",
"entity_id": lead_id,
"body": "<p>First note</p>",
"is_pinned": True,
},
headers=tenant_headers(TENANT_A),
)
assert note.status_code == 201
note_id = note.json()["id"]
note_upd = await client.patch(
f"/api/v1/notes/{note_id}",
json={"body": "<p>Updated note</p>"},
headers=tenant_headers(TENANT_A),
)
assert note_upd.status_code == 200
assert note_upd.json()["version"] == 2
attachment = await client.post(
"/api/v1/attachments",
json={
"entity_type": "lead",
"entity_id": lead_id,
"file_storage_id": "fs-123",
"file_name": "contract.pdf",
"content_type": "application/pdf",
"kind": "contract",
},
headers=tenant_headers(TENANT_A),
)
assert attachment.status_code == 201
field = await client.post(
"/api/v1/custom-fields",
json={
"entity_type": "lead",
"field_key": "budget_band",
"label": "Budget Band",
"field_type": "select",
"options": ["low", "mid", "high"],
"is_required": True,
},
headers=tenant_headers(TENANT_A),
)
assert field.status_code == 201
field_id = field.json()["id"]
value = await client.put(
"/api/v1/custom-fields/values",
json={
"custom_field_id": field_id,
"entity_type": "lead",
"entity_id": lead_id,
"value_text": "mid",
},
headers=tenant_headers(TENANT_A),
)
assert value.status_code == 200
assign = await client.post(
f"/api/v1/leads/{lead_id}/assign",
json={"assigned_user_id": "rep-9", "note": "handoff"},
headers=tenant_headers(TENANT_A),
)
assert assign.status_code == 200
assert assign.json()["assigned_user_id"] == "rep-9"
audit = await client.get(
"/api/v1/audit",
params={"entity_type": "lead", "entity_id": lead_id},
headers=tenant_headers(TENANT_A),
)
assert audit.status_code == 200
assert len(audit.json()) >= 1
events = get_event_publisher().published
assert any(e.event_type == CrmEventType.NOTE_CREATED.value for e in events)
assert any(e.event_type == CrmEventType.ATTACHMENT_ADDED.value for e in events)
# tenant isolation on note list
notes_b = await client.get(
"/api/v1/notes",
params={"entity_type": "lead", "entity_id": lead_id},
headers=tenant_headers(TENANT_B),
)
assert notes_b.status_code == 200
assert notes_b.json() == []
Reference in New Issue View Git Blame Copy Permalink