T02: companies + contacts + import/export + N:M + soft-delete + GDPR + FTS
- Company CRUD with soft-delete, FTS search (tsvector + GIN), filter, pagination - Contact CRUD with N:M company linking via company_contacts - CSV/XLSX export, CSV import with dry-run preview - GDPR hard-delete with deletion_log - Audit log on all mutations - 27 new tests (24 ACs), 56 total tests pass - Migration 0002: contacts, company_contacts, FTS search_tsv - Fixed T01 tests: POST→201, PATCH→PUT compatibility
This commit is contained in:
leocrm-bot
@@ -0,0 +1,166 @@
|
||||
"""Contact tests — ACs 14-19: CRUD, N:M, soft-delete, GDPR hard-delete."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestContactList:
|
||||
"""AC 14: List contacts paginated."""
|
||||
|
||||
async def test_list_contacts_returns_200_paginated(self, client: AsyncClient, db_session):
|
||||
"""AC 14: GET /api/v1/contacts -> 200 + paginated."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "items" in data
|
||||
assert "total" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestContactCreate:
|
||||
"""AC 15: Create contact with company_ids array -> N:M links."""
|
||||
|
||||
async def test_create_contact_with_company_ids_returns_201(self, client: AsyncClient, db_session):
|
||||
"""AC 15: POST /api/v1/contacts mit company_ids array -> 201 + N:M links."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
# Get seeded company ID
|
||||
list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
|
||||
company_id = list_resp.json()["items"][0]["id"]
|
||||
resp = await client.post(
|
||||
"/api/v1/contacts",
|
||||
json={
|
||||
"first_name": "Alice",
|
||||
"last_name": "Wonderland",
|
||||
"email": "alice@example.com",
|
||||
"company_ids": [company_id],
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["first_name"] == "Alice"
|
||||
assert data["last_name"] == "Wonderland"
|
||||
# Verify N:M link via company detail
|
||||
comp_detail = await client.get(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
|
||||
contacts = comp_detail.json()["contacts"]
|
||||
assert any(c["first_name"] == "Alice" for c in contacts)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestContactDetail:
|
||||
"""AC 16: Get contact detail with companies array."""
|
||||
|
||||
async def test_get_contact_returns_200_with_companies(self, client: AsyncClient, db_session):
|
||||
"""AC 16: GET /api/v1/contacts/{id} -> 200 + detail inkl. companies array."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
list_resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
|
||||
company_id = list_resp.json()["items"][0]["id"]
|
||||
create_resp = await client.post(
|
||||
"/api/v1/contacts",
|
||||
json={
|
||||
"first_name": "Bob",
|
||||
"last_name": "Builder",
|
||||
"company_ids": [company_id],
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
contact_id = create_resp.json()["id"]
|
||||
resp = await client.get(f"/api/v1/contacts/{contact_id}", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["first_name"] == "Bob"
|
||||
assert "companies" in data
|
||||
assert isinstance(data["companies"], list)
|
||||
assert len(data["companies"]) == 1
|
||||
assert data["companies"][0]["name"] == "Company Alpha"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestContactUpdate:
|
||||
"""AC 17: Update contact."""
|
||||
|
||||
async def test_update_contact_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 17: PUT /api/v1/contacts/{id} -> 200."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
create_resp = await client.post(
|
||||
"/api/v1/contacts",
|
||||
json={"first_name": "Old", "last_name": "Name"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
contact_id = create_resp.json()["id"]
|
||||
resp = await client.put(
|
||||
f"/api/v1/contacts/{contact_id}",
|
||||
json={"first_name": "New", "last_name": "Name", "email": "new@example.com"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["first_name"] == "New"
|
||||
assert data["email"] == "new@example.com"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestContactDelete:
|
||||
"""ACs 18-19: Soft-delete, GDPR hard-delete."""
|
||||
|
||||
async def test_delete_contact_soft_delete_returns_204(self, client: AsyncClient, db_session):
|
||||
"""AC 18: DELETE /api/v1/contacts/{id} -> 204, soft-delete."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
create_resp = await client.post(
|
||||
"/api/v1/contacts",
|
||||
json={"first_name": "Delete", "last_name": "Me"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
contact_id = create_resp.json()["id"]
|
||||
resp = await client.delete(f"/api/v1/contacts/{contact_id}", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 204
|
||||
# Verify contact not in list
|
||||
list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
|
||||
names = [f"{item['first_name']} {item['last_name']}" for item in list_resp.json()["items"]]
|
||||
assert "Delete Me" not in names
|
||||
|
||||
async def test_delete_contact_gdpr_hard_delete_returns_204(self, client: AsyncClient, db_session):
|
||||
"""AC 19: DELETE /api/v1/contacts/{id}?gdpr=true -> 204, hard-delete + deletion_log."""
|
||||
from sqlalchemy import select
|
||||
from app.models.audit import DeletionLog
|
||||
from app.models.contact import Contact
|
||||
import uuid as uuid_mod
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
create_resp = await client.post(
|
||||
"/api/v1/contacts",
|
||||
json={"first_name": "GDPR", "last_name": "Delete"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
contact_id = create_resp.json()["id"]
|
||||
resp = await client.delete(
|
||||
f"/api/v1/contacts/{contact_id}?gdpr=true",
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
# Verify physical delete — contact should not exist in DB
|
||||
q = select(Contact).where(Contact.id == uuid_mod.UUID(contact_id))
|
||||
result = await db_session.execute(q)
|
||||
assert result.scalar_one_or_none() is None
|
||||
# Verify deletion_log entry exists
|
||||
dl_q = select(DeletionLog).where(
|
||||
DeletionLog.entity_type == "contact",
|
||||
DeletionLog.entity_id == uuid_mod.UUID(contact_id),
|
||||
)
|
||||
dl_result = await db_session.execute(dl_q)
|
||||
dl_entries = dl_result.scalars().all()
|
||||
assert len(dl_entries) >= 1
|
||||
assert dl_entries[0].entity_snapshot["first_name"] == "GDPR"
|
||||
Reference in New Issue
Block a user