learningcircuit--local-deep-research/tests/auth_tests/test_auth_integration.py

"""
Integration tests for the complete authentication system.
"""

import os
import shutil
import tempfile
from pathlib import Path

import pytest

from local_deep_research.database.auth_db import (
    get_auth_db_session,
    init_auth_database,
)
from local_deep_research.database.models.auth import User
from local_deep_research.web.app_factory import create_app


@pytest.fixture
def temp_data_dir():
    """Create a temporary data directory for testing."""
    temp_dir = tempfile.mkdtemp()
    yield Path(temp_dir)
    shutil.rmtree(temp_dir)


@pytest.fixture
def app(temp_data_dir, monkeypatch):
    """Create a Flask app configured for testing."""
    monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))

    # Clear database manager state before creating app
    from local_deep_research.database.encrypted_db import db_manager

    db_manager.close_all_databases()

    # Reset db_manager's data directory to temp directory
    db_manager.data_dir = temp_data_dir / "encrypted_databases"
    db_manager.data_dir.mkdir(parents=True, exist_ok=True)

    # Remove any existing user databases in the temp directory
    encrypted_db_dir = temp_data_dir / "encrypted_databases"
    if encrypted_db_dir.exists():
        shutil.rmtree(encrypted_db_dir)
        encrypted_db_dir.mkdir(parents=True, exist_ok=True)

    # Initialize fresh auth database
    init_auth_database()

    # Clean up any existing test users before starting
    auth_db = get_auth_db_session()
    auth_db.query(User).filter(User.username.like("integrationtest%")).delete()
    auth_db.query(User).filter(User.username.like("testuser%")).delete()
    auth_db.commit()
    auth_db.close()

    app, _ = create_app()
    app.config["TESTING"] = True
    app.config["WTF_CSRF_ENABLED"] = False
    app.config["SESSION_COOKIE_SECURE"] = False

    yield app

    # Cleanup after test
    db_manager.close_all_databases()

    # Clean up test users from auth database
    auth_db = get_auth_db_session()
    auth_db.query(User).filter(User.username.like("integrationtest%")).delete()
    auth_db.query(User).filter(User.username.like("testuser%")).delete()
    auth_db.commit()
    auth_db.close()


@pytest.fixture
def client(app):
    """Create a test client."""
    return app.test_client()


