《软件自动化测试开发》《自动化平台测试开发》作者,公众号"测试开发社区" ,博客www.autotestplat.com 。

Django+Celery框架自动化定时任务开发

上一篇 / 下一篇  2019-11-29 13:48:17

 本章介绍使用DjCeleryDjango+Celery框架开发定时任务功能,在Autotestplat平台上实现单一接口自动化测试脚本、业务场景接口自动化测试脚本、App自动化测试脚本、Web自动化测试脚本等任务的定时执行、调度、管理等,从而取代Jenkins上的定时执行脚本和发送邮件等功能。

自动化测试逻辑流程图11.1所示。       

640?wx_fmt=png

    ▲图11.1

11.1  环境搭建

1.安装

步骤1  安装Celerypyramid_celery-3.0.0

    配置https://pypi.python.org/pypi/pyramid_celery/

步骤2  安装django-clerydjango-celery-3.2.2

    配置https://pypi.python.org/pypi/django- celery

         INSTALLED_APPS= []

   加入2

             'djcelery',

               运行 Python manage.py migrate

步骤 3  安装celery-with-redis-3.0

    地址为https://pypi.python.org/pypi/celery-with-redis/

步骤 4  安装django-clery-beatdjango-celery-beat-1.1.0

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

步骤下载Redis-x64-3.2.100

 Redis-x64-3.2.100.zip         https://github.com/MicrosoftArchive/redis/releases

2.配置

步骤1  Settings.py中增加如下内容。

加入1

import djcelery 

djcelery.setup_loader()    #加载djcelery 

加入2:

#数据库调度

CELERYBEAT_SCHEDULER ='djcelery.schedulers.DatabaseScheduler'

加入3:

BROKER_URL = 'redis://127.0.0.1:6379/0'

BROKER_TRANSPORT = 'redis'

步骤2  在应用Apitest目录下新建celery.py文件1,加入如下内容。

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)

步骤3  新建tasks.py文件,加入如下内容。

# -*-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('已运行')

步骤4  启动服务python manage.py runserver

步骤5  解压缩后,运行CMD,切换到相应目录,输入启动Redis指令redis-server redis. windows.conf,成功后出现如图11.2所示界面。

640?wx_fmt=png

▲图11.2

步骤 启动指令python manage.py celery worker -l info

步骤7  启动指令python manage.py celery beat

11.2  前端功能实现

1.功能描述

完成实现单一接口测试用例、业务场景接口API测试用例AppUI测试用例、WebUI测试用例的自动化定时任务。

2.程序清单

autotest\apitest\templates目录下新建periodic_task.html文件,加入如下内容。

<html>

<head>

{% load bootstrap4 %}

{% bootstrap_css %}

{% bootstrap_javascript. %}

<title>产品自动化测试平台</title>

<link rel="stylesheet"type="text/css" href="/static/admin/css/forms.css" />

<script. type="text/javascript"src="/admin/jsi18n/"></script>

<script. type="text/javascript"src="/static/admin/js/vendor/jquery/jquery.js"></script>

<script. type="text/javascript"src="/static/admin/js/jquery.init.js"></script>

<script. type="text/javascript"src="/static/admin/js/core.js"></script>

<script. type="text/javascript"src="/static/admin/js/admin/RelatedObjectLookups.js"></script>

<script. type="text/javascript"src="/static/admin/js/actions.js"></script>

<script. type="text/javascript"src="/static/admin/js/urlify.js"></script>

<script. type="text/javascript"src="/static/admin/js/prepopulate.js"></script>

<script. type="text/javascript"src="/static/admin/js/vendor/xregexp/xregexp.js"></script>

<meta. name="viewport"content="user-scalable=no, width=device-width, initial-scale=1.0,maximum-scale=1.0">

   

<link rel="stylesheet"type="text/css" href="/static/admin/css/responsive.css"/>

 

<meta. name="robots"content="NONE,NOARCHIVE" />

 

</head>

<body role="document">

<!-- 导航栏-->

<nav class="navbar navbar-expand-smbg-dark navbar-dark fixed-top">

<div>

<ahref="#">&nbsp;</a>

<ul>

</ul>

<ul>

<li><astyle='color:white' href="#"></a></li>

<li><astyle='color:white' href="/logout/"></a></li>

</ul>

</div>

</nav>

<!-- 搜索栏-->

<div>

<formmethod="get" action="/tasksearch/">

 

{% csrf_token %}    

<input type="search"name="task" placeholder="名称" required>

 

<button type="submit">搜索</button>

  

<!-- 增加定时任务-->

<div >

 

<select name="PeriodicTask"id="PeriodicTask">

<option value="" selected>----定时任务----</option>

</select>

<a id="change_id_PeriodicTask"data-href-template="/admin/djcelery/periodictask/__fk__/change/?_to_field=id&amp;_popup=1"title="更改选中的定时任务">

<imgsrc="/static/admin/img/icon-changelink.svg" alt="修改"/>

</a>

<a style='color:light blue' id="add_id_PeriodicTask" href="/admin/djcelery/periodictask/add/?_to_field=id&amp;_popup=1"title="增加另一个测试用例">

<imgsrc="/static/admin/img/icon-addlink.svg" alt="增加"/>增加

</a>

</form>

</div>

 

<!-- 任务计划列表-->

