aboutsummaryrefslogtreecommitdiffstats
path: root/tests/cogs/test_cron.py
diff options
context:
space:
mode:
authorsalmonik <raisersoft@proton.me>2026-07-08 02:13:35 +0200
committersalmonik <raisersoft@proton.me>2026-07-08 02:13:35 +0200
commitcd8ed8bc5793985df2c1692072a932ceee517b95 (patch)
tree42d18a4140259b5c68b4dd05b7fcb398f78b8d7f /tests/cogs/test_cron.py
parentMerge pull request 'Voice related fixes, alongside simplifying assistant' (#5... (diff)
downloadsenator-sperma-cd8ed8bc5793985df2c1692072a932ceee517b95.tar.gz
senator-sperma-cd8ed8bc5793985df2c1692072a932ceee517b95.zip
OPEN THE RUST GATES
Signed-off-by: salmonik <raisersoft@proton.me>
Diffstat (limited to 'tests/cogs/test_cron.py')
-rw-r--r--tests/cogs/test_cron.py147
1 files changed, 0 insertions, 147 deletions
diff --git a/tests/cogs/test_cron.py b/tests/cogs/test_cron.py
deleted file mode 100644
index 28ccfe1..0000000
--- a/tests/cogs/test_cron.py
+++ /dev/null
@@ -1,147 +0,0 @@
1import pytest
2from datetime import datetime
3from unittest.mock import MagicMock, patch
4
5from cogs.cron import Cron
6from cogs.utils.database import _ConnectionContextManager
7
8
9@pytest.fixture
10def mock_bot():
11 bot = MagicMock()
12 bot.loop = MagicMock()
13 bot.is_closed = MagicMock(return_value=False)
14 bot.wait_until_ready = MagicMock()
15 bot.loop.create_task = MagicMock(return_value=MagicMock())
16 return bot
17
18
19def create_cog(mock_bot):
20 with patch.object(Cron, "__init__", lambda self, bot: None):
21 with patch("cogs.cron.init_database"):
22 cog = Cron.__new__(Cron)
23 cog.bot = mock_bot
24 cog.scheduled_tasks = []
25 cog.task = MagicMock()
26 return cog
27
28
29class TestLoadTasks:
30 def test_load_tasks_with_data(self, mock_bot):
31 mock_conn = MagicMock()
32 mock_cursor = MagicMock()
33 mock_cursor.fetchall.return_value = [
34 (1, "* * * * *", "audio", 123, 1, None, "test message")
35 ]
36 mock_conn.cursor.return_value = mock_cursor
37
38 # Create a context manager mock
39 mock_ctx = MagicMock()
40 mock_ctx.__enter__ = MagicMock(return_value=mock_conn)
41 mock_ctx.__exit__ = MagicMock(return_value=False)
42
43 with patch("cogs.cron.get_connection", return_value=mock_ctx):
44 cog = Cron.__new__(Cron)
45 cog.bot = mock_bot
46 cog.scheduled_tasks = []
47 cog.task = MagicMock()
48 cog.load_tasks()
49 assert len(cog.scheduled_tasks) == 1
50 assert cog.scheduled_tasks[0]["id"] == 1
51
52class TestSaveTasks:
53 def test_save_tasks(self, mock_bot):
54 cog = create_cog(mock_bot)
55 cog.scheduled_tasks = [{"id": 1, "cron_expression": "* * * * *", "type": "audio", "channel_id": 123, "active": True}]
56
57 mock_conn = MagicMock()
58 mock_cursor = MagicMock()
59 mock_conn.cursor.return_value = mock_cursor
60
61 # Create a context manager mock
62 mock_ctx = MagicMock()
63 mock_ctx.__enter__ = MagicMock(return_value=mock_conn)
64 mock_ctx.__exit__ = MagicMock(return_value=False)
65
66 with patch("cogs.cron.get_connection", return_value=mock_ctx):
67 cog.save_tasks()
68 assert len(cog.scheduled_tasks) == 1
69
70class TestShouldRunTask:
71 def test_should_run_task_invalid_cron(self, mock_bot):
72 cog = create_cog(mock_bot)
73 task = {"cron_expression": "invalid"}
74 now = datetime.now()
75 result = cog.should_run_task(task, now)
76 assert result is False
77
78 def test_should_run_task_missing_cron_expression(self, mock_bot):
79 cog = create_cog(mock_bot)
80 task = {}
81 now = datetime.now()
82 result = cog.should_run_task(task, now)
83 assert result is False
84
85class TestCogUnload:
86 def test_cog_unload_cancels_task(self, mock_bot):
87 with patch.object(Cron, "__init__", lambda self, bot: None):
88 cog = Cron.__new__(Cron)
89 cog.bot = mock_bot
90 cog.task = MagicMock()
91
92 cog.cog_unload()
93 cog.task.cancel.assert_called_once()
94
95class TestTaskManagement:
96 def test_add_task_increments_id(self, mock_bot):
97 cog = create_cog(mock_bot)
98 cog.scheduled_tasks = [
99 {"id": 1, "cron_expression": "* * * * *"},
100 {"id": 2, "cron_expression": "0 * * * *"},
101 ]
102
103 new_id = max([t["id"] for t in cog.scheduled_tasks], default=0) + 1
104 assert new_id == 3
105
106 def test_find_task_by_id(self, mock_bot):
107 cog = create_cog(mock_bot)
108 cog.scheduled_tasks = [
109 {"id": 1, "cron_expression": "* * * * *"},
110 {"id": 2, "cron_expression": "0 * * * *"},
111 ]
112
113 found_task = next((t for t in cog.scheduled_tasks if t["id"] == 2), None)
114 assert found_task is not None
115 assert found_task["cron_expression"] == "0 * * * *"
116
117 def test_remove_task_by_index(self, mock_bot):
118 cog = create_cog(mock_bot)
119 cog.scheduled_tasks = [
120 {"id": 1, "cron_expression": "* * * * *"},
121 {"id": 2, "cron_expression": "0 * * * *"},
122 ]
123
124 cog.scheduled_tasks.pop(0)
125 assert len(cog.scheduled_tasks) == 1
126 assert cog.scheduled_tasks[0]["id"] == 2
127
128 def test_toggle_task_active(self, mock_bot):
129 cog = create_cog(mock_bot)
130 cog.scheduled_tasks = [{"id": 1, "active": True}]
131
132 cog.scheduled_tasks[0]["active"] = False
133 assert cog.scheduled_tasks[0]["active"] is False
134
135 cog.scheduled_tasks[0]["active"] = True
136 assert cog.scheduled_tasks[0]["active"] is True
137
138class TestTaskTypes:
139 def test_task_type_audio_aliases(self, mock_bot):
140 audio_types = ("zagraj_audio", "audio")
141 for task_type in audio_types:
142 assert task_type in ("zagraj_audio", "audio")
143
144 def test_task_type_message_aliases(self, mock_bot):
145 message_types = ("wyślij_wiadomość", "send_message")
146 for task_type in message_types:
147 assert task_type in ("wyślij_wiadomość", "send_message")