agnes-the-ai-analyst/tests/test_auth_providers.py

"""Tests for auth providers — password, email magic link, google OAuth."""

import os
import pytest
from fastapi.testclient import TestClient


@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setenv("DATA_DIR", str(tmp_path))
    monkeypatch.setenv("JWT_SECRET_KEY", "test-secret-32chars-minimum!!!!!")

    from app.main import create_app
    from src.db import get_system_db
    from src.repositories.users import UserRepository

    conn = get_system_db()
    ur = UserRepository(conn)
    # User with password
    try:
        from argon2 import PasswordHasher
        ph = PasswordHasher()
        pw_hash = ph.hash("testpass123")
    except ImportError:
        import hashlib
        pw_hash = hashlib.sha256(b"testpass123").hexdigest()

    ur.create(id="pw1", email="pw@test.com", name="PW User", role="analyst", password_hash=pw_hash)
    # User with setup token (and fresh created timestamp so the JSON /setup
    # endpoint's TTL check accepts it)
    from datetime import datetime, timezone
    ur.create(id="setup1", email="setup@test.com", name="Setup User", role="analyst")
    ur.update(id="setup1", setup_token="setup-token-123",
              setup_token_created=datetime.now(timezone.utc))
    # User for magic link
    ur.create(id="ml1", email="ml@test.com", name="ML User", role="analyst")
    conn.close()

    app = create_app()
    return TestClient(app)


class TestTokenEndpoint:
    """Tests for /auth/token — password bypass fix."""

    def test_token_empty_password_rejected_when_user_has_hash(self, client):
        """Empty password must be rejected when user has password_hash."""
        resp = client.post("/auth/token", json={"email": "pw@test.com", "password": ""})
        assert resp.status_code == 401

    def test_token_missing_password_rejected_when_user_has_hash(self, client):
        """Omitting password field (defaults to '') must be rejected when user has password_hash."""
        resp = client.post("/auth/token", json={"email": "pw@test.com"})
        assert resp.status_code == 401

    def test_token_wrong_password_rejected(self, client):
        """Wrong password must be rejected with 401."""
        resp = client.post("/auth/token", json={"email": "pw@test.com", "password": "wrongpass"})
        assert resp.status_code == 401

    def test_token_correct_password_succeeds(self, client):
        """Correct password must issue a token."""
        resp = client.post("/auth/token", json={"email": "pw@test.com", "password": "testpass123"})
        assert resp.status_code == 200
        data = resp.json()
        assert "access_token" in data
        assert data["email"] == "pw@test.com"

    def test_token_no_password_hash_user_gets_token(self, client):
        """User without password_hash (OAuth-only) must be rejected at /auth/token."""
        resp = client.post("/auth/token", json={"email": "ml@test.com"})
        assert resp.status_code == 401

    def test_token_rejected_for_oauth_only_user(self, client):
        """OAuth-only user (no password_hash) must not receive a token via /auth/token."""
        resp = client.post("/auth/token", json={"email": "ml@test.com"})
        assert resp.status_code == 401
        assert "external authentication" in resp.json()["detail"]

    def test_token_unknown_user_rejected(self, client):
        """Unknown email must return 401."""
        resp = client.post("/auth/token", json={"email": "nobody@test.com", "password": "anything"})
        assert resp.status_code == 401


class TestPasswordAuth:
    def test_login_success(self, client):
        resp = client.post("/auth/password/login", json={
            "email": "pw@test.com", "password": "testpass123",
        })
        assert resp.status_code == 200
        assert "access_token" in resp.json()

    def test_login_wrong_password(self, client):
        resp = client.post("/auth/password/login", json={
            "email": "pw@test.com", "password": "wrongpass",
        })
        assert resp.status_code == 401

    def test_login_unknown_user(self, client):
        resp = client.post("/auth/password/login", json={
            "email": "unknown@test.com", "password": "test",
        })
        assert resp.status_code == 401

    def test_setup_password(self, client):
        resp = client.post("/auth/password/setup", json={
            "email": "setup@test.com", "token": "setup-token-123", "password": "newpass456",
        })
        assert resp.status_code == 200
        assert "access_token" in resp.json()

    def test_setup_wrong_token(self, client):
        resp = client.post("/auth/password/setup", json={
            "email": "setup@test.com", "token": "wrong-token", "password": "newpass",
        })
        assert resp.status_code == 400


class TestEmailAuth:
    def test_send_link_registered(self, client):
        resp = client.post("/auth/email/send-link", json={"email": "ml@test.com"})
        assert resp.status_code == 200
        # Always returns same message (anti-enumeration)
        assert "If this email" in resp.json()["message"]

    def test_send_link_unregistered(self, client):
        resp = client.post("/auth/email/send-link", json={"email": "nobody@test.com"})
        assert resp.status_code == 200
        assert "If this email" in resp.json()["message"]

    def test_verify_invalid_token(self, client):
        resp = client.post("/auth/email/verify", json={
            "email": "ml@test.com", "token": "invalid",
        })
        assert resp.status_code == 401