<div>

<div>

<table class="table table-striped">

<thead>

 

<tr>

<th>ID</th><th>任务名称</th><th>任务模块</th><th>时间计划</th><th>修改时间</th><th>开启</th><th>立即</th><th>编辑</th><th>删除</th>

</tr>

</thead>

<tbody>

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

<tr>

 

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

<td>{{ task.id }}</td>

<td>{{ task.name }}</td>

<td>{{ task.task }}</td>

<td><a style='color:green'>{{ periodic.period }} {{ periodic.every}}</a></td>

<td>{{ task.date_changed }}</td>

<td>{{ task.enabled }}</td>

<td>{% if task.id == 1 %}

<a href="../task_apis"target="mainFrame">运行</a>

{% elif task.id == 2 %}

<a href="../task_apitest"target="mainFrame">运行</a>

{% else %}

{% endif %}

</td>

<td><a style='color:light blue'class="related-widget-wrapper-link add-related"id="add_id_Apitest" href="../admin/djcelery/periodictask/{{task.id }}/change/?_to_field=id&_popup=1"><imgsrc="/static/admin/img/icon-changelink.svg"/></a></td>

<td><a style='color:light blue'class="related-widget-wrapper-link add-related" id="add_id_Apitest"href="../admin/djcelery/periodictask/{{ task.id}}/delete/?_to_field=id&_popup=1"><imgsrc="/static/admin/img/icon-deletelink.svg"/></a></td>

 

{% else %}

 

{% endif %}

 

{% for crontab in crontabs %}

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

<td>{{ task.id }}</td>

<td>{{ task.name }}</td>

<td>{{ task.task }}</td>

<td><a style='color:green'>{{crontab.month_of_year }}{{crontab.day_of_month }}{{crontab.day_of_week }}{{crontab.hour }}{{ crontab.minute}}</a></td>

<td>{{ task.date_changed }}</td>

<td>{{ task.enabled }}</td>

<td><a href="../task_apis"target="mainFrame">运行</a></td>

<td><a style='color:light blue'class="related-widget-wrapper-link add-related"id="add_id_Apitest" href="../admin/djcelery/periodictask/{{task.id }}/change/?_to_field=id&_popup=1"><imgsrc="/static/admin/img/icon-changelink.svg"/></a></td>

<td><a style='color:light blue'class="related-widget-wrapper-link add-related"id="add_id_Apitest" href="../admin/djcelery/periodictask/{{task.id }}/delete/?_to_field=id&_popup=1"><imgsrc="/static/admin/img/icon-deletelink.svg"/></a></td>

{% else %}

{% endif %}

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

 

</tbody>

</table>

</div>

</div>

 

<span >   {# 把翻页功能固定显示在右下角#}

 

<div >

<tr><th>总数</th><td>{{ taskcounts }}</td></tr> {# 前端读取定义的变量#}

</div>

 

<div>

    <ulclass="pagination" id="pager">

          {#上一页链接开始#}

        {%if tasks.has_previous %}

           {#  如果有上一页,则正常显示上一页链接#}

           <li><ahref="/periodic_task/?page={{ tasks.previous_page_number }}">上一页</a></li>    {#  上一页标签 #}

        {%else %}

           <li class="previous disabled"><ahref="#">上一页</a></li>{# 如果当前不存在上一页,则上一页的链接不可单击#}

        {%endif %}

        {# 上一页链接开始#}

       

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

        

           {% if num == currentPage %}

                <li><a href="/periodic_task/?page={{ num }}">{{ num}}</a></li> {#显示当前页数链接#}

            {% else %}

                <liclass="item"><a href="/periodic_task/?page={{ num}}">{{ num }}</a></li>

            {% endif %}

        {% endfor %}

       

        {# 下一页链接开始#}

        {% if tasks.has_next %} {#  如果有下一页,则正常显示下一页链接#}

            <liclass="next"><a href="/periodic_task/?page={{tasks.next_page_number }}">下一页</a></li>

        {% else %}

            <li><a href="#">下一页</a></li>

        {% endif %}

        {# 下一页链接结束#}

    </ul>

</div>

 

</body>

</html>

功能描述:实现自动化测试任务调度执行,包括单一接口、扫描、流程接口、业务场景、Web搜索、自动化平台测试开发、App登录,CSDN定时任务**,定时任务执行等功能。

程序清单:在apitest/views.py中加入如下内容。

from .tasks importhello_world

from .tasks importtest_readSQLcase

from djcelery.modelsimport PeriodicTask,CrontabSchedule,IntervalSchedule

 

 

任务计划

@login_required

defperiodic_task(request):

TAG:

 

评分:0

我来说两句

显示全部

:loveliness: :handshake :victory: :funk: :time: :kiss: :call: :hug: :lol :'( :Q :L ;P :$ :P :o :@ :D :( :)

finer

finer

《软件自动化测试开发》《自动化平台测试开发》作者,公众号"测试开发社区" ,博客www.autotestplat.com 。

日历

« 2020-05-01  
     12
3456789
10111213141516
17181920212223
24252627282930
31      

数据统计

  • 访问量: 1916
  • 日志数: 17
  • 图片数: 1
  • 建立时间: 2019-11-29
  • 更新时间: 2020-03-26

RSS订阅

Open Toolbar