impactflow_discovery/tests/test_auth.py

"""Tests for the auth module: dual-auth dependency, JWT issue/decode,
domain allow-list. Uses a temp SQLite DB so it doesn't touch the real one.
"""
import os

import pytest
from httpx import ASGITransport, AsyncClient


@pytest.fixture
async def app_client(tmp_path, monkeypatch):
    """Spin up a fresh app with an isolated DB and known auth secrets."""
    db_path = tmp_path / "auth_test.db"
    monkeypatch.setenv(
        "DATABASE_URL", f"sqlite+aiosqlite:///{db_path}"
    )
    monkeypatch.setenv("JWT_SECRET", "test-jwt-secret")
    monkeypatch.setenv("IMPACTFLOW_API_KEY", "test-api-key")
    monkeypatch.setenv("GOOGLE_CLIENT_ID", "fake-client-id")
    monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "fake-client-secret")
    monkeypatch.setenv("ALLOWED_EMAIL_DOMAINS", "")

    # Re-import in a way that picks up the patched env. The simplest way is
    # to clear modules that read env at import time.
    import importlib
    import sys

    for mod in list(sys.modules):
        if mod.startswith("app"):
            del sys.modules[mod]

    from app import database
    importlib.reload(database)
    from app.database import Base, engine

    from app.main import app

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    transport = ASGITransport(app=app)
    async with AsyncClient(
        transport=transport, base_url="http://test"
    ) as client:
        yield client


async def test_unauthenticated_request_returns_401(app_client):
    r = await app_client.get("/api/me")
    assert r.status_code == 401
    assert r.json()["detail"] == "Not authenticated"


async def test_api_key_resolves_to_admin(app_client):
    r = await app_client.get(
        "/api/me", headers={"X-API-Key": "test-api-key"}
    )
    assert r.status_code == 200
    body = r.json()
    assert body["role"] == "admin"
    assert body["id"] == "api-key-admin"


async def test_wrong_api_key_is_rejected(app_client):
    r = await app_client.get(
        "/api/me", headers={"X-API-Key": "wrong-key"}
    )
    assert r.status_code == 401


async def test_bogus_bearer_is_rejected(app_client):
    r = await app_client.get(
        "/api/me", headers={"Authorization": "Bearer not-a-jwt"}
    )
    assert r.status_code == 401


async def test_valid_jwt_authenticates(app_client, tmp_path):
    """Mint a JWT for a user we insert directly into the DB."""
    from datetime import datetime, timezone
    from app.auth import create_access_token
    from app.database import AsyncSessionLocal
    from app.models import User

    async with AsyncSessionLocal() as db:
        user = User(
            id="u-1",
            email="real@example.com",
            display_name="Real User",
            google_id="g-1",
            role="user",
            created_at=datetime.now(timezone.utc),
        )
        db.add(user)
        await db.commit()

    token = create_access_token("u-1", "real@example.com")
    r = await app_client.get(
        "/api/me", headers={"Authorization": f"Bearer {token}"}
    )
    assert r.status_code == 200
    assert r.json()["email"] == "real@example.com"
    assert r.json()["role"] == "user"


async def test_oauth_login_redirects_to_google(app_client):
    r = await app_client.get(
        "/api/auth/login", follow_redirects=False
    )
    assert r.status_code in (302, 303, 307)
    assert "accounts.google.com" in r.headers["location"]


async def test_admin_only_endpoint_requires_admin(app_client):
    """Regular users get 403 on /api/admin/activity."""
    from datetime import datetime, timezone
    from app.auth import create_access_token
    from app.database import AsyncSessionLocal
    from app.models import User

    async with AsyncSessionLocal() as db:
        db.add(User(
            id="u-regular",
            email="reg@example.com",
            display_name="Reg",
            google_id="g-reg",
            role="user",
            created_at=datetime.now(timezone.utc),
        ))
        await db.commit()
    token = create_access_token("u-regular", "reg@example.com")
    r = await app_client.get(
        "/api/admin/activity",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r.status_code == 403

    # API-key admin should be allowed.
    r = await app_client.get(
        "/api/admin/activity",
        headers={"X-API-Key": "test-api-key"},
    )
    assert r.status_code == 200


def test_email_domain_allowlist(monkeypatch):
    from app.auth import email_domain_allowed

    monkeypatch.setenv("ALLOWED_EMAIL_DOMAINS", "")
    assert email_domain_allowed("anyone@example.com")

    monkeypatch.setenv("ALLOWED_EMAIL_DOMAINS", "computerim.com")
    assert email_domain_allowed("j@computerim.com")
    assert not email_domain_allowed("j@gmail.com")

    monkeypatch.setenv("ALLOWED_EMAIL_DOMAINS", "a.com, b.com")
    assert email_domain_allowed("x@a.com")
    assert email_domain_allowed("y@B.COM")
    assert not email_domain_allowed("z@c.com")