impactflow_discovery/tests/test_auth.py

"""Tests for the auth module: dual-auth dependency, JWT issue/decode,
domain allow-list. The `app_client` fixture (tests/conftest.py) provides an
isolated app + temp DB so these never touch the real database.
"""


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_cookie_access_token_authenticates(app_client):
    """A valid access token in the httpOnly cookie authenticates the browser."""
    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-cookie",
            email="cookie@example.com",
            display_name="Cookie User",
            google_id="g-cookie",
            role="user",
            created_at=datetime.now(timezone.utc),
        ))
        await db.commit()

    token = create_access_token("u-cookie", "cookie@example.com")
    r = await app_client.get("/api/me", cookies={"access_token": token})
    assert r.status_code == 200
    assert r.json()["email"] == "cookie@example.com"


async def test_refresh_via_cookie_sets_new_access_cookie(app_client):
    """POST /api/auth/refresh with only the refresh cookie mints a new access
    token and writes it back as a cookie (no JSON body required)."""
    from datetime import datetime, timezone
    from app.auth import issue_refresh_token
    from app.database import AsyncSessionLocal
    from app.models import User

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

    r = await app_client.post(
        "/api/auth/refresh", cookies={"refresh_token": raw_refresh}
    )
    assert r.status_code == 200
    assert r.json()["access_token"]
    # The response re-sets the access cookie, and it actually authenticates.
    new_access = r.cookies.get("access_token")
    assert new_access
    me = await app_client.get("/api/me", cookies={"access_token": new_access})
    assert me.status_code == 200
    assert me.json()["email"] == "refresh@example.com"


async def test_logout_revokes_refresh_token(app_client):
    """Logout via the refresh cookie revokes it, so a later refresh fails."""
    from datetime import datetime, timezone
    from app.auth import issue_refresh_token
    from app.database import AsyncSessionLocal
    from app.models import User

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

    out = await app_client.post(
        "/api/auth/logout", cookies={"refresh_token": raw_refresh}
    )
    assert out.status_code == 200

    again = await app_client.post(
        "/api/auth/refresh", cookies={"refresh_token": raw_refresh}
    )
    assert again.status_code == 401


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