03abed6d39
- Implemented tests for ScheduledTaskRepository covering task creation, retrieval, filtering, and status updates. - Developed tests for SchedulerService including task creation, cancellation, user task retrieval, and maintenance jobs. - Created tests for TaskHandlerRegistry to validate task execution for various types, including credit recharge and sound playback. - Ensured proper error handling and edge cases in task execution scenarios. - Added fixtures and mocks to facilitate isolated testing of services and repositories.
36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Create a test task with a future execution time."""
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
|
|
from app.core.database import get_session_factory
|
|
from app.repositories.scheduled_task import ScheduledTaskRepository
|
|
from app.models.scheduled_task import TaskType, RecurrenceType
|
|
|
|
async def create_future_task():
|
|
session_factory = get_session_factory()
|
|
|
|
# Create a task for 2 minutes from now
|
|
future_time = datetime.utcnow() + timedelta(minutes=2)
|
|
|
|
async with session_factory() as session:
|
|
repo = ScheduledTaskRepository(session)
|
|
|
|
task_data = {
|
|
"name": f"Future Task {future_time.strftime('%H:%M:%S')}",
|
|
"task_type": TaskType.PLAY_SOUND,
|
|
"scheduled_at": future_time,
|
|
"timezone": "UTC",
|
|
"parameters": {"sound_id": 1},
|
|
"user_id": 1,
|
|
"recurrence_type": RecurrenceType.NONE,
|
|
}
|
|
|
|
task = await repo.create(task_data)
|
|
print(f"Created task: {task.name} (ID: {task.id}) scheduled for {task.scheduled_at}")
|
|
print(f"Current time: {datetime.utcnow()}")
|
|
print(f"Task will execute in: {future_time - datetime.utcnow()}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(create_future_task()) |