Upload tests/conftest.py

tests/conftest.py

@@ -0,0 +1,275 @@
+"""Test fixtures: in-memory SQLite DB, async test client, auth helpers.
+
+Design:
+- One AsyncEngine + session_factory per test (function-scoped) so each test
+  has a fresh, isolated DB.
+- The FastAPI app's get_db dependency is overridden to use the same engine.
+- register/headers/login helpers talk to the HTTP API (integration tests).
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from typing import Any
+
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import (
+    AsyncEngine,
+    AsyncSession,
+    async_sessionmaker,
+    create_async_engine,
+)
+
+from app.core.db import Base, get_db
+from app.main import create_app
+
+
+@pytest_asyncio.fixture
+async def engine() -> AsyncGenerator[AsyncEngine, None]:
+    """Per-test in-memory SQLite engine with schema created from metadata."""
+    eng = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        echo=False,
+        future=True,
+    )
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield eng
+    await eng.dispose()
+
+
+@pytest_asyncio.fixture
+async def session_factory(
+    engine: AsyncEngine,
+) -> async_sessionmaker[AsyncSession]:
+    """Session factory bound to the test engine."""
+    return async_sessionmaker(
+        bind=engine, expire_on_commit=False, class_=AsyncSession
+    )
+
+
+@pytest_asyncio.fixture
+async def client(
+    session_factory: async_sessionmaker[AsyncSession],
+) -> AsyncGenerator[AsyncClient, None]:
+    """httpx.AsyncClient wired to a fresh FastAPI app with the test DB injected."""
+    app = create_app()
+
+    async def _override_get_db() -> AsyncGenerator[AsyncSession, None]:
+        async with session_factory() as session:
+            try:
+                yield session
+            except Exception:
+                await session.rollback()
+                raise
+
+    app.dependency_overrides[get_db] = _override_get_db
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+# === Convenience: register via API (covers bootstrap, user, token, headers) ===
+
+
+@pytest_asyncio.fixture
+async def registered_user(
+    client: AsyncClient,
+) -> dict[str, Any]:
+    """Register a bootstrap user via the API.
+
+    Returns a dict with user, token, headers, email, password, name.
+    Use this fixture (or the dependent `auth_headers`) for all auth-required tests.
+    """
+    payload = {
+        "email": "admin@test.com",
+        "password": "TestPass123!",
+        "name": "Test Admin",
+    }
+    resp = await client.post("/api/v1/auth/register", json=payload)
+    assert resp.status_code == 201, f"register failed: {resp.status_code} {resp.text}"
+    data = resp.json()
+    token = data["access_token"]
+    return {
+        "user": data["user"],
+        "token": token,
+        "headers": {"Authorization": f"Bearer {token}"},
+        "email": payload["email"],
+        "password": payload["password"],
+        "name": payload["name"],
+    }
+
+
+@pytest_asyncio.fixture
+async def auth_headers(registered_user: dict[str, Any]) -> dict[str, str]:
+    """Authorization headers for the bootstrap user."""
+    return registered_user["headers"]
+
+
+@pytest_asyncio.fixture
+async def second_user(
+    client: AsyncClient,
+    registered_user: dict[str, Any],
+) -> dict[str, Any]:
+    """Register a second user (admin-only flow) and return its auth info."""
+    # Use admin headers to create another user
+    admin_headers = registered_user["headers"]
+    payload = {
+        "email": "rep@test.com",
+        "password": "RepPass123!",
+        "name": "Test Rep",
+        "role": "sales_rep",
+    }
+    resp = await client.post("/api/v1/users/", json=payload, headers=admin_headers)
+    assert resp.status_code == 201, f"create user failed: {resp.status_code} {resp.text}"
+    new_user = resp.json()
+
+    # Log in as the new user to get a token
+    login_resp = await client.post(
+        "/api/v1/auth/login",
+        data={"username": payload["email"], "password": payload["password"]},
+    )
+    assert login_resp.status_code == 200, f"login failed: {login_resp.status_code} {login_resp.text}"
+    token = login_resp.json()["access_token"]
+
+    return {
+        "user": new_user,
+        "token": token,
+        "headers": {"Authorization": f"Bearer {token}"},
+        "email": payload["email"],
+        "password": payload["password"],
+        "name": payload["name"],
+    }
+
+
+@pytest_asyncio.fixture
+async def seed_data(
+    client: AsyncClient,
+    registered_user: dict[str, Any],
+    session_factory: async_sessionmaker[AsyncSession],
+) -> dict[str, Any]:
+    """Seed a representative data set for business-logic tests.
+
+    Creates (via the API to exercise FK constraints end-to-end):
+    - 2 accounts owned by the bootstrap admin user
+    - 3 contacts (2 attached to accounts, 1 standalone)
+    - 5 deals across various stages
+    - 10 activities (some overdue, some completed)
+    - 3 tags (VIP, Strategic, Enterprise)
+    - 4 notes (mixed parents)
+
+    Returns a dict with all IDs + a ref to the auth headers for convenience.
+    """
+    headers = registered_user["headers"]
+
+    # 2 accounts
+    acc1 = await client.post(
+        "/api/v1/accounts/",
+        json={"name": "Acme Corp", "industry": "sme", "size": "sme", "website": "https://acme.test"},
+        headers=headers,
+    )
+    assert acc1.status_code == 201, f"seed acc1: {acc1.status_code} {acc1.text}"
+    acc1_id = acc1.json()["id"]
+
+    acc2 = await client.post(
+        "/api/v1/accounts/",
+        json={"name": "Globex GmbH", "industry": "enterprise", "size": "enterprise"},
+        headers=headers,
+    )
+    assert acc2.status_code == 201, f"seed acc2: {acc2.status_code} {acc2.text}"
+    acc2_id = acc2.json()["id"]
+
+    # 3 contacts (2 with account, 1 standalone)
+    c1 = await client.post(
+        "/api/v1/contacts/",
+        json={"first_name": "Anna", "last_name": "Schmidt", "email": "anna@acme.example", "account_id": acc1_id},
+        headers=headers,
+    )
+    assert c1.status_code == 201, c1.text
+    c1_id = c1.json()["id"]
+
+    c2 = await client.post(
+        "/api/v1/contacts/",
+        json={"first_name": "Bob", "last_name": "Mueller", "email": "bob@globex.example", "account_id": acc2_id},
+        headers=headers,
+    )
+    assert c2.status_code == 201, c2.text
+    c2_id = c2.json()["id"]
+
+    c3 = await client.post(
+        "/api/v1/contacts/",
+        json={"first_name": "Clara", "last_name": "Weber", "email": "clara@x.example"},
+        headers=headers,
+    )
+    assert c3.status_code == 201, c3.text
+    c3_id = c3.json()["id"]
+
+    # 5 deals (different stages)
+    deal_ids: list[int] = []
+    for i, (title, stage) in enumerate([
+        ("Deal A", "lead"),
+        ("Deal B", "qualified"),
+        ("Deal C", "proposal"),
+        ("Deal D", "negotiation"),
+        ("Deal E", "won"),
+    ]):
+        d = await client.post(
+            "/api/v1/deals/",
+            json={
+                "title": title,
+                "value": str(1000 * (i + 1)),
+                "stage": stage,
+                "account_id": acc1_id if i % 2 == 0 else acc2_id,
+            },
+            headers=headers,
+        )
+        assert d.status_code == 201, d.text
+        deal_ids.append(d.json()["id"])
+
+    # 10 activities (4 overdue, 4 future, 2 completed)
+    from datetime import UTC, datetime, timedelta
+    past = (datetime.now(UTC) - timedelta(days=2)).isoformat()
+    future = (datetime.now(UTC) + timedelta(days=2)).isoformat()
+    future2 = (datetime.now(UTC) + timedelta(days=10)).isoformat()
+    activity_ids: list[int] = []
+    for i in range(10):
+        if i < 4:
+            due = past
+            body_data: dict[str, object] = {"type": "task", "subject": f"Overdue task {i}", "due_date": due, "deal_id": deal_ids[i % 5]}
+        elif i < 8:
+            due = future if i % 2 == 0 else future2
+            body_data = {"type": "call", "subject": f"Upcoming call {i}", "due_date": due, "account_id": acc1_id}
+        else:
+            body_data = {"type": "meeting", "subject": f"Done meeting {i}", "account_id": acc1_id}
+        a = await client.post("/api/v1/activities/", json=body_data, headers=headers)
+        assert a.status_code == 201, a.text
+        activity_ids.append(a.json()["id"])
+
+    # 3 tags
+    tag_ids: list[int] = []
+    for name in ["VIP", "Strategic", "Enterprise"]:
+        t = await client.post("/api/v1/tags/", json={"name": name, "color": "#FF0080"}, headers=headers)
+        assert t.status_code == 201, t.text
+        tag_ids.append(t.json()["id"])
+
+    # 4 notes (mixed parents: 2 account, 1 contact, 1 deal)
+    n1 = await client.post("/api/v1/notes/", json={"body": "Note on Acme", "parent_type": "account", "parent_id": acc1_id}, headers=headers)
+    assert n1.status_code == 201, n1.text
+    n2 = await client.post("/api/v1/notes/", json={"body": "Note on Globex", "parent_type": "account", "parent_id": acc2_id}, headers=headers)
+    assert n2.status_code == 201, n2.text
+    n3 = await client.post("/api/v1/notes/", json={"body": "Note on Anna", "parent_type": "contact", "parent_id": c1_id}, headers=headers)
+    assert n3.status_code == 201, n3.text
+    n4 = await client.post("/api/v1/notes/", json={"body": "Note on Deal A", "parent_type": "deal", "parent_id": deal_ids[0]}, headers=headers)
+    assert n4.status_code == 201, n4.text
+
+    return {
+        "account_ids": [acc1_id, acc2_id],
+        "contact_ids": [c1_id, c2_id, c3_id],
+        "deal_ids": deal_ids,
+        "activity_ids": activity_ids,
+        "tag_ids": tag_ids,
+        "note_ids": [n1.json()["id"], n2.json()["id"], n3.json()["id"], n4.json()["id"]],
+        "headers": headers,
+    }