add page specific tests and github actions setup for testing with celery

Author: Yasser Toruno

.github/workflows/test.yml

@@ -45,16 +45,54 @@ jobs:
             make install
             pip install "Django~=${{ matrix.django-version }}.0"
 
+      - name: Run database migrations
+        run: |
+            cd example_project
+            python manage.py migrate --noinput
+        env:
+          DB_ENGINE: postgresql
+          POSTGRES_HOST: localhost
+          POSTGRES_PORT: 5432
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: postgres
+          CELERY_BROKER_URL: redis://localhost:6379/0
+
+      - name: Start Celery worker
+        run: |
+            cd example_project
+            celery -A example_project worker --loglevel=info --detach --logfile=celery_worker.log
+        env:
+          DB_ENGINE: postgresql
+          POSTGRES_HOST: localhost
+          POSTGRES_PORT: 5432
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: postgres
+          CELERY_BROKER_URL: redis://localhost:6379/0
+
+      - name: Wait for worker to be ready
+        run: |
+            echo "Waiting for Celery worker to register..."
+            sleep 5
+
       - name: Run tests with coverage
         run: |
-            pytest --cov=dj_cache_panel --cov-report=xml
+            pytest --cov=dj_celery_panel --cov-report=xml
         env:
           TEST_DB_BACKEND: postgresql
           POSTGRES_HOST: localhost
           POSTGRES_PORT: 5432
           POSTGRES_USER: postgres
           POSTGRES_PASSWORD: postgres
           POSTGRES_DB: postgres
+          CELERY_BROKER_URL: redis://localhost:6379/0
+
+      - name: Show Celery worker logs on failure
+        if: failure()
+        run: |
+            echo "=== Celery Worker Logs ==="
+            cat example_project/celery_worker.log || echo "No worker log file found"
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

example_project/example_project/settings.py

@@ -166,7 +166,8 @@
 # Celery Configuration #
 #########################
 # Celery Configuration (using modern Celery 4.x+ naming)
