aboutsummaryrefslogtreecommitdiffstats
path: root/tests/cogs/test_cron.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/cogs/test_cron.py')
-rw-r--r--tests/cogs/test_cron.py85
1 files changed, 44 insertions, 41 deletions
diff --git a/tests/cogs/test_cron.py b/tests/cogs/test_cron.py
index 91fcb82..28ccfe1 100644
--- a/tests/cogs/test_cron.py
+++ b/tests/cogs/test_cron.py
@@ -3,6 +3,7 @@ from datetime import datetime
3from unittest.mock import MagicMock, patch 3from unittest.mock import MagicMock, patch
4 4
5from cogs.cron import Cron 5from cogs.cron import Cron
6from cogs.utils.database import _ConnectionContextManager
6 7
7 8
8@pytest.fixture 9@pytest.fixture
@@ -15,35 +16,31 @@ def mock_bot():
15 return bot 16 return bot
16 17
17 18
18def create_cog(mock_bot, tmp_path): 19def create_cog(mock_bot):
19 with patch.object(Cron, "__init__", lambda self, bot: None): 20 with patch.object(Cron, "__init__", lambda self, bot: None):
20 with patch("cogs.cron.init_database"): 21 with patch("cogs.cron.init_database"):
21 mock_conn = MagicMock() 22 cog = Cron.__new__(Cron)
22 mock_cursor = MagicMock() 23 cog.bot = mock_bot
23 mock_cursor.fetchall.return_value = [] 24 cog.scheduled_tasks = []
24 mock_conn.cursor.return_value = mock_cursor 25 cog.task = MagicMock()
25 mock_conn.close.return_value = None 26 return cog
26
27 with patch("cogs.cron.get_connection", return_value=mock_conn):
28 with patch("cogs.utils.database.get_db_path", return_value=":memory:"):
29 cog = Cron.__new__(Cron)
30 cog.bot = mock_bot
31 cog.scheduled_tasks = []
32 cog.task = MagicMock()
33 return cog
34 27
35 28
36class TestLoadTasks: 29class TestLoadTasks:
37 def test_load_tasks_with_data(self, mock_bot, tmp_path): 30 def test_load_tasks_with_data(self, mock_bot):
38 mock_conn = MagicMock() 31 mock_conn = MagicMock()
39 mock_cursor = MagicMock() 32 mock_cursor = MagicMock()
40 mock_cursor.fetchall.return_value = [ 33 mock_cursor.fetchall.return_value = [
41 (1, "* * * * *", "audio", 123, 1, None, "test message") 34 (1, "* * * * *", "audio", 123, 1, None, "test message")
42 ] 35 ]
43 mock_conn.cursor.return_value = mock_cursor 36 mock_conn.cursor.return_value = mock_cursor
44 mock_conn.close.return_value = None
45 37
46 with patch("cogs.cron.get_connection", return_value=mock_conn): 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):
47 cog = Cron.__new__(Cron) 44 cog = Cron.__new__(Cron)
48 cog.bot = mock_bot 45 cog.bot = mock_bot
49 cog.scheduled_tasks = [] 46 cog.scheduled_tasks = []
@@ -52,33 +49,41 @@ class TestLoadTasks:
52 assert len(cog.scheduled_tasks) == 1 49 assert len(cog.scheduled_tasks) == 1
53 assert cog.scheduled_tasks[0]["id"] == 1 50 assert cog.scheduled_tasks[0]["id"] == 1
54 51
55
56class TestSaveTasks: 52class TestSaveTasks:
57 def test_save_tasks(self, mock_bot, tmp_path): 53 def test_save_tasks(self, mock_bot):
58 cog = create_cog(mock_bot, tmp_path) 54 cog = create_cog(mock_bot)
59 cog.scheduled_tasks = [{"id": 1, "cron_expression": "* * * * *", "type": "audio", "channel_id": 123, "active": True}] 55 cog.scheduled_tasks = [{"id": 1, "cron_expression": "* * * * *", "type": "audio", "channel_id": 123, "active": True}]
60 cog.save_tasks()
61 assert len(cog.scheduled_tasks) == 1
62 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
63 69
64class TestShouldRunTask: 70class TestShouldRunTask:
65 def test_should_run_task_invalid_cron(self, mock_bot, tmp_path): 71 def test_should_run_task_invalid_cron(self, mock_bot):
66 cog = create_cog(mock_bot, tmp_path) 72 cog = create_cog(mock_bot)
67 task = {"cron_expression": "invalid"} 73 task = {"cron_expression": "invalid"}
68 now = datetime.now() 74 now = datetime.now()
69 result = cog.should_run_task(task, now) 75 result = cog.should_run_task(task, now)
70 assert result is False 76 assert result is False
71 77
72 def test_should_run_task_missing_cron_expression(self, mock_bot, tmp_path): 78 def test_should_run_task_missing_cron_expression(self, mock_bot):
73 cog = create_cog(mock_bot, tmp_path) 79 cog = create_cog(mock_bot)
74 task = {} 80 task = {}
75 now = datetime.now() 81 now = datetime.now()
76 result = cog.should_run_task(task, now) 82 result = cog.should_run_task(task, now)
77 assert result is False 83 assert result is False
78 84
79
80class TestCogUnload: 85class TestCogUnload:
81 def test_cog_unload_cancels_task(self, mock_bot, tmp_path): 86 def test_cog_unload_cancels_task(self, mock_bot):
82 with patch.object(Cron, "__init__", lambda self, bot: None): 87 with patch.object(Cron, "__init__", lambda self, bot: None):
83 cog = Cron.__new__(Cron) 88 cog = Cron.__new__(Cron)
84 cog.bot = mock_bot 89 cog.bot = mock_bot
@@ -87,10 +92,9 @@ class TestCogUnload:
87 cog.cog_unload() 92 cog.cog_unload()
88 cog.task.cancel.assert_called_once() 93 cog.task.cancel.assert_called_once()
89 94
90
91class TestTaskManagement: 95class TestTaskManagement:
92 def test_add_task_increments_id(self, mock_bot, tmp_path): 96 def test_add_task_increments_id(self, mock_bot):
93 cog = create_cog(mock_bot, tmp_path) 97 cog = create_cog(mock_bot)
94 cog.scheduled_tasks = [ 98 cog.scheduled_tasks = [
95 {"id": 1, "cron_expression": "* * * * *"}, 99 {"id": 1, "cron_expression": "* * * * *"},
96 {"id": 2, "cron_expression": "0 * * * *"}, 100 {"id": 2, "cron_expression": "0 * * * *"},
@@ -99,8 +103,8 @@ class TestTaskManagement:
99 new_id = max([t["id"] for t in cog.scheduled_tasks], default=0) + 1 103 new_id = max([t["id"] for t in cog.scheduled_tasks], default=0) + 1
100 assert new_id == 3 104 assert new_id == 3
101 105
102 def test_find_task_by_id(self, mock_bot, tmp_path): 106 def test_find_task_by_id(self, mock_bot):
103 cog = create_cog(mock_bot, tmp_path) 107 cog = create_cog(mock_bot)
104 cog.scheduled_tasks = [ 108 cog.scheduled_tasks = [
105 {"id": 1, "cron_expression": "* * * * *"}, 109 {"id": 1, "cron_expression": "* * * * *"},
106 {"id": 2, "cron_expression": "0 * * * *"}, 110 {"id": 2, "cron_expression": "0 * * * *"},
@@ -110,8 +114,8 @@ class TestTaskManagement:
110 assert found_task is not None 114 assert found_task is not None
111 assert found_task["cron_expression"] == "0 * * * *" 115 assert found_task["cron_expression"] == "0 * * * *"
112 116
113 def test_remove_task_by_index(self, mock_bot, tmp_path): 117 def test_remove_task_by_index(self, mock_bot):
114 cog = create_cog(mock_bot, tmp_path) 118 cog = create_cog(mock_bot)
115 cog.scheduled_tasks = [ 119 cog.scheduled_tasks = [
116 {"id": 1, "cron_expression": "* * * * *"}, 120 {"id": 1, "cron_expression": "* * * * *"},
117 {"id": 2, "cron_expression": "0 * * * *"}, 121 {"id": 2, "cron_expression": "0 * * * *"},
@@ -121,8 +125,8 @@ class TestTaskManagement:
121 assert len(cog.scheduled_tasks) == 1 125 assert len(cog.scheduled_tasks) == 1
122 assert cog.scheduled_tasks[0]["id"] == 2 126 assert cog.scheduled_tasks[0]["id"] == 2
123 127
124 def test_toggle_task_active(self, mock_bot, tmp_path): 128 def test_toggle_task_active(self, mock_bot):
125 cog = create_cog(mock_bot, tmp_path) 129 cog = create_cog(mock_bot)
126 cog.scheduled_tasks = [{"id": 1, "active": True}] 130 cog.scheduled_tasks = [{"id": 1, "active": True}]
127 131
128 cog.scheduled_tasks[0]["active"] = False 132 cog.scheduled_tasks[0]["active"] = False
@@ -131,14 +135,13 @@ class TestTaskManagement:
131 cog.scheduled_tasks[0]["active"] = True 135 cog.scheduled_tasks[0]["active"] = True
132 assert cog.scheduled_tasks[0]["active"] is True 136 assert cog.scheduled_tasks[0]["active"] is True
133 137
134
135class TestTaskTypes: 138class TestTaskTypes:
136 def test_task_type_audio_aliases(self, mock_bot, tmp_path): 139 def test_task_type_audio_aliases(self, mock_bot):
137 audio_types = ("zagraj_audio", "audio") 140 audio_types = ("zagraj_audio", "audio")
138 for task_type in audio_types: 141 for task_type in audio_types:
139 assert task_type in ("zagraj_audio", "audio") 142 assert task_type in ("zagraj_audio", "audio")
140 143
141 def test_task_type_message_aliases(self, mock_bot, tmp_path): 144 def test_task_type_message_aliases(self, mock_bot):
142 message_types = ("wyślij_wiadomość", "send_message") 145 message_types = ("wyślij_wiadomość", "send_message")
143 for task_type in message_types: 146 for task_type in message_types:
144 assert task_type in ("wyślij_wiadomość", "send_message") \ No newline at end of file 147 assert task_type in ("wyślij_wiadomość", "send_message")