Ken
164 lines
5.7 KiB
Python
164 lines
5.7 KiB
Python
"""Tests for FastAPI app — health, auth, artifacts, and MCP apps endpoints."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import bcrypt
|
|
|
|
# Set env vars BEFORE importing app (auth module reads at import time)
|
|
os.environ["AUTH_USER"] = "test@example.com"
|
|
os.environ["AUTH_PASS_HASH"] = bcrypt.hashpw(b"testpass", bcrypt.gensalt()).decode()
|
|
|
|
# Add backend to sys.path so we can import app
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import app # noqa: E402 — must come after env vars and sys.path setup
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
"""Fresh unauthenticated TestClient for each test."""
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_client():
|
|
"""Authenticated TestClient — logs in before yielding."""
|
|
with TestClient(app) as c:
|
|
resp = c.post(
|
|
"/api/auth/login",
|
|
json={"email": "test@example.com", "password": "testpass"},
|
|
)
|
|
assert resp.status_code == 200, f"Login failed in fixture: {resp.text}"
|
|
yield c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestHealthEndpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHealthEndpoint:
|
|
def test_health_returns_200_with_ok(self, client):
|
|
"""GET /api/health returns 200 with {status: 'ok'}."""
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "ok"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestAuthEndpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAuthEndpoints:
|
|
def test_login_success_returns_token_and_cookie(self, client):
|
|
"""Successful login returns 200 with token, email, and sets session cookie."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"email": "test@example.com", "password": "testpass"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "token" in data
|
|
assert "email" in data
|
|
assert data["email"] == "test@example.com"
|
|
assert "session" in client.cookies
|
|
|
|
def test_login_wrong_password_returns_401(self, client):
|
|
"""Wrong password → 401."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"email": "test@example.com", "password": "wrongpassword"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
def test_login_wrong_email_returns_401(self, client):
|
|
"""Wrong email → 401."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={"email": "wrong@example.com", "password": "testpass"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
def test_me_unauthenticated_returns_401(self, client):
|
|
"""GET /api/auth/me without session cookie → 401."""
|
|
response = client.get("/api/auth/me")
|
|
assert response.status_code == 401
|
|
|
|
def test_me_authenticated_returns_email(self, auth_client):
|
|
"""GET /api/auth/me with valid session → 200 with email."""
|
|
response = auth_client.get("/api/auth/me")
|
|
assert response.status_code == 200
|
|
assert response.json()["email"] == "test@example.com"
|
|
|
|
def test_logout_invalidates_session(self, client):
|
|
"""POST /api/auth/logout invalidates the session token."""
|
|
# Login to get a session
|
|
login = client.post(
|
|
"/api/auth/login",
|
|
json={"email": "test@example.com", "password": "testpass"},
|
|
)
|
|
assert login.status_code == 200
|
|
|
|
# Verify me works before logout
|
|
me = client.get("/api/auth/me")
|
|
assert me.status_code == 200
|
|
|
|
# Logout
|
|
logout = client.post("/api/auth/logout")
|
|
assert logout.status_code == 200
|
|
|
|
# Me should now return 401 (token invalidated, cookie cleared)
|
|
me_after = client.get("/api/auth/me")
|
|
assert me_after.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestArtifactEndpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestArtifactEndpoints:
|
|
def test_list_artifacts_unauthenticated_returns_401(self, client):
|
|
"""GET /api/artifacts/{session_id} without auth → 401."""
|
|
response = client.get("/api/artifacts/some-session-id")
|
|
assert response.status_code == 401
|
|
|
|
def test_list_artifacts_empty_when_session_has_none(
|
|
self, auth_client, monkeypatch, tmp_path
|
|
):
|
|
"""Authenticated list for nonexistent session → 200 with empty list."""
|
|
import artifacts
|
|
|
|
monkeypatch.setattr(artifacts, "ARTIFACTS_DIR", tmp_path)
|
|
|
|
response = auth_client.get("/api/artifacts/nonexistent-session")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestMCPAppsEndpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMCPAppsEndpoint:
|
|
def test_get_app_unauthenticated_returns_401(self, client):
|
|
"""GET /api/apps/{name} without auth → 401."""
|
|
response = client.get("/api/apps/nonexistent")
|
|
assert response.status_code == 401
|
|
|
|
def test_get_nonexistent_app_authenticated_returns_404(self, auth_client):
|
|
"""GET /api/apps/{name} for missing app file → 404."""
|
|
response = auth_client.get("/api/apps/nonexistent-app-xyz")
|
|
assert response.status_code == 404
|