|
| 1 | +import io |
| 2 | +import unittest |
| 3 | +import contextlib |
| 4 | +import urllib.parse |
| 5 | +from unittest import mock |
| 6 | + |
| 7 | +from django.http import QueryDict |
| 8 | +from django.urls import resolve |
| 9 | + |
| 10 | +from django_lightweight_queue import task |
| 11 | +from django_lightweight_queue.job import Job |
| 12 | +from django_lightweight_queue.types import QueueName |
| 13 | +from django_lightweight_queue.backends.debug_web import DebugWebBackend |
| 14 | + |
| 15 | + |
| 16 | +@task() |
| 17 | +def demo_task() -> None: |
| 18 | + pass |
| 19 | + |
| 20 | + |
| 21 | +class DebugWebBackendTests(unittest.TestCase): |
| 22 | + def test_enqueue_prints_valid_url(self) -> None: |
| 23 | + backend = DebugWebBackend() |
| 24 | + |
| 25 | + job = Job('tests.test_debug_web_backend.demo_task', ('positional',), {'keyword': '&arg='}) |
| 26 | + |
| 27 | + with mock.patch('tests.test_debug_web_backend.demo_task') as demo_task_mock: |
| 28 | + with contextlib.redirect_stdout(io.StringIO()) as mock_stdout: |
| 29 | + backend.enqueue(job, QueueName('test-queue')) |
| 30 | + |
| 31 | + url = mock_stdout.getvalue().strip() |
| 32 | + parse_result = urllib.parse.urlparse(url) |
| 33 | + |
| 34 | + match = resolve(parse_result.path) |
| 35 | + self.assertIsNotNone(match, f"Failed to match {parse_result.path}") |
| 36 | + |
| 37 | + query = QueryDict(parse_result.query) |
| 38 | + |
| 39 | + self.assertEqual( |
| 40 | + {'job': [job.to_json()]}, |
| 41 | + dict(query), |
| 42 | + "Wrong query arguments printed", |
| 43 | + ) |
| 44 | + |
| 45 | + demo_task_mock.assert_not_called() |
0 commit comments