jpmschweitzerandClaude Opus 4.5
64574bcc39
Build and Push / build (release) Failing after 17s
Extracted standalone scheduler service with: - FastAPI REST API for task management - APScheduler-based task execution - PostgreSQL persistence - Docker container support - Gitea Actions CI/CD workflow 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
237 lines
9.5 KiB
Python
237 lines
9.5 KiB
Python
"""
|
|
Tests for The Scheduler API endpoints.
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch, MagicMock
|
|
import json
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestHealthEndpoint:
|
|
"""Tests for /health endpoint."""
|
|
|
|
def test_health_endpoint_returns_healthy(self, client: TestClient):
|
|
"""Test that health endpoint returns healthy status."""
|
|
with patch('src.main.get_scheduler') as mock_get_scheduler:
|
|
mock_scheduler = MagicMock()
|
|
mock_scheduler.running = True
|
|
mock_scheduler.get_jobs.return_value = [MagicMock()]
|
|
mock_get_scheduler.return_value = mock_scheduler
|
|
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
assert data["scheduler_running"] is True
|
|
assert data["jobs_count"] == 1
|
|
|
|
def test_health_endpoint_no_auth_required(self, client: TestClient):
|
|
"""Test that health endpoint doesn't require authentication."""
|
|
with patch('src.main.get_scheduler') as mock_get_scheduler:
|
|
mock_scheduler = MagicMock()
|
|
mock_scheduler.running = True
|
|
mock_get_scheduler.return_value = mock_scheduler
|
|
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestAuthenticationEndpoints:
|
|
"""Tests for API authentication."""
|
|
|
|
def test_missing_api_key_returns_401(self, client: TestClient):
|
|
"""Test that missing API key returns 401."""
|
|
response = client.get("/tasks")
|
|
assert response.status_code == 401
|
|
|
|
def test_invalid_api_key_returns_403(self, client: TestClient):
|
|
"""Test that invalid API key returns 403."""
|
|
response = client.get(
|
|
"/tasks",
|
|
headers={"Authorization": "Bearer wrong-key"}
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
def test_valid_api_key_allows_access(self, client: TestClient, auth_headers: dict):
|
|
"""Test that valid API key allows access."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_executor.return_value.get_db_connection.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchall.return_value = []
|
|
|
|
response = client.get("/tasks", headers=auth_headers)
|
|
# May fail with 500 due to DB, but should not be 401/403
|
|
assert response.status_code not in [401, 403]
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestTaskEndpoints:
|
|
"""Tests for task management endpoints."""
|
|
|
|
def test_create_task_missing_fields_returns_400(self, client: TestClient, auth_headers: dict):
|
|
"""Test that creating task without required fields returns 400."""
|
|
incomplete_task = {
|
|
"task_name": "test",
|
|
# Missing service, executor, priority
|
|
}
|
|
response = client.post(
|
|
"/tasks",
|
|
headers=auth_headers,
|
|
json=incomplete_task
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_create_task_with_valid_data(self, client: TestClient, auth_headers: dict, sample_task_data: dict):
|
|
"""Test creating a task with valid data."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
|
|
# Setup mock to return task data
|
|
mock_cursor.fetchone.return_value = {**sample_task_data, "id": 1}
|
|
|
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
|
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
response = client.post(
|
|
"/tasks",
|
|
headers=auth_headers,
|
|
json=sample_task_data
|
|
)
|
|
|
|
# Verify the call was made
|
|
assert mock_cursor.execute.called
|
|
# Check that config was JSON-encoded
|
|
call_args = mock_cursor.execute.call_args
|
|
assert 'config' in call_args[0][1]
|
|
|
|
def test_trigger_task_endpoint(self, client: TestClient, auth_headers: dict):
|
|
"""Test manually triggering a task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
|
|
# Mock task retrieval
|
|
mock_cursor.fetchone.return_value = {
|
|
"task_name": "test_task",
|
|
"enabled": True,
|
|
"priority": 50,
|
|
"executor": "example_executor"
|
|
}
|
|
|
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
|
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
with patch('asyncio.create_task'):
|
|
response = client.post(
|
|
"/tasks/test_task/trigger",
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Should return success message
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["task_name"] == "test_task"
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestStatsEndpoint:
|
|
"""Tests for /stats endpoint."""
|
|
|
|
def test_stats_endpoint_requires_auth(self, client: TestClient):
|
|
"""Test that stats endpoint requires authentication."""
|
|
response = client.get("/stats")
|
|
assert response.status_code == 401
|
|
|
|
def test_stats_endpoint_returns_metrics(self, client: TestClient, auth_headers: dict):
|
|
"""Test that stats endpoint returns system metrics."""
|
|
with patch('src.main.get_scheduler') as mock_scheduler, \
|
|
patch('src.main.get_task_executor') as mock_executor:
|
|
|
|
mock_scheduler.return_value.running = True
|
|
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.side_effect = [
|
|
{"count": 3}, # enabled tasks
|
|
{"count": 0}, # running tasks
|
|
]
|
|
mock_cursor.fetchall.return_value = [
|
|
{"status": "success", "count": 10},
|
|
{"status": "failed", "count": 1}
|
|
]
|
|
|
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
|
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
response = client.get("/stats", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert "scheduler_running" in data
|
|
assert "tasks_enabled" in data
|
|
assert "concurrent_limit" in data
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestExecutionHistoryEndpoint:
|
|
"""Tests for /executions endpoint."""
|
|
|
|
def test_executions_endpoint_returns_history(self, client: TestClient, auth_headers: dict):
|
|
"""Test that executions endpoint returns execution history."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchall.return_value = [
|
|
{
|
|
"id": 1,
|
|
"task_name": "test_task",
|
|
"status": "success",
|
|
"duration_seconds": 5
|
|
}
|
|
]
|
|
|
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
|
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
response = client.get("/executions", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert "executions" in data
|
|
assert "count" in data
|
|
|
|
def test_executions_filter_by_task_name(self, client: TestClient, auth_headers: dict):
|
|
"""Test filtering executions by task name."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchall.return_value = []
|
|
|
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
|
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
|
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
|
|
|
response = client.get(
|
|
"/executions?task_name=test_task&limit=10",
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Should execute query with filters
|
|
assert mock_cursor.execute.called
|