-CELERY_BROKER_URL = "redis://redis:6379/0"
+# Use environment variable for broker URL (defaults to Docker Compose hostname)
+CELERY_BROKER_URL = os.environ.get("CELERY_BROKER_URL", "redis://redis:6379/0")
 
 CELERY_BROKER_TRANSPORT_OPTIONS = {
     "visibility_timeout": 3600,

tests/conftest.py

@@ -35,6 +35,9 @@ def pytest_configure(config):
     else:
         os.environ.setdefault("DB_ENGINE", "sqlite")
 
+    # Set Celery broker URL (defaults to localhost for tests)
+    os.environ.setdefault("CELERY_BROKER_URL", "redis://localhost:6379/0")
+
     os.environ.setdefault("DJANGO_SETTINGS_MODULE", "example_project.settings")
 
     if not settings.configured:

tests/test_configuration.py

@@ -0,0 +1,81 @@
+"""
+Tests for the configuration page.
+"""
+
+from django.urls import reverse
+
+from .base import CeleryPanelTestCase
+
+
+class TestConfigurationPage(CeleryPanelTestCase):
+    """Test cases for the configuration page."""
+
+    def test_configuration_page_loads(self):
+        """Test that the configuration page loads successfully."""
+        response = self.client.get(reverse("dj_celery_panel:configuration"))
+        
+        self.assertEqual(response.status_code, 200)
+
+    def test_configuration_shows_panel_settings(self):
+        """Test that the configuration page shows DJ Celery Panel settings."""
+        response = self.client.get(reverse("dj_celery_panel:configuration"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "DJ Celery Panel Settings")
+        self.assertContains(response, "Backend Configuration")
+
+    def test_configuration_shows_celery_settings(self):
+        """Test that the configuration page shows Celery settings."""
+        response = self.client.get(reverse("dj_celery_panel:configuration"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Celery Settings")
+        self.assertContains(response, "Connection & Serialization")
+
+    def test_configuration_shows_task_execution_settings(self):
+        """Test that the configuration page shows task execution settings."""
+        response = self.client.get(reverse("dj_celery_panel:configuration"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Task Execution")
+
+    def test_configuration_shows_queue_routing_settings(self):
+        """Test that the configuration page shows queue and routing settings."""
+        response = self.client.get(reverse("dj_celery_panel:configuration"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Queue & Routing")
+
+    def test_configuration_shows_worker_settings(self):
+        """Test that the configuration page shows worker settings."""
+        response = self.client.get(reverse("dj_celery_panel:configuration"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Worker Settings")
+
+    def test_configuration_requires_authentication(self):
+        """Test that unauthenticated users cannot access the configuration page."""
+        from django.test import Client
+        
+        client = Client()
+        response = client.get(reverse("dj_celery_panel:configuration"))
+        
+        # Should redirect to login
+        self.assertEqual(response.status_code, 302)
+
+    def test_configuration_requires_staff_permission(self):
+        """Test that non-staff users cannot access the configuration page."""
+        from django.test import Client
+        from django.contrib.auth import get_user_model
+        
+        User = get_user_model()
+        user = User.objects.create_user(
+            username="regular_user", password="testpass123", is_staff=False
+        )
+        
+        client = Client()
+        client.force_login(user)
+        response = client.get(reverse("dj_celery_panel:configuration"))
+        
+        # Should redirect to admin login
+        self.assertEqual(response.status_code, 302)

tests/test_index.py

@@ -0,0 +1,68 @@
+"""
+Tests for the Celery Panel index/overview page.
+"""
+
+from django.urls import reverse
+
+from .base import CeleryPanelTestCase
+
+
+class TestIndexPage(CeleryPanelTestCase):
+    """Test cases for the index page."""
+
+    def test_index_page_loads(self):
+        """Test that the index page loads successfully."""
+        response = self.client.get(reverse("dj_celery_panel:index"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Django Celery Panel")
+
+    def test_index_shows_configuration_section(self):
+        """Test that the index page shows configuration information."""
+        response = self.client.get(reverse("dj_celery_panel:index"))
+        
+        self.assertEqual(response.status_code, 200)
+        # Should show broker and result backend info
+        self.assertContains(response, "Broker")
+        self.assertContains(response, "Result Backend")
+
+    def test_index_shows_registered_tasks(self):
+        """Test that the index page shows registered tasks."""
+        response = self.client.get(reverse("dj_celery_panel:index"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Registered Tasks")
+
+    def test_index_shows_periodic_tasks(self):
+        """Test that the index page shows periodic tasks."""
+        response = self.client.get(reverse("dj_celery_panel:index"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Periodic Tasks")
+
+    def test_index_requires_authentication(self):
+        """Test that unauthenticated users cannot access the index page."""
+        from django.test import Client
+        
+        client = Client()
+        response = client.get(reverse("dj_celery_panel:index"))
+        
+        # Should redirect to login
+        self.assertEqual(response.status_code, 302)
+
+    def test_index_requires_staff_permission(self):
+        """Test that non-staff users cannot access the index page."""
+        from django.test import Client
+        from django.contrib.auth import get_user_model
+        
+        User = get_user_model()
+        user = User.objects.create_user(
+            username="regular_user", password="testpass123", is_staff=False
+        )
+        
+        client = Client()
+        client.force_login(user)
+        response = client.get(reverse("dj_celery_panel:index"))
+        
+        # Should redirect to admin login
+        self.assertEqual(response.status_code, 302)

tests/test_queues.py

@@ -0,0 +1,80 @@
+"""
+Tests for the queues page and queue detail page.
+"""
+
+from django.urls import reverse
+
+from .base import CeleryPanelTestCase
+
+
+class TestQueuesPage(CeleryPanelTestCase):
+    """Test cases for the queues list page."""
+
+    def test_queues_page_loads(self):
+        """Test that the queues page loads successfully."""
+        response = self.client.get(reverse("dj_celery_panel:queues"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Active Task Queues")
+
+    def test_queues_page_shows_table_headers(self):
+        """Test that the queues page shows appropriate table headers."""
+        response = self.client.get(reverse("dj_celery_panel:queues"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Queue Name")
+        self.assertContains(response, "Messages")
+        self.assertContains(response, "Exchange")
+        self.assertContains(response, "Routing Key")
+
+    def test_queues_page_handles_no_queues(self):
+        """Test that the queues page handles case when no queues are active."""
+        response = self.client.get(reverse("dj_celery_panel:queues"))
+        
+        # Should still load successfully even with no queues
+        self.assertEqual(response.status_code, 200)
+
+    def test_queues_requires_authentication(self):
+        """Test that unauthenticated users cannot access the queues page."""
+        from django.test import Client
+        
+        client = Client()
+        response = client.get(reverse("dj_celery_panel:queues"))
+        
+        # Should redirect to login
+        self.assertEqual(response.status_code, 302)
+
+
+class TestQueueDetailPage(CeleryPanelTestCase):
+    """Test cases for the queue detail page."""
+
+    def test_queue_detail_page_loads_with_valid_queue_name(self):
+        """Test that the queue detail page loads with a queue name."""
+        # Using a dummy queue name - the page should load even if queue doesn't exist
+        response = self.client.get(
+            reverse("dj_celery_panel:queue_detail", kwargs={"queue_name": "celery"})
+        )
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Queue Details")
+
+    def test_queue_detail_shows_not_found_for_invalid_queue(self):
+        """Test that the queue detail page shows error for invalid queue."""
+        response = self.client.get(
+            reverse("dj_celery_panel:queue_detail", kwargs={"queue_name": "nonexistent-queue"})
+        )
+        
+        self.assertEqual(response.status_code, 200)
+        # Should show some indication that queue wasn't found
+
+    def test_queue_detail_requires_authentication(self):
+        """Test that unauthenticated users cannot access queue detail."""
+        from django.test import Client
+        
+        client = Client()
+        response = client.get(
+            reverse("dj_celery_panel:queue_detail", kwargs={"queue_name": "celery"})
+        )
+        
+        # Should redirect to login
+        self.assertEqual(response.status_code, 302)

tests/test_tasks.py

@@ -0,0 +1,90 @@
+"""
+Tests for the tasks page and task detail page.
+"""
+
+from django.urls import reverse
+
+from .base import CeleryPanelTestCase
+
+
+class TestTasksPage(CeleryPanelTestCase):
+    """Test cases for the tasks list page."""
+
+    def test_tasks_page_loads(self):
+        """Test that the tasks page loads successfully."""
+        response = self.client.get(reverse("dj_celery_panel:tasks"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "Task Execution History")
+
+    def test_tasks_page_shows_search_bar(self):
+        """Test that the tasks page has a search bar."""
+        response = self.client.get(reverse("dj_celery_panel:tasks"))
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, 'name="search"')
+
+    def test_tasks_page_with_search_query(self):
+        """Test that the tasks page accepts search queries."""
+        response = self.client.get(reverse("dj_celery_panel:tasks"), {"search": "test_task"})
+        
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, 'value="test_task"')
+
+    def test_tasks_page_pagination(self):
+        """Test that the tasks page handles pagination."""
+        response = self.client.get(reverse("dj_celery_panel:tasks"), {"page": "1"})
+        
+        self.assertEqual(response.status_code, 200)
+
+    def test_tasks_page_invalid_page_number(self):
+        """Test that the tasks page handles invalid page numbers gracefully."""
+        response = self.client.get(reverse("dj_celery_panel:tasks"), {"page": "invalid"})
+        
+        # Should default to page 1
+        self.assertEqual(response.status_code, 200)
+
+    def test_tasks_requires_authentication(self):
+        """Test that unauthenticated users cannot access the tasks page."""
+        from django.test import Client
+        
+        client = Client()
+        response = client.get(reverse("dj_celery_panel:tasks"))
+        
+        # Should redirect to login
+        self.assertEqual(response.status_code, 302)
+
+
+class TestTaskDetailPage(CeleryPanelTestCase):
+    """Test cases for the task detail page."""
+
+    def test_task_detail_page_loads_with_valid_task_id(self):
+        """Test that the task detail page loads with a task ID."""
+        # Using a dummy task ID - the page should load even if task doesn't exist
+        response = self.client.get(
+            reverse("dj_celery_panel:task_detail", kwargs={"task_id": "test-task-id"})
+        )
+        
+        self.assertEqual(response.status_code, 200)
+
+    def test_task_detail_shows_not_found_for_invalid_task(self):
+        """Test that the task detail page shows error for invalid task."""
+        response = self.client.get(
+            reverse("dj_celery_panel:task_detail", kwargs={"task_id": "nonexistent-task"})
+        )
+        
+        self.assertEqual(response.status_code, 200)
+        # Should show some indication that task wasn't found
+        # Either in messages or in the page content
+
+    def test_task_detail_requires_authentication(self):
+        """Test that unauthenticated users cannot access task detail."""
+        from django.test import Client
+        
+        client = Client()
+        response = client.get(
+            reverse("dj_celery_panel:task_detail", kwargs={"task_id": "test-task-id"})
+        )
+        
+        # Should redirect to login
+        self.assertEqual(response.status_code, 302)