research-workbench/tests/test_app.py

"""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


# ---------------------------------------------------------------------------
# TestCopilotKitAuth
# ---------------------------------------------------------------------------

# NOTE: /api/copilotkit is not fully tested here because it requires a live
# amplifierd stream. See integration tests (requires container runtime).
# Auth enforcement is tested below.


class TestCopilotKitAuth:
    def test_copilotkit_unauthenticated_returns_401(self, client):
        """POST /api/copilotkit without session cookie → 401."""
        response = client.post(
            "/api/copilotkit",
            json={"threadId": "test-thread", "messages": []},
        )
        assert response.status_code == 401