Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions tests/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,3 +5,7 @@
INSTALLED_APPS = [
'django_lightweight_queue',
]

ROOT_URLCONF = 'tests.urls'

SITE_URL = 'http://localhost:8000'
45 changes: 45 additions & 0 deletions tests/test_debug_web_backend.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import io
import unittest
import contextlib
import urllib.parse
from unittest import mock

from django.http import QueryDict
from django.urls import resolve

from django_lightweight_queue import task
from django_lightweight_queue.job import Job
from django_lightweight_queue.types import QueueName
from django_lightweight_queue.backends.debug_web import DebugWebBackend


@task()
def demo_task() -> None:
pass


class DebugWebBackendTests(unittest.TestCase):
def test_enqueue_prints_valid_url(self) -> None:
backend = DebugWebBackend()

job = Job('tests.test_debug_web_backend.demo_task', ('positional',), {'keyword': '&arg='})

with mock.patch('tests.test_debug_web_backend.demo_task') as demo_task_mock:
with contextlib.redirect_stdout(io.StringIO()) as mock_stdout:
backend.enqueue(job, QueueName('test-queue'))

url = mock_stdout.getvalue().strip()
parse_result = urllib.parse.urlparse(url)

match = resolve(parse_result.path)
self.assertIsNotNone(match, f"Failed to match {parse_result.path}")

query = QueryDict(parse_result.query)

self.assertEqual(
{'job': [job.to_json()]},
dict(query),
"Wrong query arguments printed",
)

demo_task_mock.assert_not_called()
5 changes: 5 additions & 0 deletions tests/urls.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
from django.urls import path, include

urlpatterns = [
path('', include('django_lightweight_queue.urls', namespace='django-lightweight-queue')),
]