1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
import pytest
import time
from unittest.mock import MagicMock, AsyncMock, patch
@pytest.fixture
def cron_cog(bot):
# Patch create_task to prevent background task creation
with patch.object(bot.loop, 'create_task', return_value=MagicMock()):
from cogs.cron import Cron
cog = Cron(bot)
yield cog
@pytest.mark.slow
@pytest.mark.asyncio
async def test_cron_job_execution(cron_cog):
"""Test cron job executes when due."""
cog = cron_cog
job = {
"id": "test_job",
"cron_expression": "* * * * *",
"type": "message",
"channel_id": 456,
"active": True,
"message": "Test",
}
cog.scheduled_tasks.append(job)
# Mock the execute_task method
with patch.object(cog, 'execute_task', new_callable=AsyncMock) as mock_exec:
# Mock should_run_task to return True
with patch.object(cog, 'should_run_task', return_value=True):
# Test single iteration of schedule_checker logic
for task in cog.scheduled_tasks:
if task.get("active", True) and cog.should_run_task(task, None):
await cog.execute_task(task)
mock_exec.assert_called_once_with(job)
|