class TestGoogleOAuth:
    def test_google_login_not_configured(self, client):
        """Without GOOGLE_CLIENT_ID, should redirect to login with error."""
        resp = client.get("/auth/google/login", follow_redirects=False)
        assert resp.status_code == 302 or resp.status_code == 307
        assert "error" in resp.headers.get("location", "")


class TestGoogleGroupsFetch:
    """Unit tests for _fetch_google_groups — the helper must be tolerant of
    every realistic failure mode (non-Workspace tenants return 403, expired
    tokens return 401, network errors bubble from httpx) and never raise."""

    def test_parses_groups_from_success_response(self, monkeypatch):
        import asyncio
        from app.auth.providers import google as gp


        # searchTransitiveGroups returns {"memberships": [...]}, not {"groups": [...]}.
        # Each item carries the group identity in groupKey.id + displayName,
        # matching the actual API response shape.
        fake_payload = {
            "memberships": [
                {
                    "group": "groups/abc123",
                    "groupKey": {"id": "team-eng@example.com"},
                    "displayName": "Engineering",
                },
                {
                    "group": "groups/def456",
                    "groupKey": {"id": "everyone@example.com"},
                    # No displayName — falls back to id
                },
            ],
        }

        class _Resp:
            status_code = 200
            text = ""
            def json(self):
                return fake_payload

        class _FakeClient:
            def __init__(self, *a, **kw):
                pass
            async def __aenter__(self):
                return self
            async def __aexit__(self, *a):
                return False
            async def get(self, url, params=None, headers=None):
                return _Resp()

        monkeypatch.setattr(gp.httpx, "AsyncClient", _FakeClient)

        groups = asyncio.run(gp._fetch_google_groups("fake-token", "user@example.com"))
        assert groups == [
            {"id": "team-eng@example.com", "name": "Engineering"},
            {"id": "everyone@example.com", "name": "everyone@example.com"},
        ]

    def test_returns_empty_on_403(self, monkeypatch):
        """Cloud Identity not enabled (non-Workspace tenant) → 403 → [] + warning."""
        import asyncio
        from app.auth.providers import google as gp

        class _Resp:
            status_code = 403
            text = "Cloud Identity API has not been enabled"

        class _FakeClient:
            def __init__(self, *a, **kw): pass
            async def __aenter__(self): return self
            async def __aexit__(self, *a): return False
            async def get(self, url, params=None, headers=None):
                return _Resp()

        monkeypatch.setattr(gp.httpx, "AsyncClient", _FakeClient)

        groups = asyncio.run(gp._fetch_google_groups("fake-token", "user@example.com"))
        assert groups == []

    def test_returns_empty_on_exception(self, monkeypatch):
        """Network error inside httpx must be swallowed, not propagated."""
        import asyncio
        from app.auth.providers import google as gp

        class _FakeClient:
            def __init__(self, *a, **kw): pass
            async def __aenter__(self): return self
            async def __aexit__(self, *a): return False
            async def get(self, *a, **kw):
                raise RuntimeError("boom")

        monkeypatch.setattr(gp.httpx, "AsyncClient", _FakeClient)

        groups = asyncio.run(gp._fetch_google_groups("fake-token", "user@example.com"))
        assert groups == []


class TestLocalDevGroupsParser:
    """Unit tests for get_local_dev_groups() — must tolerate every malformed
    input shape (typos, wrong type, missing id) and never raise. Bad input
    becomes [] + a WARNING log so the dev mock can't break the dev flow."""

    def test_returns_empty_when_unset(self, monkeypatch):
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.delenv("LOCAL_DEV_GROUPS", raising=False)
        assert get_local_dev_groups() == []

    def test_returns_empty_when_blank(self, monkeypatch):
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv("LOCAL_DEV_GROUPS", "   ")
        assert get_local_dev_groups() == []

    def test_parses_valid_json_array(self, monkeypatch):
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv(
            "LOCAL_DEV_GROUPS",
            '[{"id":"eng@x.com","name":"Engineering"},'
            '{"id":"admins@x.com","name":"Admins"}]',
        )
        assert get_local_dev_groups() == [
            {"id": "eng@x.com", "name": "Engineering"},
            {"id": "admins@x.com", "name": "Admins"},
        ]

    def test_defaults_name_to_id(self, monkeypatch):
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv("LOCAL_DEV_GROUPS", '[{"id":"eng@x.com"}]')
        assert get_local_dev_groups() == [{"id": "eng@x.com", "name": "eng@x.com"}]

    def test_preserves_extra_fields(self, monkeypatch):
        """Forward-compat: unknown fields like roles/labels survive parsing
        so future group-aware code can be exercised in dev without parser changes."""
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv(
            "LOCAL_DEV_GROUPS",
            '[{"id":"eng@x.com","name":"Eng","roles":["MEMBER","OWNER"]}]',
        )
        result = get_local_dev_groups()
        assert result == [
            {"id": "eng@x.com", "name": "Eng", "roles": ["MEMBER", "OWNER"]},
        ]

    def test_returns_empty_on_invalid_json(self, monkeypatch):
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv("LOCAL_DEV_GROUPS", "not-json,foo")
        assert get_local_dev_groups() == []

    def test_returns_empty_on_non_list(self, monkeypatch):
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv("LOCAL_DEV_GROUPS", '{"id":"eng@x.com"}')
        assert get_local_dev_groups() == []

    def test_skips_items_without_id(self, monkeypatch):
        """Bad items are dropped, valid siblings survive — partial config
        still produces something useful instead of nuking the whole list."""
        from app.auth.dependencies import get_local_dev_groups
        monkeypatch.setenv(
            "LOCAL_DEV_GROUPS",
            '[{"name":"no-id"},{"id":"eng@x.com","name":"Eng"},"string-not-object"]',
        )
        assert get_local_dev_groups() == [{"id": "eng@x.com", "name": "Eng"}]


