- Add fetcher/tests/ with 21 tests covering all bug fixes:
- test_utils: get_with_protocol URL normalization (5 tests)
- test_config: scheduled_tasks.json PK uniqueness, Celery queue
settings (7 tests)
- test_publish: URL routing and view for /task/publish_<id>/ (4 tests)
- test_template_field: url_detail.html renders images_url field (2 tests)
- test_search_postprocess: URL host filtering and param stripping (3 tests)
- Replace empty fetcher/tests.py with tests/ package
- Add .gitea/workflows/test.yml: runs on PRs and pushes to main,
installs dependencies and runs `manage.py test fetcher.tests`
This commit is contained in:
1 parent
3a33697ca3
commit
d18eda2fc4
8 files changed
+298
-3
No files matched your search
@@ -0,0 +1,68 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
from django.conf import settings
|
||||
from django.test import SimpleTestCase
|
||||
|
||||
SCHEDULED_TASKS_PATH = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
||||
"scheduled_tasks.json",
|
||||
)
|
||||
|
||||
|
||||
class ScheduledTasksFixtureTests(SimpleTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
with open(SCHEDULED_TASKS_PATH) as f:
|
||||
cls.data = json.load(f)
|
||||
cls.periodic_tasks = [
|
||||
e
|
||||
for e in cls.data
|
||||
if e.get("model") == "django_celery_beat.periodictask"
|
||||
]
|
||||
|
||||
def test_json_is_valid_and_loads(self):
|
||||
self.assertIsInstance(self.data, list)
|
||||
self.assertTrue(len(self.data) > 0)
|
||||
|
||||
def test_periodic_task_primary_keys_are_unique(self):
|
||||
pks = [e["pk"] for e in self.periodic_tasks]
|
||||
self.assertEqual(len(pks), len(set(pks)))
|
||||
|
||||
def test_notify_status_has_unique_pk(self):
|
||||
notify = [
|
||||
e
|
||||
for e in self.periodic_tasks
|
||||
if e["fields"].get("task") == "fetcher.tasks.notify_status"
|
||||
]
|
||||
self.assertEqual(len(notify), 1)
|
||||
other_pks = {
|
||||
e["pk"]
|
||||
for e in self.periodic_tasks
|
||||
if e["fields"].get("task") != "fetcher.tasks.notify_status"
|
||||
}
|
||||
self.assertNotIn(notify[0]["pk"], other_pks)
|
||||
|
||||
def test_interval_schedule_primary_keys_are_unique(self):
|
||||
schedules = [e for e in self.data if e["model"] == "django_celery_beat.intervalschedule"]
|
||||
pks = [e["pk"] for e in schedules]
|
||||
self.assertEqual(len(pks), len(set(pks)))
|
||||
|
||||
def test_expected_critical_tasks_present(self):
|
||||
tasks = {e["fields"].get("task") for e in self.periodic_tasks}
|
||||
self.assertIn("fetcher.tasks.process_raw_urls", tasks)
|
||||
self.assertIn("fetcher.tasks.fetch_feeds", tasks)
|
||||
self.assertIn("fetcher.tasks.notify_status", tasks)
|
||||
|
||||
|
||||
class CeleryQueueSettingsTests(SimpleTestCase):
|
||||
def test_contains_worker_queues(self):
|
||||
queues = [q.name for q in settings.CELERY_TASK_QUEUES]
|
||||
self.assertIn("default", queues)
|
||||
self.assertIn("light", queues)
|
||||
self.assertIn("heavy", queues)
|
||||
|
||||
def test_does_not_contain_deprecated_low_queue(self):
|
||||
queues = [q.name for q in settings.CELERY_TASK_QUEUES]
|
||||
self.assertNotIn("low", queues)
|
||||
Reference in new issue
Block a user