oliver-sales-ops-platform/backend/tests/conftest.py
DJP 5ad6d01846 Add pytest integration test harness for backend
Test harness hits the live backend on http://localhost:8003 (override via
OSOP_BASE_URL) rather than spinning an in-process FastAPI app, so we test
the actual deployed surface. 33 tests pass, 1 intentionally skipped
(real Anthropic-backed intake — gated behind @pytest.mark.requires_anthropic
to avoid spending money on every run).

Coverage:
- Opportunity CRUD: create with defaults + invalid model_type, all 5
  model_type values round-trip, list/get/update, full create-delete-404.
- Stage machine: 17 rows initialised, advance unlocks next, double-complete
  rejected, out-of-order rejected, out-of-range rejected, notes persist.
- Stage gating: stage 3 without approvals returns 400 with "approval"/
  "gated" in the detail and does not mutate state.
- File upload: .txt, .md, synthesized .docx, synthesized .xlsx all
  upload + extract; .exe rejected; list/delete round-trip.
- Schema sanity: list endpoints return the expected shape.

Re-run:
  python3 -m venv /tmp/osop_test_venv && \
    /tmp/osop_test_venv/bin/pip install -q -r backend/requirements-dev.txt && \
    cd backend && /tmp/osop_test_venv/bin/pytest tests/ -v

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-27 13:22:32 -04:00

93 lines
2.8 KiB
Python

"""Shared pytest fixtures for the OSOP integration test suite.
These tests run against a LIVE backend (default http://localhost:8003) — they
do NOT spin up an in-process FastAPI app, so we exercise the same code paths
as production.
Override the base URL with the OSOP_BASE_URL environment variable.
"""
from __future__ import annotations
import os
from datetime import date
from typing import AsyncIterator
import httpx
import pytest
import pytest_asyncio
BASE_URL = os.environ.get("OSOP_BASE_URL", "http://localhost:8003").rstrip("/")
API = f"{BASE_URL}/api"
def pytest_configure(config: pytest.Config) -> None:
config.addinivalue_line(
"markers",
"requires_anthropic: marks tests that perform real Anthropic API calls (skipped by default)",
)
@pytest_asyncio.fixture
async def client() -> AsyncIterator[httpx.AsyncClient]:
"""Async HTTP client pointed at the running backend."""
async with httpx.AsyncClient(base_url=API, timeout=30.0) as c:
yield c
def pytest_sessionstart(session: pytest.Session) -> None:
"""Fail fast (sync) if the backend isn't reachable. No event-loop dance."""
import urllib.request
import urllib.error
try:
with urllib.request.urlopen(f"{API}/health", timeout=5) as r:
if r.status != 200:
raise RuntimeError(f"health returned {r.status}")
except Exception as e:
pytest.exit(
f"Backend is not reachable at {API}/health: {e}\n"
"Bring it up with `docker compose up -d` (or set OSOP_BASE_URL).",
returncode=2,
)
def _opportunity_payload(name_suffix: str = "") -> dict:
return {
"name": f"PYTEST Harness Opp {name_suffix}".strip(),
"client_name": "Test Client Ltd",
"region": "EMEA",
"brands": "BrandA, BrandB",
"service_types": "Content Production, Social",
"description": "Synthetic opportunity created by the integration test suite.",
"deadline": date(2027, 1, 15).isoformat(),
"go_live": date(2027, 3, 1).isoformat(),
"model_type": "ai_oplus",
}
@pytest_asyncio.fixture
async def opportunity(client: httpx.AsyncClient) -> AsyncIterator[dict]:
"""Create a fresh opportunity, yield it, then delete it on teardown.
Cleanup runs even if the test fails so we don't leak rows.
"""
payload = _opportunity_payload(name_suffix="(fixture)")
r = await client.post("/opportunities", json=payload)
r.raise_for_status()
opp = r.json()
try:
yield opp
finally:
try:
await client.delete(f"/opportunities/{opp['id']}")
except Exception:
# Best-effort cleanup — never fail teardown.
pass
@pytest.fixture
def opp_payload_factory():
"""Returns a fresh payload dict each call so tests can mutate it."""
return _opportunity_payload