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>
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
"""
|
|
Tests for the example executor.
|
|
"""
|
|
import pytest
|
|
import asyncio
|
|
from src.executors import example_executor
|
|
from src.config import Settings
|
|
|
|
|
|
@pytest.mark.executor
|
|
@pytest.mark.unit
|
|
class TestExampleExecutor:
|
|
"""Tests for example_executor module."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_returns_success_message(self, test_settings: Settings):
|
|
"""Test that executor returns success message."""
|
|
config = {
|
|
"message": "Test message",
|
|
"delay_seconds": 0
|
|
}
|
|
|
|
result = await example_executor.execute(config, test_settings)
|
|
|
|
assert "Test message" in result
|
|
assert "took" in result.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_with_delay(self, test_settings: Settings):
|
|
"""Test that executor respects delay_seconds."""
|
|
import time
|
|
|
|
config = {
|
|
"message": "Delayed test",
|
|
"delay_seconds": 1
|
|
}
|
|
|
|
start = time.time()
|
|
result = await example_executor.execute(config, test_settings)
|
|
elapsed = time.time() - start
|
|
|
|
# Should take at least 1 second
|
|
assert elapsed >= 1.0
|
|
assert "Delayed test" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_with_zero_delay(self, test_settings: Settings):
|
|
"""Test that executor works with zero delay."""
|
|
config = {
|
|
"message": "Instant",
|
|
"delay_seconds": 0
|
|
}
|
|
|
|
result = await example_executor.execute(config, test_settings)
|
|
|
|
assert "Instant" in result
|
|
assert "took 0s" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_missing_config_fields(self, test_settings: Settings):
|
|
"""Test that executor handles missing config fields gracefully."""
|
|
config = {}
|
|
|
|
# Should use defaults
|
|
result = await example_executor.execute(config, test_settings)
|
|
|
|
assert isinstance(result, str)
|
|
assert "took" in result.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_executor_invalid_delay_type(self, test_settings: Settings):
|
|
"""Test that executor handles invalid delay type."""
|
|
config = {
|
|
"message": "Test",
|
|
"delay_seconds": "invalid" # Should be int
|
|
}
|
|
|
|
# Should raise TypeError or handle gracefully
|
|
with pytest.raises((TypeError, ValueError)):
|
|
await example_executor.execute(config, test_settings)
|