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>
308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""
|
|
Database integration tests using real test database.
|
|
These tests require the test database to be set up.
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestDatabaseTaskOperations:
|
|
"""Integration tests for task CRUD with real database."""
|
|
|
|
def test_create_task_in_database(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test creating a task writes to database."""
|
|
task_data = {
|
|
"task_name": "integration_test_task",
|
|
"service": "scheduler",
|
|
"executor": "example_executor",
|
|
"priority": 40,
|
|
"description": "Integration test task",
|
|
"config": {"message": "Test", "delay_seconds": 0}
|
|
}
|
|
|
|
response = client.post("/tasks", headers=auth_headers, json=task_data)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["task_name"] == "integration_test_task"
|
|
assert data["priority"] == 40
|
|
|
|
# Verify it's in database
|
|
cursor = clean_database.cursor()
|
|
cursor.execute("SELECT task_name, priority FROM scheduled_tasks WHERE task_name = %s",
|
|
("integration_test_task",))
|
|
result = cursor.fetchone()
|
|
assert result is not None
|
|
assert result[0] == "integration_test_task"
|
|
assert result[1] == 40
|
|
|
|
def test_list_tasks_from_database(self, client: TestClient, auth_headers: dict, sample_task_in_db):
|
|
"""Test listing tasks reads from database."""
|
|
response = client.get("/tasks", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["count"] >= 1
|
|
assert any(task["task_name"] == "test_task" for task in data["tasks"])
|
|
|
|
def test_get_task_details_from_database(self, client: TestClient, auth_headers: dict, sample_task_in_db):
|
|
"""Test getting task details from database."""
|
|
response = client.get("/tasks/test_task", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["task_name"] == "test_task"
|
|
assert data["executor"] == "example_executor"
|
|
assert data["priority"] == 50
|
|
|
|
def test_update_task_in_database(self, client: TestClient, auth_headers: dict, sample_task_in_db, clean_database):
|
|
"""Test updating a task modifies database."""
|
|
update_data = {"priority": 99, "description": "Updated description"}
|
|
|
|
response = client.put("/tasks/test_task", headers=auth_headers, json=update_data)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["priority"] == 99
|
|
assert data["description"] == "Updated description"
|
|
|
|
# Verify in database
|
|
cursor = clean_database.cursor()
|
|
cursor.execute("SELECT priority, description FROM scheduled_tasks WHERE task_name = %s",
|
|
("test_task",))
|
|
result = cursor.fetchone()
|
|
assert result[0] == 99
|
|
assert result[1] == "Updated description"
|
|
|
|
def test_delete_task_from_database(self, client: TestClient, auth_headers: dict, sample_task_in_db, clean_database):
|
|
"""Test deleting a task removes from database."""
|
|
response = client.delete("/tasks/test_task", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Verify removed from database
|
|
cursor = clean_database.cursor()
|
|
cursor.execute("SELECT COUNT(*) FROM scheduled_tasks WHERE task_name = %s",
|
|
("test_task",))
|
|
count = cursor.fetchone()[0]
|
|
assert count == 0
|
|
|
|
def test_filter_tasks_by_service(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test filtering tasks by service."""
|
|
# Create tasks with different services
|
|
cursor = clean_database.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO scheduled_tasks (task_name, service, executor, priority)
|
|
VALUES ('task1', 'scheduler', 'example_executor', 50),
|
|
('task2', 'backup', 'backup_executor', 30)
|
|
""")
|
|
clean_database.commit()
|
|
|
|
response = client.get("/tasks?service=scheduler", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(task["service"] == "scheduler" for task in data["tasks"])
|
|
|
|
def test_filter_tasks_by_enabled(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test filtering tasks by enabled status."""
|
|
cursor = clean_database.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO scheduled_tasks (task_name, service, executor, priority, enabled)
|
|
VALUES ('enabled_task', 'scheduler', 'example_executor', 50, true),
|
|
('disabled_task', 'scheduler', 'example_executor', 50, false)
|
|
""")
|
|
clean_database.commit()
|
|
|
|
response = client.get("/tasks?enabled=true", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(task["enabled"] is True for task in data["tasks"])
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestDatabaseTaskExecution:
|
|
"""Integration tests for task execution with database tracking."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_task_creates_execution_record(self, sample_task_in_db, clean_database):
|
|
"""Test that executing a task creates execution record."""
|
|
from src.tasks.executor import TaskExecutor
|
|
from src.config import get_settings
|
|
|
|
settings = get_settings()
|
|
executor = TaskExecutor(settings)
|
|
|
|
# Get task from database
|
|
cursor = clean_database.cursor()
|
|
cursor.execute("""
|
|
SELECT id, task_name, service, executor, priority, minute, hour,
|
|
day_of_month, month, day_of_week, enabled, description,
|
|
config, max_retries, timeout_seconds, retry_count,
|
|
last_run, last_status, last_duration_seconds, created_by
|
|
FROM scheduled_tasks WHERE id = %s
|
|
""", (sample_task_in_db,))
|
|
|
|
row = cursor.fetchone()
|
|
task = {
|
|
"id": row[0],
|
|
"task_name": row[1],
|
|
"service": row[2],
|
|
"executor": row[3],
|
|
"priority": row[4],
|
|
"minute": row[5],
|
|
"hour": row[6],
|
|
"day_of_month": row[7],
|
|
"month": row[8],
|
|
"day_of_week": row[9],
|
|
"enabled": row[10],
|
|
"description": row[11],
|
|
"config": row[12],
|
|
"max_retries": row[13],
|
|
"timeout_seconds": row[14],
|
|
"retry_count": row[15],
|
|
"last_run": row[16],
|
|
"last_status": row[17],
|
|
"last_duration_seconds": row[18],
|
|
"created_by": row[19]
|
|
}
|
|
|
|
# Execute task
|
|
await executor.execute_task(task)
|
|
|
|
# Verify execution record was created
|
|
cursor.execute("""
|
|
SELECT status, task_name, executor FROM task_executions
|
|
WHERE task_id = %s ORDER BY id DESC LIMIT 1
|
|
""", (sample_task_in_db,))
|
|
|
|
execution = cursor.fetchone()
|
|
assert execution is not None
|
|
assert execution[0] in ["success", "failed", "timeout"]
|
|
assert execution[1] == "test_task"
|
|
assert execution[2] == "example_executor"
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestDatabaseExecutionHistory:
|
|
"""Integration tests for execution history endpoints."""
|
|
|
|
def test_get_execution_history(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test retrieving execution history from database."""
|
|
# Create test execution records
|
|
cursor = clean_database.cursor()
|
|
|
|
# First create a task
|
|
cursor.execute("""
|
|
INSERT INTO scheduled_tasks (task_name, service, executor, priority)
|
|
VALUES ('history_task', 'scheduler', 'example_executor', 50)
|
|
RETURNING id
|
|
""")
|
|
task_id = cursor.fetchone()[0]
|
|
|
|
# Create execution records
|
|
cursor.execute("""
|
|
INSERT INTO task_executions
|
|
(task_id, task_name, service, executor, priority, status, duration_seconds)
|
|
VALUES
|
|
(%s, 'history_task', 'scheduler', 'example_executor', 50, 'success', 5),
|
|
(%s, 'history_task', 'scheduler', 'example_executor', 50, 'success', 3),
|
|
(%s, 'history_task', 'scheduler', 'example_executor', 50, 'failed', 2)
|
|
""", (task_id, task_id, task_id))
|
|
clean_database.commit()
|
|
|
|
response = client.get("/executions", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["count"] >= 3
|
|
assert any(ex["task_name"] == "history_task" for ex in data["executions"])
|
|
|
|
def test_filter_executions_by_task_name(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test filtering execution history by task name."""
|
|
cursor = clean_database.cursor()
|
|
|
|
# Create tasks
|
|
cursor.execute("""
|
|
INSERT INTO scheduled_tasks (task_name, service, executor, priority)
|
|
VALUES ('task_a', 'scheduler', 'example_executor', 50),
|
|
('task_b', 'scheduler', 'example_executor', 50)
|
|
RETURNING id
|
|
""")
|
|
task_ids = [row[0] for row in cursor.fetchall()]
|
|
|
|
# Create executions
|
|
cursor.execute("""
|
|
INSERT INTO task_executions (task_id, task_name, service, executor, priority, status)
|
|
VALUES (%s, 'task_a', 'scheduler', 'example_executor', 50, 'success'),
|
|
(%s, 'task_b', 'scheduler', 'example_executor', 50, 'success')
|
|
""", task_ids)
|
|
clean_database.commit()
|
|
|
|
response = client.get("/executions?task_name=task_a", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(ex["task_name"] == "task_a" for ex in data["executions"])
|
|
|
|
def test_filter_executions_by_status(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test filtering execution history by status."""
|
|
cursor = clean_database.cursor()
|
|
|
|
cursor.execute("""
|
|
INSERT INTO scheduled_tasks (task_name, service, executor, priority)
|
|
VALUES ('status_task', 'scheduler', 'example_executor', 50)
|
|
RETURNING id
|
|
""")
|
|
task_id = cursor.fetchone()[0]
|
|
|
|
cursor.execute("""
|
|
INSERT INTO task_executions (task_id, task_name, service, executor, priority, status)
|
|
VALUES (%s, 'status_task', 'scheduler', 'example_executor', 50, 'success'),
|
|
(%s, 'status_task', 'scheduler', 'example_executor', 50, 'failed')
|
|
""", (task_id, task_id))
|
|
clean_database.commit()
|
|
|
|
response = client.get("/executions?status=success", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
success_executions = [ex for ex in data["executions"] if ex["task_name"] == "status_task"]
|
|
assert all(ex["status"] == "success" for ex in success_executions)
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestDatabaseStats:
|
|
"""Integration tests for stats endpoint with database."""
|
|
|
|
def test_stats_endpoint_with_database(self, client: TestClient, auth_headers: dict, clean_database):
|
|
"""Test stats endpoint returns accurate database counts."""
|
|
cursor = clean_database.cursor()
|
|
|
|
# Create test data
|
|
cursor.execute("""
|
|
INSERT INTO scheduled_tasks (task_name, service, executor, priority, enabled)
|
|
VALUES ('enabled_1', 'scheduler', 'example_executor', 50, true),
|
|
('enabled_2', 'scheduler', 'example_executor', 50, true),
|
|
('disabled_1', 'scheduler', 'example_executor', 50, false)
|
|
RETURNING id
|
|
""")
|
|
task_ids = [row[0] for row in cursor.fetchall()]
|
|
|
|
cursor.execute("""
|
|
INSERT INTO task_executions (task_id, task_name, service, executor, priority, status)
|
|
VALUES (%s, 'enabled_1', 'scheduler', 'example_executor', 50, 'success'),
|
|
(%s, 'enabled_2', 'scheduler', 'example_executor', 50, 'success'),
|
|
(%s, 'enabled_1', 'scheduler', 'example_executor', 50, 'failed')
|
|
""", task_ids)
|
|
clean_database.commit()
|
|
|
|
response = client.get("/stats", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["tasks_enabled"] >= 2
|
|
assert "execution_stats_24h" in data
|