Upload tests/test_deals.py

tests/test_deals.py

@@ -0,0 +1,88 @@
+"""Tests for FR-4 (Deal entity) and stage-history audit."""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_create_deal(
+    client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
+) -> None:
+    acc_id = seed_data["account_ids"][0]
+    resp = await client.post(
+        "/api/v1/deals/",
+        json={"title": "Big Deal", "value": "5000.00", "stage": "qualified", "account_id": acc_id},
+        headers=auth_headers,
+    )
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["title"] == "Big Deal"
+    assert body["stage"] == "qualified"
+
+
+@pytest.mark.asyncio
+async def test_update_deal_stage_creates_history(
+    client: AsyncClient, seed_data: dict, session_factory
+) -> None:
+    deal_id = seed_data["deal_ids"][0]
+    resp = await client.patch(
+        f"/api/v1/deals/{deal_id}/stage",
+        json={"stage": "qualified"},
+        headers=seed_data["headers"],
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["stage"] == "qualified"
+
+    # Verify DealStageHistory row exists in DB
+    from sqlalchemy import text
+    async with session_factory() as session:
+        result = await session.execute(
+            text("SELECT to_stage FROM deal_stage_history WHERE deal_id = :did ORDER BY id"),
+            {"did": deal_id},
+        )
+        stages = [r[0] for r in result.fetchall()]
+    assert "qualified" in stages
+
+
+@pytest.mark.asyncio
+async def test_get_pipeline_grouped_by_stage(client: AsyncClient, seed_data: dict) -> None:
+    resp = await client.get("/api/v1/deals/pipeline", headers=seed_data["headers"])
+    assert resp.status_code == 200
+    body = resp.json()
+    # Each entry has stage, count, total_value
+    assert all({"stage", "count", "total_value"} <= set(item.keys()) for item in body)
+    # Seeded deals cover at least 5 distinct stages
+    stages = {item["stage"] for item in body}
+    assert len(stages) >= 3
+
+
+@pytest.mark.asyncio
+async def test_filter_deals_by_owner(client: AsyncClient, seed_data: dict) -> None:
+    owner_id = seed_data["headers"].get("X-Test-User-Id")  # may be None; fallback: just check 200
+    # we filter by an owner-id that does not exist; expect empty list
+    resp = await client.get("/api/v1/deals/?owner_id=99999", headers=seed_data["headers"])
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body == [] or all(d.get("owner_id") == 99999 for d in body)
+
+
+@pytest.mark.asyncio
+async def test_db_write_deal(
+    client: AsyncClient, auth_headers: dict[str, str], seed_data: dict, session_factory
+) -> None:
+    """DB roundtrip: read back a seeded deal directly via SQL."""
+    from sqlalchemy import text
+    deal_id = seed_data["deal_ids"][2]
+    async with session_factory() as session:
+        result = await session.execute(
+            text("SELECT title, value, stage, account_id FROM deals WHERE id = :did"),
+            {"did": deal_id},
+        )
+        row = result.fetchone()
+    assert row is not None
+    assert row[0] is not None  # title
+    assert row[1] is not None  # value
+    assert row[2] is not None  # stage
+    assert row[3] is not None  # account_id