55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
|
|
def test_router_simple_create_contact_supports_extended_payload_and_company_links(monkeypatch):
|
|
from app.contacts.backend import router_simple
|
|
|
|
insert_calls = []
|
|
update_calls = []
|
|
|
|
def fake_execute_query(query, params=None):
|
|
if "UPDATE contact_companies" in query:
|
|
update_calls.append((query, params))
|
|
return []
|
|
|
|
def fake_execute_insert(query, params=None):
|
|
insert_calls.append((query, params))
|
|
if "INSERT INTO contacts" in query:
|
|
return 321
|
|
return 1
|
|
|
|
async def fake_get_contact(contact_id):
|
|
return {"id": contact_id, "first_name": "Ada"}
|
|
|
|
monkeypatch.setattr(router_simple, "execute_query", fake_execute_query)
|
|
monkeypatch.setattr(router_simple, "execute_insert", fake_execute_insert)
|
|
monkeypatch.setattr(router_simple, "get_contact", fake_get_contact)
|
|
|
|
payload = router_simple.ContactCreate(
|
|
first_name="Ada",
|
|
last_name="Lovelace",
|
|
email="ada@example.com",
|
|
phone="11111111",
|
|
mobile="22222222",
|
|
title="CTO",
|
|
department="IT",
|
|
company_ids=[10, 20],
|
|
is_primary=True,
|
|
role="Decision maker",
|
|
is_active=True,
|
|
)
|
|
|
|
created = asyncio.run(router_simple.create_contact(payload))
|
|
|
|
assert created["id"] == 321
|
|
assert any("INSERT INTO contacts" in query for query, _ in insert_calls)
|
|
company_link_params = [params for query, params in insert_calls if "INSERT INTO contact_companies" in query]
|
|
assert len(company_link_params) == 2
|
|
assert company_link_params[0][1] == 10
|
|
assert company_link_params[1][1] == 20
|
|
assert update_calls
|