diff options
Diffstat (limited to '')
| -rw-r--r-- | tests/cogs/test_cron.py | 88 |
1 files changed, 33 insertions, 55 deletions
diff --git a/tests/cogs/test_cron.py b/tests/cogs/test_cron.py index 93031f8..91fcb82 100644 --- a/tests/cogs/test_cron.py +++ b/tests/cogs/test_cron.py | |||
| @@ -1,5 +1,3 @@ | |||
| 1 | import json | ||
| 2 | import os | ||
| 3 | import pytest | 1 | import pytest |
| 4 | from datetime import datetime | 2 | from datetime import datetime |
| 5 | from unittest.mock import MagicMock, patch | 3 | from unittest.mock import MagicMock, patch |
| @@ -19,68 +17,48 @@ def mock_bot(): | |||
| 19 | 17 | ||
| 20 | def create_cog(mock_bot, tmp_path): | 18 | def create_cog(mock_bot, tmp_path): |
| 21 | with patch.object(Cron, "__init__", lambda self, bot: None): | 19 | with patch.object(Cron, "__init__", lambda self, bot: None): |
| 22 | cog = Cron.__new__(Cron) | 20 | with patch("cogs.cron.init_database"): |
| 23 | cog.bot = mock_bot | 21 | mock_conn = MagicMock() |
| 24 | cog.scheduled_tasks = [] | 22 | mock_cursor = MagicMock() |
| 25 | cog.cron_file = str(tmp_path / "cron_jobs.json") | 23 | mock_cursor.fetchall.return_value = [] |
| 26 | cog.task = MagicMock() | 24 | mock_conn.cursor.return_value = mock_cursor |
| 27 | return cog | 25 | mock_conn.close.return_value = None |
| 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 | ||
| 28 | 34 | ||
| 29 | 35 | ||
| 30 | class TestLoadTasks: | 36 | class TestLoadTasks: |
| 31 | def test_load_tasks_file_not_exists(self, mock_bot, tmp_path): | 37 | def test_load_tasks_with_data(self, mock_bot, tmp_path): |
| 32 | cog = create_cog(mock_bot, tmp_path) | 38 | mock_conn = MagicMock() |
| 33 | cog.cron_file = str(tmp_path / "nonexistent.json") | 39 | mock_cursor = MagicMock() |
| 34 | cog.load_tasks() | 40 | mock_cursor.fetchall.return_value = [ |
| 35 | assert cog.scheduled_tasks == [] | 41 | (1, "* * * * *", "audio", 123, 1, None, "test message") |
| 36 | 42 | ] | |
| 37 | def test_load_tasks_empty_file(self, mock_bot, tmp_path): | 43 | mock_conn.cursor.return_value = mock_cursor |
| 38 | cron_file = tmp_path / "empty.json" | 44 | mock_conn.close.return_value = None |
| 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 | 45 | ||
| 55 | def test_load_tasks_io_error(self, mock_bot, tmp_path): | 46 | with patch("cogs.cron.get_connection", return_value=mock_conn): |
| 56 | cog = create_cog(mock_bot, tmp_path) | 47 | cog = Cron.__new__(Cron) |
| 57 | with patch("builtins.open", side_effect=IOError("test error")): | 48 | cog.bot = mock_bot |
| 49 | cog.scheduled_tasks = [] | ||
| 50 | cog.task = MagicMock() | ||
| 58 | cog.load_tasks() | 51 | cog.load_tasks() |
| 59 | assert cog.scheduled_tasks == [] | 52 | assert len(cog.scheduled_tasks) == 1 |
| 53 | assert cog.scheduled_tasks[0]["id"] == 1 | ||
| 60 | 54 | ||
| 61 | 55 | ||
| 62 | class TestSaveTasks: | 56 | class TestSaveTasks: |
| 63 | def test_save_tasks_with_real_file(self, mock_bot, tmp_path): | 57 | def test_save_tasks(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) | 58 | cog = create_cog(mock_bot, tmp_path) |
| 76 | new_dir = tmp_path / "new_dir" | 59 | cog.scheduled_tasks = [{"id": 1, "cron_expression": "* * * * *", "type": "audio", "channel_id": 123, "active": True}] |
| 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() | 60 | cog.save_tasks() |
| 82 | 61 | assert len(cog.scheduled_tasks) == 1 | |
| 83 | assert os.path.exists(new_dir) | ||
| 84 | 62 | ||
| 85 | 63 | ||
| 86 | class TestShouldRunTask: | 64 | class TestShouldRunTask: |
| @@ -163,4 +141,4 @@ class TestTaskTypes: | |||
| 163 | def test_task_type_message_aliases(self, mock_bot, tmp_path): | 141 | def test_task_type_message_aliases(self, mock_bot, tmp_path): |
| 164 | message_types = ("wyślij_wiadomość", "send_message") | 142 | message_types = ("wyślij_wiadomość", "send_message") |
| 165 | for task_type in message_types: | 143 | for task_type in message_types: |
| 166 | assert task_type in ("wyślij_wiadomość", "send_message") | 144 | assert task_type in ("wyślij_wiadomość", "send_message") \ No newline at end of file |
