Backend Development 12 min read

Integrating DjCelery (Django + Celery) for Automated Test Task Scheduling in the Autotest Platform

This guide explains how to set up DjCelery—combining Django and Celery—to create, schedule, and manage automated testing tasks such as single‑interface, API, App, and Web UI scripts, replacing Jenkins‑based cron jobs with a fully integrated backend solution.

360 Quality & Efficiency
360 Quality & Efficiency
360 Quality & Efficiency
Integrating DjCelery (Django + Celery) for Automated Test Task Scheduling in the Autotest Platform

The chapter introduces the use of DjCelery (the Django + Celery framework) to develop timed‑task functionality for the Autotest platform, enabling automated execution of single‑interface, API, App, and Web UI test scripts and replacing Jenkins‑based scheduled scripts.

11.1 Environment Setup

Installation steps

1. Install Celery (pyramid_celery‑3.0.0) and configure it from https://pypi.python.org/pypi/pyramid_celery/ .

2. Install django‑celery (django‑celery‑3.2.2) and configure it from https://pypi.python.org/pypi/django-celery .

3. Add 'djcelery', to INSTALLED_APPS = [] and run python manage.py migrate .

4. Install celery‑with‑redis‑3.0 ( https://pypi.python.org/pypi/celery-with-redis/ ).

5. Install django‑celery‑beat‑1.1.0 ( https://pypi.python.org/pypi/django_celery_beat ).

6. Download and unzip Redis‑x64‑3.2.100 from https://github.com/MicrosoftArchive/redis/releases .

2. Configuration

In settings.py add:

import djcelery
djcelery.setup_loader()  # load djcelery
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
BROKER_URL = 'redis://127.0.0.1:6379/0'
BROKER_TRANSPORT = 'redis'

Create apitest/celery.py :

from __future__ import absolute_import
import os, django
from celery import Celery
from django.conf import settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'autotest.settings')
django.setup()
app = Celery('autotest')
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

Create apitest/tasks.py with a sample task:

# -*-coding:utf-8 -*-
from djcelery import app
@app.task
def hello_world():
    print('已运行')

Run the services:

Start Django development server: python manage.py runserver

Start Redis server: redis-server redis.windows.conf

Start Celery worker: python manage.py celery worker -l info

Start Celery beat scheduler: python manage.py celery beat

11.2 Front‑end Function Implementation

1. Description – implement periodic tasks for single‑interface, API, AppUI, and WebUI test cases.

2. Template – create periodic_task.html under autotest/apitest/templates containing the necessary Bootstrap imports, navigation bar, search form, and a table that lists tasks with columns for ID, name, module, schedule, last modified, enable flag, run link, edit, and delete actions. The table uses Django template tags to iterate over tasks , periodics , and crontabs .

Pagination controls are added at the bottom of the page using Django’s Paginator object, with logic for previous/next links and page number links.

Additional view functions in apitest/views.py :

from .tasks import hello_world, test_readSQLcase
from djcelery.models import PeriodicTask, CrontabSchedule, IntervalSchedule
from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger

@login_required
def periodic_task(request):
    username = request.session.get('user', '')
    task_list = PeriodicTask.objects.all()
    task_count = task_list.count()
    periodic_list = IntervalSchedule.objects.all()
    crontab_list = CrontabSchedule.objects.all()
    paginator = Paginator(task_list, 5)
    page = request.GET.get('page', 1)
    try:
        task_page = paginator.page(page)
    except PageNotAnInteger:
        task_page = paginator.page(1)
    except EmptyPage:
        task_page = paginator.page(paginator.num_pages)
    return render(request, "periodic_task.html", {
        "user": username,
        "tasks": task_page,
        "taskcounts": task_count,
        "periodics": periodic_list,
        "crontabs": crontab_list,
    })

@login_required
def tasksearch(request):
    username = request.session.get('user', '')
    search_name = request.GET.get('task', '')
    task_list = PeriodicTask.objects.filter(task__icontains=search_name)
    periodic_list = IntervalSchedule.objects.all()
    crontab_list = CrontabSchedule.objects.all()
    return render(request, 'periodic_task.html', {
        "user": username,
        "tasks": task_list,
        "periodics": periodic_list,
        "crontabs": crontab_list,
    })

URL patterns added in autotest/urls.py :

path('periodic_task/', views.periodic_task),
path('tasksearch/', views.tasksearch),

Navigation link added to apitest/left.html to access the task schedule page.

Front‑end page screenshots (Figures 11.1‑11.3) illustrate the workflow diagram, the Redis startup window, and the final task‑list UI.

The content is extracted from the book “Automated Platform Test Development – Python Test Development in Practice”.

Backend Developmenttask schedulingDjangoCeleryAutomation Testing
360 Quality & Efficiency
Written by

360 Quality & Efficiency

360 Quality & Efficiency focuses on seamlessly integrating quality and efficiency in R&D, sharing 360’s internal best practices with industry peers to foster collaboration among Chinese enterprises and drive greater efficiency value.

0 followers
Reader feedback

How this landed with the community

login Sign in to like

Rate this article

Was this worth your time?

Sign in to rate
Discussion

0 Comments

Thoughtful readers leave field notes, pushback, and hard-won operational detail here.