Django+Celery framework automated scheduled task development

This chapter introduces the use of DjCelery, that is, the Django + Celery framework, to develop the scheduled task function, and implement scheduled execution, scheduling, and management of tasks such as single interface automated test scripts, business scenario interface automated test scripts, App automated test scripts, and Web automated test scripts on the Autotestplat platform. etc., thereby replacing functions such as scheduled execution of scripts and sending emails on Jenkins. **

The automated test logic flow is shown in Figure 11.1.

11.1 Environment setup

1． Installation

Step 1 Install Celery. pyramid_celery-3.0.0,

Configure https://pypi.python.org/pypi/pyramid_celery/.

Step 2 Install django-clery. django-celery-3.2.2,

Configure https://pypi.python.org/pypi/django-celery. INSTALLED_APPS= []

Join 2:

'djcelery', run Python manage.py migrate

Step 3 Install celery-with-redis-3.0,

The address is https://pypi.python.org/pypi/celery-with-redis/.

Step 4 Install django-clery-beat. django-celery-beat-1.1.0,

https://pypi.python.org/pypi/ django_celery_beat.

Step 5 Download Redis-x64-3.2.100,

Redis-x64-3.2.100.zip github.com/MicrosoftAr…

2． Configuration

Step 1 Add the following content to Settings.py.

Join 1:

importdjcelery

djcelery.setup_loader() #Load djcelery

Join 2:

#Database scheduling

CELERYBEAT_SCHEDULER =’djcelery.schedulers.DatabaseScheduler’

Join 3:

BROKER_URL = ‘redis://127.0.0.1:6379/0’

BROKER_TRANSPORT = ‘redis’

Step 2 Create a new celery.py file 1 in the application Apitest directory and add the following content.

from future import absolute_import

import os

import 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)

Step 3 Create a new tasks.py file and add the following content.

–-coding:utf-8 ––

importrequests, time, sys, re

importurllib, zlib#,

importpymysql

importunittest

from traceimport CoverageResults

importjson

fromidlelib.rpc import response_queue

fromapitest.celery import app

from timeimport sleep

@app.task

def hello_world():

print('Running')

Step 4 Start the service python manage.py runserver.

Step 5 After decompressing, run CMD, switch to the corresponding directory, and enter the start Redis command redis-server redis. windows.conf. After success, the interface shown in Figure 11.2 will appear.

▲Figure 11.2

Step 6 Start the command python manage.py celery worker -l info.

Step 7 Start the command python manage.py celery beat.

11.2 Front-end function implementation

1. Function description

Complete automated scheduled tasks for implementing single interface test cases, business scenario interface API test cases, AppUI test cases, and WebUI test cases.

2. Program List

Create a new periodic_task.html file in the autotest\apitest\templates directory and add the following content.

{% load bootstrap4 %}

{% bootstrap_css %}

{% bootstrap_javascript %}

{% csrf_token %}

Search

—-Schedule task—-

Add

th>

{% for task in tasks %}{% for periodic inperiodics %}

{% if task.interval_id != null andtask.interval_id == periodic.id %}

{% else %}

{% endif %}

{% for crontab in crontabs %}

{% if task.crontab_id != null and task.crontab_id ==crontab.id and task.interval_id == 1 %}

{% else %}

{% endif %}

{% endfor %}{% endfor %}{% endfor %}

ID	Task name	Task module	Time plan	Modify time	Enable	Edit Now	Edit	Delete
{{ task.id }}	{{ task.name }}	{{ task.task }}	Every {{ periodic.period }} {{ periodic.every}} times	{{ task.date_changed }}	{{ task.enabled }}	{% if task.id == 1 %} Run {% elif task.id == 2 %} Run {% else %} {% endif %}			{{ task.id }}	{{ task.name }}	{{ task.task }}	{{crontab.month_of_year }}year {{crontab.day_of_month }}month{{crontab.day_of_week }} day {{crontab.hour }} hour {{ crontab.minute}} minute	{{ task.date_changed }}	{{ task.enabled }}	Run

Total number {{ taskcounts }}

