leocrm/tests/test_contacts.py

"""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, login_client, seed_tenant_and_users


@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={
                "firstname": "Alice",
                "surname": "Wonderland",
                "email": "alice@example.com",
                "company_ids": [company_id],
            },
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 201
        data = resp.json()
        assert data["firstname"] == "Alice"
        assert data["surname"] == "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["firstname"] == "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={
                "firstname": "Bob",
                "surname": "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["firstname"] == "Bob"
        assert "contact_persons" in data
        assert isinstance(data["contact_persons"], list)


@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={"firstname": "Old", "surname": "Name"},
            headers=ORIGIN_HEADER,
        )
        contact_id = create_resp.json()["id"]
        resp = await client.put(
            f"/api/v1/contacts/{contact_id}",
            json={"firstname": "New", "surname": "Name", "email_1": "new@example.com"},
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["firstname"] == "New"
        assert data["email_1"] == "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={"firstname": "Delete", "surname": "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["firstname"]} {item["surname"]}" 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}?hard=true -> 204, hard-delete + audit log."""
        import uuid as uuid_mod

        from sqlalchemy import select

        from app.models.audit import AuditLog
        from app.models.contact import Contact

        await seed_tenant_and_users(db_session)
        await login_client(client, "admin@tenanta.com")
        create_resp = await client.post(
            "/api/v1/contacts",
            json={"firstname": "GDPR", "surname": "Delete"},
            headers=ORIGIN_HEADER,
        )
        contact_id = create_resp.json()["id"]
        resp = await client.delete(
            f"/api/v1/contacts/{contact_id}?hard=true",
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 204
        # Refresh session to see committed changes from API
        db_session.expire_all()
        # 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 audit log entry exists
        al_q = select(AuditLog).where(
            AuditLog.entity_type == "contact",
            AuditLog.entity_id == uuid_mod.UUID(contact_id),
        )
        al_result = await db_session.execute(al_q)
        al_entries = al_result.scalars().all()
        assert len(al_entries) >= 1
        assert any(e.action == "hard_delete" for e in al_entries)


# ── Visibility filter test ──


@pytest.mark.asyncio
class TestContactVisibilityFilter:
    """Row-level visibility filter hides non-owned, non-shared contacts."""

    async def test_visibility_filter_hides_owned_contact_from_viewer(self, client: AsyncClient, db_session):
        """A contact owned by admin_a is not visible to viewer_a in the list."""
        await seed_tenant_and_users(db_session)
        await login_client(client, "admin@tenanta.com")
        # Create a contact as admin (owner_id = admin_a)
        create_resp = await client.post(
            "/api/v1/contacts",
            json={"firstname": "Owned", "surname": "Contact"},
            headers=ORIGIN_HEADER,
        )
        assert create_resp.status_code == 201
        owned_id = create_resp.json()["id"]

        # Admin (owner) sees it in the list
        list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
        assert list_resp.status_code == 200
        assert any(c["id"] == owned_id for c in list_resp.json()["items"])

        # Viewer (non-owner, not shared) must NOT see it
        await login_client(client, "viewer@tenanta.com")
        viewer_list = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
        assert viewer_list.status_code == 200
        assert all(c["id"] != owned_id for c in viewer_list.json()["items"])

    async def test_visibility_filter_shows_tenant_owned_contact(self, client: AsyncClient, db_session):
        """A tenant-owned contact (owner_id NULL) is visible to all users with read permission."""
        await seed_tenant_and_users(db_session)
        await login_client(client, "viewer@tenanta.com")
        # Seed company 'Company Alpha' has owner_id NULL → tenant-owned → visible to viewer
        list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
        assert list_resp.status_code == 200
        names = [c["name"] for c in list_resp.json()["items"]]
        assert "Company Alpha" in names