本章介绍使用DjCelery即Django+Celery框架开发定时任务功能,在Autotestplat平台上实现单一接口自动化测试脚本、业务场景接口自动化测试脚本、App自动化测试脚本、Web自动化测试脚本等任务的定时执行、调度、管理等,从而取代Jenkins上的定时执行脚本和发送邮件等功能。**
自动化测试逻辑流程图11.1所示。
11.1 环境搭建
1.安装
步骤1 安装Celery。pyramid_celery-3.0.0,
配置https://pypi.python.org/pypi/pyramid_celery/。
步骤2 安装django-clery。django-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-beat。django-celery-beat-1.1.0,
https://pypi.python.org/pypi/ django_celery_beat。
步骤5 下载Redis-x64-3.2.100,
Redis-x64-3.2.100.zip github.com/MicrosoftAr…
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文件,加入如下内容。
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所示界面。
▲图11.2
步骤6 启动指令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文件,加入如下内容。
{% load bootstrap4 %}
{% bootstrap_css %}
{% bootstrap_javascript %}
{% csrf_token %}
ID | 任务名称 | 任务模块 | 时间计划 | 修改时间 | 开启 | 立即 | 编辑 | 删除 | |||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
{{ task.id }} | {{ task.name }} | {{ task.task }} | 每{{ periodic.period }} {{ periodic.every}}次 | {{ task.date_changed }} | {{ task.enabled }} | {% if task.id == 1 %}
{% elif task.id == 2 %} {% else %} {% endif %} |
{{ task.id }} | {{ task.name }} | {{ task.task }} | {{crontab.month_of_year }}年{{crontab.day_of_month }}月{{crontab.day_of_week }}日{{crontab.hour }}时{{ crontab.minute}}分 | {{ task.date_changed }} | {{ task.enabled }} | 运行 |
{# 把翻页功能固定显示在右下角#}
- <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>
功能描述:实现自动化测试任务调度执行,包括单一接口、扫描、流程接口、业务场景、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):
- username = request.session.get('user', '')
-
- task_list = PeriodicTask.objects.all()
-
- task_count =PeriodicTask.objects.all().count() #统计数
-
- periodic_list =IntervalSchedule.objects.all() # 周期任务(如每隔1小时执行1次)
-
- crontab_list =CrontabSchedule.objects.all() # 定时任务(如某年某月某日的某时,每# 天的某时)
-
- paginator = Paginator(task_list, 5) #生成paginator对象,设置每页显示5条记录
-
- page = request.GET.get('page',1) #获取当前的页码数,默认为第1页
-
- currentPage=int(page) #把获取的当前页码数转换成整数类型
-
- try:
-
- task_list = paginator.page(page)#获取当前页码数的记录列表
-
- except PageNotAnInteger:
-
- task_list = paginator.page(1)#如果输入的页数不是整数,则显示第1页内容
-
- except EmptyPage:
-
- task_list =paginator.page(paginator.num_pages)#如果输入的页数不在系统的页数中,# 则显示最后一页内容
-
- return render(request,"periodic_task.html", {"user": username,"tasks":task_list,"taskcounts": task_count, "periodics":periodic_list,"crontabs": crontab_list })
@login_required
deftasksearch(request):
- username = request.session.get('user', '')# 读取浏览器登录Session
-
- search_name =request.GET.get("task", "")
-
- task_list = PeriodicTask.objects.filter(task__icontains=search_name)
-
- periodic_list =IntervalSchedule.objects.all() # 周期任务(如每隔1小时执行1次)
-
- crontab_list =CrontabSchedule.objects.all() # 定时任务(如某年某月某日的某时,每# 天的某时)
-
- return render(request,'periodic_task.html',{"user": username,"tasks":task_list,"periodics":periodic_list,"crontabs": crontab_list })
在autotest/urls.py中加入:
path('periodic_task/',views.periodic_task),
path('tasksearch/', views.tasksearch),
在apitest/left.html中加入:
- <li>
-
- <a href="../periodic_task"target="mainFrame">
-
- <iclass="glyphicon glyphicon-fire"></i>
-
- 任务计划
-
- </a>
-
- </li>
<tr><td> </td></tr>
查看前端页面效果,如图11.3所示。
▲图11.3
最后感谢每一个认真阅读我文章的人,礼尚往来总是要有的,虽然不是什么很值钱的东西,如果你用得到的话可以直接拿走:
这些资料,对于【软件测试】的朋友来说应该是最全面最完整的备战仓库,这个仓库也陪伴上万个测试工程师们走过最艰难的路程,希望也能帮助到你!