{# Front-end reads defined variables#}

 & amp;lt;ulclass="pagination" id="pager" & amp;gt;

      {#Previous page link starts#}

    {%if tasks.has_previous %}

       {# If there is a previous page, display the previous page link normally#}

        & amp;lt;li & amp;gt; & amp;lt;ahref="/periodic_task/?page={<!-- -->{ tasks.previous_page_number }}" & amp;gt;Previous page & amp;lt;/a & amp;gt; & amp;lt;/li & amp;gt; {# Previous page tag#}

    {%else%}

        & amp;lt;li class="previous disabled" & amp;gt; & amp;lt;ahref="#" & amp;gt;Previous page& amp;lt;/a & amp;gt ; & amp;lt;/li & amp;gt;{# If the previous page does not currently exist, the link to the previous page cannot be clicked#}

    {%endif %}

    {# Previous page link starts#}

   

    {%for num in tasks.paginator.page_range %}

    

       {% if num == currentPage %}

             & amp;lt;li & amp;gt; & amp;lt;a href="/periodic_task/?page={<!-- -->{ num }}" & amp;gt;{<!- - -->{ num}} & amp;lt;/a & amp;gt; & amp;lt;/li & amp;gt; {#Display current page number link#}

        {% else %}

             & amp;lt;liclass="item" & amp;gt; & amp;lt;a href="/periodic_task/?page={<!-- -->{ num}}" & amp; gt;{<!-- -->{ num }} & amp;lt;/a & amp;gt; & amp;lt;/li & amp;gt;

        {% endif %}

    {% endfor %}

   

    {# Next page link starts#}

    {% if tasks.has_next %} {# If there is a next page, display the next page link normally#}

         & amp;lt;liclass="next" & amp;gt; & amp;lt;a href="/periodic_task/?page={<!-- -->{tasks.next_page_number }}" & amp;gt;Next page& amp;lt;/a & amp;gt; & amp;lt;/li & amp;gt;

    {% else %}

         & amp;lt;li & amp;gt; & amp;lt;a href="#" & amp;gt;Next page& amp;lt;/a & amp;gt; & amp;lt;/ li&gt;

    {% endif %}

    {#End of next page link#}

 & amp;lt;/ul & amp;gt;

username = request.session.get('user', '')

task_list = PeriodicTask.objects.all()

task_count =PeriodicTask.objects.all().count() #Statistics

periodic_list =IntervalSchedule.objects.all() # Periodic tasks (such as executed every 1 hour)

crontab_list =CrontabSchedule.objects.all() # Scheduled tasks (such as a certain time of a certain month and day of a certain year, a certain time of every # day)

paginator = Paginator(task_list, 5) #Generate paginator object and set 5 records to be displayed on each page

page = request.GET.get('page',1) #Get the current page number, the default is page 1

currentPage=int(page) #Convert the obtained current page number into an integer type

try:

    task_list = paginator.page(page)#Get the record list of the current page number

except PageNotAnInteger:

    task_list = paginator.page(1)#If the number of pages entered is not an integer, the content of page 1 will be displayed.

except EmptyPage:

    task_list =paginator.page(paginator.num_pages)#If the entered page number is not among the system's page numbers, # then display the content of the last page

return render(request,"periodic_task.html", {"user": username,"tasks":task_list,"taskcounts": task_count, "periodics":periodic_list,"crontabs\ ": crontab_list })

username = request.session.get('user', '')# Read browser login Session

search_name =request.GET.get("task", "")

task_list = PeriodicTask.objects.filter(task__icontains=search_name)

periodic_list =IntervalSchedule.objects.all() # Periodic tasks (such as executed every 1 hour)

crontab_list =CrontabSchedule.objects.all() # Scheduled tasks (such as a certain time of a certain month and day of a certain year, a certain time of every # day)

return render(request,'periodic_task.html',{"user": username,"tasks":task_list,"periodics":periodic_list,"crontabs": crontab_list })

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

 & amp;lt;li & amp;gt;

                     & amp;lt;a href="../periodic_task"target="mainFrame" & amp;gt;

                         & amp;lt;iclass="glyphicon glyphicon-fire" & amp;gt; & amp;lt;/i & amp;gt;

                       mission plan       

                     & amp;lt;/a & amp;gt;

                 & amp;lt;/li & amp;gt;