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>
312 lines
14 KiB
Python
312 lines
14 KiB
Python
"""
|
|
Comprehensive API tests to improve coverage of main.py.
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch, MagicMock
|
|
import json
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestTaskCRUDOperations:
|
|
"""Comprehensive CRUD tests for task endpoints."""
|
|
|
|
def test_list_tasks_empty(self, client: TestClient, auth_headers: dict):
|
|
"""Test listing tasks when none exist."""
|
|
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("/tasks", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert "tasks" in data
|
|
assert data["count"] == 0
|
|
|
|
def test_list_tasks_with_filters(self, client: TestClient, auth_headers: dict):
|
|
"""Test listing tasks with enabled and service filters."""
|
|
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(
|
|
"/tasks?enabled=true&service=scheduler",
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Should execute filtered query
|
|
assert mock_cursor.execute.called
|
|
call_args = str(mock_cursor.execute.call_args)
|
|
assert "enabled" in call_args.lower() or response.status_code in [200, 500]
|
|
|
|
def test_get_task_details_not_found(self, client: TestClient, auth_headers: dict):
|
|
"""Test getting details for non-existent task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = None
|
|
|
|
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("/tasks/nonexistent", headers=auth_headers)
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_update_task(self, client: TestClient, auth_headers: dict):
|
|
"""Test updating a task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = {
|
|
"task_name": "test",
|
|
"priority": 60
|
|
}
|
|
|
|
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.put(
|
|
"/tasks/test",
|
|
headers=auth_headers,
|
|
json={"priority": 60}
|
|
)
|
|
|
|
# Should have attempted update
|
|
assert mock_cursor.execute.called
|
|
|
|
def test_update_task_not_found(self, client: TestClient, auth_headers: dict):
|
|
"""Test updating non-existent task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = None
|
|
|
|
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.put(
|
|
"/tasks/nonexistent",
|
|
headers=auth_headers,
|
|
json={"priority": 60}
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_update_task_no_fields(self, client: TestClient, auth_headers: dict):
|
|
"""Test updating task with no valid fields."""
|
|
response = client.put(
|
|
"/tasks/test",
|
|
headers=auth_headers,
|
|
json={"invalid_field": "value"}
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
def test_delete_task(self, client: TestClient, auth_headers: dict):
|
|
"""Test deleting a task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = ("test_task",)
|
|
|
|
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.delete("/tasks/test_task", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert "deleted successfully" in data["message"].lower()
|
|
|
|
def test_delete_task_not_found(self, client: TestClient, auth_headers: dict):
|
|
"""Test deleting non-existent task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = None
|
|
|
|
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.delete("/tasks/nonexistent", headers=auth_headers)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestTriggerEndpoint:
|
|
"""Tests for task trigger endpoint."""
|
|
|
|
def test_trigger_disabled_task(self, client: TestClient, auth_headers: dict):
|
|
"""Test triggering a disabled task."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = {
|
|
"task_name": "test",
|
|
"enabled": False
|
|
}
|
|
|
|
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/test/trigger", headers=auth_headers)
|
|
|
|
assert response.status_code == 400
|
|
|
|
def test_trigger_nonexistent_task(self, client: TestClient, auth_headers: dict):
|
|
"""Test triggering a task that doesn't exist."""
|
|
with patch('src.main.get_task_executor') as mock_executor:
|
|
mock_conn = MagicMock()
|
|
mock_cursor = MagicMock()
|
|
mock_cursor.fetchone.return_value = None
|
|
|
|
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/nonexistent/trigger", headers=auth_headers)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestLegacyEndpoints:
|
|
"""Tests for deprecated/legacy endpoints."""
|
|
|
|
def test_trigger_backup_endpoint(self, client: TestClient, auth_headers: dict):
|
|
"""Test legacy backup trigger endpoint."""
|
|
response = client.post("/tasks/backup", headers=auth_headers)
|
|
|
|
# Should return not implemented status
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["status"] == "not_implemented"
|
|
|
|
def test_trigger_docs_update_endpoint(self, client: TestClient, auth_headers: dict):
|
|
"""Test legacy docs update endpoint."""
|
|
response = client.post("/tasks/docs/update", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["status"] == "not_implemented"
|
|
|
|
def test_trigger_docs_update_with_project(self, client: TestClient, auth_headers: dict):
|
|
"""Test legacy docs update with project filter."""
|
|
response = client.post(
|
|
"/tasks/docs/update?project=test",
|
|
headers=auth_headers
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert "test" in data["message"].lower()
|
|
|
|
def test_check_doc_versions_endpoint(self, client: TestClient, auth_headers: dict):
|
|
"""Test legacy version check endpoint."""
|
|
response = client.post("/tasks/docs/check-versions", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["status"] == "not_implemented"
|
|
|
|
def test_trigger_cleanup_endpoint(self, client: TestClient, auth_headers: dict):
|
|
"""Test legacy cleanup endpoint."""
|
|
response = client.post("/tasks/cleanup", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["status"] == "not_implemented"
|
|
|
|
|
|
@pytest.mark.api
|
|
@pytest.mark.unit
|
|
class TestExecutionFiltering:
|
|
"""Tests for execution history filtering."""
|
|
|
|
def test_filter_by_service(self, client: TestClient, auth_headers: dict):
|
|
"""Test filtering executions by service."""
|
|
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?service=scheduler",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert mock_cursor.execute.called
|
|
|
|
def test_filter_by_status(self, client: TestClient, auth_headers: dict):
|
|
"""Test filtering executions by status."""
|
|
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?status=success",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert mock_cursor.execute.called
|
|
|
|
def test_custom_limit(self, client: TestClient, auth_headers: dict):
|
|
"""Test custom limit for executions."""
|
|
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?limit=50", headers=auth_headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["limit"] == 50
|