class TestLocalDevGroupsInjection:
    """End-to-end: with LOCAL_DEV_MODE=1 + LOCAL_DEV_GROUPS, the seeded dev
    user's session.google_groups gets populated on first authenticated request
    so /profile renders the mocked groups."""

    @pytest.fixture
    def dev_client(self, tmp_path, monkeypatch):
        monkeypatch.setenv("DATA_DIR", str(tmp_path))
        monkeypatch.setenv("JWT_SECRET_KEY", "test-secret-32chars-minimum!!!!!")
        monkeypatch.setenv("SESSION_SECRET", "test-session-secret-32chars-minimum!!")
        monkeypatch.setenv("LOCAL_DEV_MODE", "1")
        monkeypatch.setenv("LOCAL_DEV_USER_EMAIL", "dev@localhost")
        monkeypatch.setenv(
            "LOCAL_DEV_GROUPS",
            '[{"id":"local-dev-engineers@example.com","name":"Local Dev Engineers"}]',
        )
        from app.main import create_app
        return TestClient(create_app())

    def test_dev_user_sees_mocked_groups_on_profile(self, dev_client):
        resp = dev_client.get("/profile")
        assert resp.status_code == 200
        body = resp.text
        assert "local-dev-engineers@example.com" in body
        assert "Local Dev Engineers" in body
        assert "No Google groups available" not in body

    def test_empty_LOCAL_DEV_GROUPS_falls_back_to_empty_state(
        self, tmp_path, monkeypatch
    ):
        monkeypatch.setenv("DATA_DIR", str(tmp_path))
        monkeypatch.setenv("JWT_SECRET_KEY", "test-secret-32chars-minimum!!!!!")
        monkeypatch.setenv("LOCAL_DEV_MODE", "1")
        monkeypatch.delenv("LOCAL_DEV_GROUPS", raising=False)
        from app.main import create_app
        client = TestClient(create_app())
        resp = client.get("/profile")
        assert resp.status_code == 200
        assert "No Google groups available" in resp.text


class TestLocalDevGroupsStartupValidation:
    """Startup banner reports on LOCAL_DEV_GROUPS so a typo or malformed JSON
    is loud at boot, not silent until the first authenticated request."""

    def _capture_startup_logs(self, tmp_path, monkeypatch, caplog, env_value):
        import logging
        monkeypatch.setenv("DATA_DIR", str(tmp_path))
        monkeypatch.setenv("JWT_SECRET_KEY", "test-secret-32chars-minimum!!!!!")
        monkeypatch.setenv("LOCAL_DEV_MODE", "1")
        if env_value is None:
            monkeypatch.delenv("LOCAL_DEV_GROUPS", raising=False)
        else:
            monkeypatch.setenv("LOCAL_DEV_GROUPS", env_value)
        from app.main import create_app
        with caplog.at_level(logging.WARNING, logger="app.main"):
            create_app()
        return caplog.text

    def test_logs_count_and_ids_on_valid_input(self, tmp_path, monkeypatch, caplog):
        text = self._capture_startup_logs(
            tmp_path, monkeypatch, caplog,
            '[{"id":"a@x.com","name":"A"},{"id":"b@x.com","name":"B"}]',
        )
        assert "mocking 2 group(s)" in text
        assert "a@x.com" in text
        assert "b@x.com" in text

    def test_warns_when_set_but_malformed(self, tmp_path, monkeypatch, caplog):
        text = self._capture_startup_logs(
            tmp_path, monkeypatch, caplog, "not-valid-json",
        )
        assert "produced no valid groups" in text

    def test_logs_unset_explicitly(self, tmp_path, monkeypatch, caplog):
        text = self._capture_startup_logs(tmp_path, monkeypatch, caplog, None)
        assert "LOCAL_DEV_GROUPS is unset" in text


class TestCookieAuth:
    def test_web_ui_with_cookie(self, client):
        """Test that web UI routes accept JWT from cookie."""
        from app.auth.jwt import create_access_token
        from src.db import get_system_db
        from src.repositories.users import UserRepository

        conn = get_system_db()
        ur = UserRepository(conn)
        # Use existing user
        user = ur.get_by_email("pw@test.com")
        conn.close()

        token = create_access_token(user["id"], user["email"], user["role"])
        # Set cookie and access dashboard
        client.cookies.set("access_token", token)
        resp = client.get("/dashboard")
        # Should not be 401 — cookie auth works
        assert resp.status_code != 401