diff options
| author | salmonik <raisersoft@proton.me> | 2026-07-08 02:13:35 +0200 |
|---|---|---|
| committer | salmonik <raisersoft@proton.me> | 2026-07-08 02:13:35 +0200 |
| commit | cd8ed8bc5793985df2c1692072a932ceee517b95 (patch) | |
| tree | 42d18a4140259b5c68b4dd05b7fcb398f78b8d7f /tests/cogs/test_cron.py | |
| parent | Merge pull request 'Voice related fixes, alongside simplifying assistant' (#5... (diff) | |
| download | senator-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.py | 147 |
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 @@ | |||
| 1 | import pytest | ||
| 2 | from datetime import datetime | ||
| 3 | from unittest.mock import MagicMock, patch | ||
| 4 | |||
| 5 | from cogs.cron import Cron | ||
| 6 | from cogs.utils.database import _ConnectionContextManager | ||
| 7 | |||
| 8 | |||
| 9 | @pytest.fixture | ||
| 10 | def 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 | |||
| 19 | def 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 | |||
| 29 | class 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 | |||
| 52 | class 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 | |||
| 70 | class 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 | |||
| 85 | class 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 | |||
| 95 | class 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 | |||
| 138 | class 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") | ||
