django-lightweight-queue/tests/test_extra_settings.py at 80d2fd4ab53e1dd60702d8a76fd81bb8c41d8141 · thread/django-lightweight-queue · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from typing import Optional
from pathlib import Path
from unittest import mock

from django.test import SimpleTestCase

from django_lightweight_queue import app_settings
from django_lightweight_queue.job import Job
from django_lightweight_queue.types import QueueName, WorkerNumber
from django_lightweight_queue.utils import get_backend, load_extra_settings
from django_lightweight_queue.backends.base import BaseBackend

TESTS_DIR = Path(__file__).parent


class TestBackend(BaseBackend):
    def enqueue(self, job: Job, queue: QueueName) -> None:
        pass

    def dequeue(self, queue: QueueName, worker_num: WorkerNumber, timeout: int) -> Optional[Job]:
        pass

    def length(self, queue: QueueName) -> int:
        pass


class ExtraSettingsTests(SimpleTestCase):
    def setUp(self) -> None:
        get_backend.cache_clear()

        self.settings: app_settings.Settings = app_settings.AppSettings([app_settings.Defaults()])
        self._settings_patch = mock.patch(
            'django_lightweight_queue.utils.app_settings',
            new=self.settings,
        )
        self._settings_patch.start()

        super().setUp()

    def tearDown(self) -> None:
        get_backend.cache_clear()
        self._settings_patch.stop()
        super().tearDown()

    def test_updates_settings(self) -> None:
        load_extra_settings(str(TESTS_DIR / '_demo_extra_settings.py'))

        backend = get_backend('test-queue')
        self.assertIsInstance(backend, TestBackend)

        self.assertEqual('a very bad password', self.settings.REDIS_PASSWORD)

    def test_warns_about_unexpected_settings(self) -> None:
        with self.assertWarnsRegex(Warning, r'Ignoring unexpected setting.+\bNOT_REDIS_PASSWORD\b'):
            load_extra_settings(str(TESTS_DIR / '_demo_extra_settings_unexpected.py'))

        self.assertEqual('expected', self.settings.REDIS_PASSWORD)

    def test_updates_settings_with_falsey_values(self) -> None:
        load_extra_settings(str(TESTS_DIR / '_demo_extra_settings.py'))
        load_extra_settings(str(TESTS_DIR / '_demo_extra_settings_falsey.py'))

        self.assertIsNone(self.settings.REDIS_PASSWORD)
        self.assertFalse(self.settings.ATOMIC_JOBS)

    def test_rejects_missing_file(self) -> None:
        with self.assertRaises(FileNotFoundError):
            load_extra_settings(str(TESTS_DIR / '_no_such_file.py'))