class TestAuthIntegration:
    """Integration tests for authentication system."""

    def test_full_user_lifecycle(self, client):
        """Test complete user lifecycle: register, login, use app, logout."""
        # 1. Start unauthenticated - should redirect to login
        response = client.get("/")
        assert response.status_code == 302
        assert "/auth/login" in response.location

        # 2. Register new user
        response = client.post(
            "/auth/register",
            data={
                "username": "integrationtest",
                "password": "TestPass123",  # pragma: allowlist secret
                "confirm_password": "TestPass123",  # pragma: allowlist secret
                "acknowledge": "true",
            },
            follow_redirects=True,
        )
        assert response.status_code == 200

        # 3. Should be logged in automatically after registration
        response = client.get("/")
        assert response.status_code == 200
        # The main page should show we're logged in (username might be displayed)

        # 4. Access protected routes
        response = client.get("/history")
        assert response.status_code == 200

        response = client.get("/settings")
        assert response.status_code == 200

        # 5. Check auth status
        response = client.get("/auth/check")
        assert response.status_code == 200
        data = response.get_json()
        assert data["authenticated"] is True
        assert data["username"] == "integrationtest"

        # 6. Logout
        response = client.post("/auth/logout", follow_redirects=True)
        assert response.status_code == 200
        assert b"You have been logged out successfully" in response.data

        # 7. Should not be able to access protected routes
        response = client.get("/history")
        assert response.status_code == 302
        assert "/auth/login" in response.location

        # 8. Login again
        response = client.post(
            "/auth/login",
            data={"username": "integrationtest", "password": "TestPass123"},
            follow_redirects=True,
        )
        assert response.status_code == 200

        # 9. Should be able to access protected routes again
        response = client.get("/history")
        assert response.status_code == 200

    def test_multiple_users(self, client):
        """Test that multiple users can register and have separate sessions."""
        # Register first user
        client.post(
            "/auth/register",
            data={
                "username": "user1",
                "password": "Password1X",
                "confirm_password": "Password1X",
                "acknowledge": "true",
            },
        )

        # Check logged in as user1
        response = client.get("/auth/check")
        data = response.get_json()
        assert data["username"] == "user1"

        # Logout
        client.post("/auth/logout")

        # Register second user
        client.post(
            "/auth/register",
            data={
                "username": "user2",
                "password": "Password2X",
                "confirm_password": "Password2X",
                "acknowledge": "true",
            },
        )

        # Check logged in as user2
        response = client.get("/auth/check")
        data = response.get_json()
        assert data["username"] == "user2"

        # Verify both users exist in auth database
        auth_db = get_auth_db_session()
        users = auth_db.query(User).all()
        assert len(users) == 2
        usernames = [user.username for user in users]
        assert "user1" in usernames
        assert "user2" in usernames
        auth_db.close()

    def test_session_persistence(self, client):
        """Test that sessions persist across requests."""
        # Register and login
        client.post(
            "/auth/register",
            data={
                "username": "sessiontest",
                "password": "TestPass123",  # pragma: allowlist secret
                "confirm_password": "TestPass123",  # pragma: allowlist secret
                "acknowledge": "true",
            },
        )

        # Make multiple requests
        for _ in range(5):
            response = client.get("/history")
            assert response.status_code == 200

            response = client.get("/auth/check")
            data = response.get_json()
            assert data["username"] == "sessiontest"

    def test_protected_api_endpoints(self, client):
        """Test that API endpoints require authentication."""
        # Test various API endpoints without auth
        endpoints = [
            ("/api/start_research", "POST"),
            ("/api/history", "GET"),
            ("/settings/api", "GET"),
            ("/history/api", "GET"),
            ("/metrics/api/metrics", "GET"),
        ]

        for endpoint, method in endpoints:
            if method == "GET":
                response = client.get(endpoint)
            else:
                response = client.post(endpoint, json={})

            # All these endpoints contain the slash-bounded `api` segment
            # (either /api/ in the middle or trailing /api), so the auth
            # decorator returns JSON 401 for unauthenticated callers.
            assert response.status_code == 401
            assert response.get_json()["error"] == "Authentication required"

        # Register and try again
        client.post(
            "/auth/register",
            data={
                "username": "apitest",
                "password": "TestPass123",  # pragma: allowlist secret
                "confirm_password": "TestPass123",  # pragma: allowlist secret
                "acknowledge": "true",
            },
        )

        # Now endpoints should be accessible (may return errors but not redirects)
        for endpoint, method in endpoints:
            if method == "GET":
                response = client.get(endpoint)
            else:
                response = client.post(endpoint, json={})

            assert response.status_code != 302  # Not a redirect

    @pytest.mark.skipif(
        os.environ.get("CI") == "true"
        or os.environ.get("GITHUB_ACTIONS") == "true",
        reason="Password change with encrypted DB re-keying is complex to test in CI",
    )
    def test_password_change_flow(self, client):
        """Test the complete password change flow."""
        # Register user
        client.post(
            "/auth/register",
            data={
                "username": "changetest",
                "password": "OldPass123",
                "confirm_password": "OldPass123",
                "acknowledge": "true",
            },
        )

        # Change password
        response = client.post(
            "/auth/change-password",
            data={
                "current_password": "OldPass123",
                "new_password": "NewPass456",  # pragma: allowlist secret
                "confirm_password": "NewPass456",
            },
            follow_redirects=True,
        )

        assert b"Password changed successfully" in response.data
        assert b"Please login with your new password" in response.data

        # Should be logged out
        response = client.get("/auth/check")
        assert response.status_code == 401

        # Old password should fail
        response = client.post(
            "/auth/login",
            data={"username": "changetest", "password": "OldPass123"},
        )
        assert response.status_code == 401

        # New password should work
        response = client.post(
            "/auth/login",
            data={"username": "changetest", "password": "NewPass456"},
            follow_redirects=True,
        )
        assert response.status_code == 200

        # Verify logged in
        response = client.get("/auth/check")
        data = response.get_json()
        assert data["username"] == "changetest"