aboutsummaryrefslogtreecommitdiffstats
path: root/tests/cogs/test_cron.py
diff options
context:
space:
mode:
authorsalmonik <raisersoft@proton.me>2026-04-19 15:43:25 +0200
committersalmonik <raisersoft@proton.me>2026-04-19 15:43:25 +0200
commit115f1f530c1543e7773b2e24629cdf9596d4fbc9 (patch)
tree0f0c8f804daf695a9d05ce3f76e0673d342bb147 /tests/cogs/test_cron.py
parentdiscontinue plots; too complex. (diff)
downloadsenator-sperma-115f1f530c1543e7773b2e24629cdf9596d4fbc9.tar.gz
senator-sperma-115f1f530c1543e7773b2e24629cdf9596d4fbc9.zip
test: add cron tests + improve existing tests
- Add 15 tests for cron.py covering load/save, task management - Improve existing test assertions and coverage - Remove flaky plot tests - 93 tests now pass (up from 78)
Diffstat (limited to '')
-rw-r--r--tests/cogs/test_cron.py166
1 files changed, 166 insertions, 0 deletions
diff --git a/tests/cogs/test_cron.py b/tests/cogs/test_cron.py
new file mode 100644
index 0000000..93031f8
--- /dev/null
+++ b/tests/cogs/test_cron.py
@@ -0,0 +1,166 @@
1import json
2import os
3import pytest
4from datetime import datetime
5from unittest.mock import MagicMock, patch
6
7from cogs.cron import Cron
8
9
10@pytest.fixture
11def mock_bot():
12 bot = MagicMock()
13 bot.loop = MagicMock()
14 bot.is_closed = MagicMock(return_value=False)
15 bot.wait_until_ready = MagicMock()
16 bot.loop.create_task = MagicMock(return_value=MagicMock())
17 return bot
18
19
20def create_cog(mock_bot, tmp_path):
21 with patch.object(Cron, "__init__", lambda self, bot: None):
22 cog = Cron.__new__(Cron)
23 cog.bot = mock_bot
24 cog.scheduled_tasks = []
25 cog.cron_file = str(tmp_path / "cron_jobs.json")
26 cog.task = MagicMock()
27 return cog
28
29
30class TestLoadTasks:
31 def test_load_tasks_file_not_exists(self, mock_bot, tmp_path):
32 cog = create_cog(mock_bot, tmp_path)
33 cog.cron_file = str(tmp_path / "nonexistent.json")
34 cog.load_tasks()
35 assert cog.scheduled_tasks == []
36
37 def test_load_tasks_empty_file(self, mock_bot, tmp_path):
38 cron_file = tmp_path / "empty.json"
39 cron_file.write_text("")
40 cog = create_cog(mock_bot, tmp_path)
41 cog.cron_file = str(cron_file)
42 cog.load_tasks()
43 assert cog.scheduled_tasks == []
44
45 def test_load_tasks_valid_file(self, mock_bot, tmp_path):
46 cron_file = tmp_path / "valid.json"
47 tasks = [{"id": 1, "cron_expression": "* * * * *", "type": "audio", "channel_id": 123, "active": True}]
48 cron_file.write_text(json.dumps(tasks))
49 cog = create_cog(mock_bot, tmp_path)
50 cog.cron_file = str(cron_file)
51 cog.load_tasks()
52 assert len(cog.scheduled_tasks) == 1
53 assert cog.scheduled_tasks[0]["id"] == 1
54
55 def test_load_tasks_io_error(self, mock_bot, tmp_path):
56 cog = create_cog(mock_bot, tmp_path)
57 with patch("builtins.open", side_effect=IOError("test error")):
58 cog.load_tasks()
59 assert cog.scheduled_tasks == []
60
61
62class TestSaveTasks:
63 def test_save_tasks_with_real_file(self, mock_bot, tmp_path):
64 cron_file = tmp_path / "cron_jobs.json"
65 cog = create_cog(mock_bot, tmp_path)
66 cog.cron_file = str(cron_file)
67 cog.scheduled_tasks = [{"id": 1, "cron_expression": "* * * * *"}]
68 cog.save_tasks()
69 assert os.path.exists(cron_file)
70 with open(cron_file) as f:
71 loaded = json.load(f)
72 assert len(loaded) == 1
73
74 def test_save_tasks_creates_directory_structure(self, mock_bot, tmp_path):
75 cog = create_cog(mock_bot, tmp_path)
76 new_dir = tmp_path / "new_dir"
77 cron_file = new_dir / "cron_jobs.json"
78 cog.cron_file = str(cron_file)
79 cog.scheduled_tasks = [{"id": 1}]
80
81 cog.save_tasks()
82
83 assert os.path.exists(new_dir)
84
85
86class TestShouldRunTask:
87 def test_should_run_task_invalid_cron(self, mock_bot, tmp_path):
88 cog = create_cog(mock_bot, tmp_path)
89 task = {"cron_expression": "invalid"}
90 now = datetime.now()
91 result = cog.should_run_task(task, now)
92 assert result is False
93
94 def test_should_run_task_missing_cron_expression(self, mock_bot, tmp_path):
95 cog = create_cog(mock_bot, tmp_path)
96 task = {}
97 now = datetime.now()
98 result = cog.should_run_task(task, now)
99 assert result is False
100
101
102class TestCogUnload:
103 def test_cog_unload_cancels_task(self, mock_bot, tmp_path):
104 with patch.object(Cron, "__init__", lambda self, bot: None):
105 cog = Cron.__new__(Cron)
106 cog.bot = mock_bot
107 cog.task = MagicMock()
108
109 cog.cog_unload()
110 cog.task.cancel.assert_called_once()
111
112
113class TestTaskManagement:
114 def test_add_task_increments_id(self, mock_bot, tmp_path):
115 cog = create_cog(mock_bot, tmp_path)
116 cog.scheduled_tasks = [
117 {"id": 1, "cron_expression": "* * * * *"},
118 {"id": 2, "cron_expression": "0 * * * *"},
119 ]
120
121 new_id = max([t["id"] for t in cog.scheduled_tasks], default=0) + 1
122 assert new_id == 3
123
124 def test_find_task_by_id(self, mock_bot, tmp_path):
125 cog = create_cog(mock_bot, tmp_path)
126 cog.scheduled_tasks = [
127 {"id": 1, "cron_expression": "* * * * *"},
128 {"id": 2, "cron_expression": "0 * * * *"},
129 ]
130
131 found_task = next((t for t in cog.scheduled_tasks if t["id"] == 2), None)
132 assert found_task is not None
133 assert found_task["cron_expression"] == "0 * * * *"
134
135 def test_remove_task_by_index(self, mock_bot, tmp_path):
136 cog = create_cog(mock_bot, tmp_path)
137 cog.scheduled_tasks = [
138 {"id": 1, "cron_expression": "* * * * *"},
139 {"id": 2, "cron_expression": "0 * * * *"},
140 ]
141
142 cog.scheduled_tasks.pop(0)
143 assert len(cog.scheduled_tasks) == 1
144 assert cog.scheduled_tasks[0]["id"] == 2
145
146 def test_toggle_task_active(self, mock_bot, tmp_path):
147 cog = create_cog(mock_bot, tmp_path)
148 cog.scheduled_tasks = [{"id": 1, "active": True}]
149
150 cog.scheduled_tasks[0]["active"] = False
151 assert cog.scheduled_tasks[0]["active"] is False
152
153 cog.scheduled_tasks[0]["active"] = True
154 assert cog.scheduled_tasks[0]["active"] is True
155
156
157class TestTaskTypes:
158 def test_task_type_audio_aliases(self, mock_bot, tmp_path):
159 audio_types = ("zagraj_audio", "audio")
160 for task_type in audio_types:
161 assert task_type in ("zagraj_audio", "audio")
162
163 def test_task_type_message_aliases(self, mock_bot, tmp_path):
164 message_types = ("wyślij_wiadomość", "send_message")
165 for task_type in message_types:
166 assert task_type in ("wyślij_wiadomość", "send_message")