portainer-core/services/scheduler/tests/conftest.py

"""
Shared pytest fixtures for The Scheduler tests.
"""
import asyncio
import os
from typing import Generator, AsyncGenerator
from unittest.mock import Mock, MagicMock

import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport

# Set test environment variables before importing app
os.environ["POSTGRES_HOST"] = "postgres-shared"  # Use real postgres for integration tests
os.environ["POSTGRES_DB"] = "test_scheduler"
os.environ["POSTGRES_USER"] = "test_scheduler_user"
os.environ["POSTGRES_PASSWORD"] = "test_password_12345"
os.environ["SCHEDULER_API_KEY"] = "test-api-key-12345"
os.environ["GITEA_USER"] = "test-librarian"
os.environ["GITEA_PASSWORD"] = "test-gitea-token"
os.environ["REDIS_HOST"] = "redis-shared"

from src.main import app
from src.config import Settings, get_settings


# Override settings for tests
@pytest.fixture
def test_settings() -> Settings:
    """Provide test settings loaded from environment variables."""
    # Settings are already loaded from environment (set at module level)
    return get_settings()


@pytest.fixture
def api_key() -> str:
    """Test API key for authenticated requests."""
    return "test-api-key-12345"


@pytest.fixture
def auth_headers(api_key: str) -> dict:
    """Authentication headers for test requests."""
    return {"Authorization": f"Bearer {api_key}"}


# Synchronous test client
@pytest.fixture
def client() -> Generator[TestClient, None, None]:
    """
    FastAPI test client for synchronous tests.
    Note: Some endpoints may fail if they require real database connections.
    """
    with TestClient(app) as c:
        yield c


# Async test client
@pytest.fixture
async def async_client() -> AsyncGenerator[AsyncClient, None]:
    """
    Async HTTP client for testing async endpoints.
    Note: Some endpoints may fail if they require real database connections.
    """
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac


# Mock database connection
@pytest.fixture
def mock_db_connection():
    """Mock database connection for testing without real database."""
    mock_conn = MagicMock()
    mock_cursor = MagicMock()

    # Setup context managers
    mock_conn.__enter__ = Mock(return_value=mock_conn)
    mock_conn.__exit__ = Mock(return_value=None)
    mock_conn.cursor.return_value.__enter__ = Mock(return_value=mock_cursor)
    mock_conn.cursor.return_value.__exit__ = Mock(return_value=None)

    return mock_conn, mock_cursor


# Mock scheduler
@pytest.fixture
def mock_scheduler():
    """Mock APScheduler for testing without real scheduler."""
    mock = MagicMock()
    mock.running = True
    mock.get_jobs.return_value = []
    return mock


# Mock task executor
@pytest.fixture
def mock_task_executor():
    """Mock TaskExecutor for testing without real executor."""
    mock = MagicMock()
    return mock


# Sample task data
@pytest.fixture
def sample_task_data() -> dict:
    """Sample task data for testing."""
    return {
        "id": 1,
        "task_name": "test_task",
        "service": "scheduler",
        "executor": "example_executor",
        "priority": 50,
        "minute": 0,
        "hour": 4,
        "day_of_month": -1,
        "month": -1,
        "day_of_week": -1,
        "enabled": True,
        "description": "Test task for unit tests",
        "config": {
            "message": "Test message",
            "delay_seconds": 1
        },
        "max_retries": 3,
        "timeout_seconds": 60,
        "retry_count": 0,
        "last_run": None,
        "last_status": None,
        "last_duration_seconds": None,
        "created_at": "2025-12-07T00:00:00",
        "updated_at": "2025-12-07T00:00:00",
        "created_by": "test"
    }


# Sample doc sync config
@pytest.fixture
def sample_doc_sync_config() -> dict:
    """Sample doc sync configuration for testing."""
    return {
        "project": "test-project",
        "upstream_repo": "https://github.com/test/repo.git",
        "docs_paths": ["/docs"],
        "gitea_repo": "library/test-docs",
        "branch": "main"
    }


# Temporary directory for file operations
@pytest.fixture
def temp_work_dir(tmp_path):
    """Temporary directory for testing file operations."""
    work_dir = tmp_path / "test-work"
    work_dir.mkdir()
    return work_dir


# Database fixtures for integration tests
@pytest.fixture(scope="function")
def db_connection():
    """Provide a database connection for integration tests."""
    import psycopg2

    conn = psycopg2.connect(
        host="postgres-shared",
        database="test_scheduler",
        user="test_scheduler_user",
        password="test_password_12345"
    )
    yield conn

    # Cleanup: rollback any uncommitted changes
    conn.rollback()
    conn.close()


@pytest.fixture(scope="function")
def clean_database(db_connection):
    """Clean test database before each test."""
    cursor = db_connection.cursor()

    # Delete all test data
    cursor.execute("DELETE FROM task_executions")
    cursor.execute("DELETE FROM scheduled_tasks")
    db_connection.commit()

    yield db_connection

    # Cleanup after test
    cursor.execute("DELETE FROM task_executions")
    cursor.execute("DELETE FROM scheduled_tasks")
    db_connection.commit()
    cursor.close()


@pytest.fixture
def sample_task_in_db(clean_database):
    """Insert a sample task into the test database."""
    cursor = clean_database.cursor()

    cursor.execute("""
        INSERT INTO scheduled_tasks
            (task_name, service, executor, priority, minute, hour,
             day_of_month, month, day_of_week, enabled, description,
             config, max_retries, timeout_seconds, created_by)
        VALUES
            (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s)
        RETURNING id
    """, (
        'test_task', 'scheduler', 'example_executor', 50,
        -1, -1, -1, -1, -1, True, 'Test task',
        '{"message": "Test", "delay_seconds": 0}',
        3, 60, 'test'
    ))

    task_id = cursor.fetchone()[0]
    clean_database.commit()
    cursor.close()

    return task_id