Luciano Gervasoni
d18eda2fc4
Tests / test (pull_request) Failing after 3m11s
- 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`
69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
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)
|