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.
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Check current tasks in the database."""
|
|
|
|
import asyncio
|
|
from datetime import datetime
|
|
|
|
from app.core.database import get_session_factory
|
|
from app.repositories.scheduled_task import ScheduledTaskRepository
|
|
|
|
async def check_tasks():
|
|
session_factory = get_session_factory()
|
|
|
|
async with session_factory() as session:
|
|
repo = ScheduledTaskRepository(session)
|
|
|
|
# Get all tasks
|
|
all_tasks = await repo.get_all(limit=20)
|
|
|
|
print("All tasks in database:")
|
|
print("=" * 80)
|
|
for task in all_tasks:
|
|
print(f"ID: {task.id}")
|
|
print(f"Name: {task.name}")
|
|
print(f"Type: {task.task_type}")
|
|
print(f"Status: {task.status}")
|
|
print(f"Scheduled: {task.scheduled_at}")
|
|
print(f"Timezone: {task.timezone}")
|
|
print(f"Active: {task.is_active}")
|
|
print(f"User ID: {task.user_id}")
|
|
print(f"Executions: {task.executions_count}")
|
|
print(f"Last executed: {task.last_executed_at}")
|
|
print(f"Error: {task.error_message}")
|
|
print(f"Parameters: {task.parameters}")
|
|
print("-" * 40)
|
|
|
|
# Check specifically for pending tasks
|
|
print(f"\nCurrent time: {datetime.utcnow()}")
|
|
print("\nPending tasks:")
|
|
from app.models.scheduled_task import TaskStatus
|
|
pending_tasks = await repo.get_all(limit=10)
|
|
pending_tasks = [t for t in pending_tasks if t.status == TaskStatus.PENDING and t.is_active]
|
|
|
|
if not pending_tasks:
|
|
print("No pending tasks found")
|
|
else:
|
|
for task in pending_tasks:
|
|
time_diff = task.scheduled_at - datetime.utcnow()
|
|
print(f"- {task.name} (ID: {task.id}): scheduled for {task.scheduled_at} (in {time_diff})")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(check_tasks()) |