本章介紹使用DjCelery即Django+Celery框架開發(fā)定時(shí)任務(wù)功能，在Autotestplat平臺(tái)上實(shí)現(xiàn)單一接口自動(dòng)化測(cè)試腳本、業(yè)務(wù)場(chǎng)景接口自動(dòng)化測(cè)試腳本、App自動(dòng)化測(cè)試腳本、Web自動(dòng)化測(cè)試腳本等任務(wù)的定時(shí)執(zhí)行、調(diào)度、管理等，從而取代Jenkins上的定時(shí)執(zhí)行腳本和發(fā)送郵件等功能。**
自動(dòng)化測(cè)試邏輯流程圖11.1所示。
11.1 環(huán)境搭建
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', 運(yùn)行 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中增加如下內(nèi)容。
加入1：
import djcelery
djcelery.setup_loader() #加載djcelery
加入2:
#數(shù)據(jù)庫(kù)調(diào)度
CELERYBEAT_SCHEDULER ='djcelery.schedulers.DatabaseScheduler'
加入3:
BROKER_URL = 'redis://127.0.0.1:6379/0'
BROKER_TRANSPORT = 'redis'
步驟2 在應(yīng)用Apitest目錄下新建celery.py文件1，加入如下內(nèi)容。
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文件，加入如下內(nèi)容。
--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('已運(yùn)行')
步驟4 啟動(dòng)服務(wù)python manage.py runserver。
步驟5 解壓縮后，運(yùn)行CMD，切換到相應(yīng)目錄，輸入啟動(dòng)Redis指令redis-server redis. windows.conf，成功后出現(xiàn)如圖11.2所示界面。
▲圖11.2
步驟6 啟動(dòng)指令python manage.py celery worker -l info。
步驟7 啟動(dòng)指令python manage.py celery beat。
11.2 前端功能實(shí)現(xiàn)
1．功能描述
完成實(shí)現(xiàn)單一接口測(cè)試用例、業(yè)務(wù)場(chǎng)景接口API測(cè)試用例、AppUI測(cè)試用例、WebUI測(cè)試用例的自動(dòng)化定時(shí)任務(wù)。
2．程序清單
在autotest\apitest\templates目錄下新建periodic_task.html文件，加入如下內(nèi)容。
<html>
<head>
{% load bootstrap4 %}
{% bootstrap_css %}
{% bootstrap_javascript %}
<title>產(chǎn)品自動(dòng)化測(cè)試平臺(tái)</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">
<!-- 導(dǎo)航欄-->
<nav class="navbar navbar-expand-smbg-dark navbar-dark fixed-top">
<div>
<ahref="#"> </a>
<ul>
</ul>
<ul>
<li><astyle='color:white' href="#"></a></li>
<li><astyle='color:white' href="/logout/"></a></li>
</ul>
</div>
</nav>
<!-- 搜索欄-->
<divstyle="padding-top: 70px;">
<formmethod="get" action="/tasksearch/">
{% csrf_token %}
<input type="search"name="task" placeholder="名稱" required>
<button type="submit">搜索</button>
<!-- 增加定時(shí)任務(wù)-->
<div style="float:right;width:73%">
<select name="PeriodicTask"id="PeriodicTask">
<option value="" selected>----定時(shí)任務(wù)----</option>
</select>
<a id="change_id_PeriodicTask"data-href-template="/admin/djcelery/periodictask/fk/change/?_to_field=id&_popup=1"title="更改選中的定時(shí)任務(wù)">
<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&_popup=1"title="增加另一個(gè)測(cè)試用例">
<imgsrc="/static/admin/img/icon-addlink.svg" alt="增加"/>增加
</a>
</form>
</div>
<!-- 任務(wù)計(jì)劃列表-->
<divstyle="padding-top: 20px;">
<div>
<table class="table table-striped">
<thead>
<tr>
<th>ID</th><th>任務(wù)名稱</th><th>任務(wù)模塊</th><th>時(shí)間計(jì)劃</th><th>修改時(shí)間</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">運(yùn)行</a>
{% elif task.id == 2 %}
<a href="../task_apitest"target="mainFrame">運(yùn)行</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 }}日{(diào){crontab.hour }}時(shí){{ crontab.minute}}分</a></td>
<td>{{ task.date_changed }}</td>
<td>{{ task.enabled }}</td>
<td><a href="../task_apis"target="mainFrame">運(yùn)行</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 style="position:absolute;right:100px; bottom:30px;"> {# 把翻頁(yè)功能固定顯示在右下角#}
<div style="position:absolute; right:100px; width:100px; ">
<tr><th>總數(shù)</th><td>{{ taskcounts }}</td></tr> {# 前端讀取定義的變量#}
</div>
<div>
<ulclass="pagination" id="pager">
{#上一頁(yè)鏈接開始#}
{%if tasks.has_previous %}
{# 如果有上一頁(yè)，則正常顯示上一頁(yè)鏈接#}
<li><ahref="/periodic_task/?page={{ tasks.previous_page_number }}">上一頁(yè)</a></li> {# 上一頁(yè)標(biāo)簽 #}
{%else %}
<li class="previous disabled"><ahref="#">上一頁(yè)</a></li>{# 如果當(dāng)前不存在上一頁(yè)，則上一頁(yè)的鏈接不可單擊#}
{%endif %}
{# 上一頁(yè)鏈接開始#}
{%for num in tasks.paginator.page_range %}
{% if num == currentPage %}
<li><a href="/periodic_task/?page={{ num }}">{{ num}}</a></li> {#顯示當(dāng)前頁(yè)數(shù)鏈接#}
{% else %}
<liclass="item"><a href="/periodic_task/?page={{ num}}">{{ num }}</a></li>
{% endif %}
{% endfor %}
{# 下一頁(yè)鏈接開始#}
{% if tasks.has_next %} {# 如果有下一頁(yè)，則正常顯示下一頁(yè)鏈接#}
<liclass="next"><a href="/periodic_task/?page={{tasks.next_page_number }}">下一頁(yè)</a></li>
{% else %}
<li><a href="#">下一頁(yè)</a></li>
{% endif %}
{# 下一頁(yè)鏈接結(jié)束#}
</ul>
</div>
</body>
</html>
功能描述：實(shí)現(xiàn)自動(dòng)化測(cè)試任務(wù)調(diào)度執(zhí)行，包括單一接口、掃描、流程接口、業(yè)務(wù)場(chǎng)景、Web搜索、自動(dòng)化平臺(tái)測(cè)試開發(fā)、App登錄，CSDN定時(shí)任務(wù)注冊(cè)，定時(shí)任務(wù)執(zhí)行等功能。
程序清單：在apitest/views.py中加入如下內(nèi)容。
from .tasks importhello_world
from .tasks importtest_readSQLcase
from djcelery.modelsimport PeriodicTask,CrontabSchedule,IntervalSchedule
任務(wù)計(jì)劃
@login_required
defperiodic_task(request):
username = request.session.get('user', '')
task_list = PeriodicTask.objects.all()
task_count =PeriodicTask.objects.all().count() #統(tǒng)計(jì)數(shù)
periodic_list =IntervalSchedule.objects.all() # 周期任務(wù)（如每隔1小時(shí)執(zhí)行1次）
crontab_list =CrontabSchedule.objects.all() # 定時(shí)任務(wù)（如某年某月某日的某時(shí)，每# 天的某時(shí)）
paginator = Paginator(task_list, 5) #生成paginator對(duì)象,設(shè)置每頁(yè)顯示5條記錄
page = request.GET.get('page',1) #獲取當(dāng)前的頁(yè)碼數(shù),默認(rèn)為第1頁(yè)
currentPage=int(page) #把獲取的當(dāng)前頁(yè)碼數(shù)轉(zhuǎn)換成整數(shù)類型
try:
task_list = paginator.page(page)#獲取當(dāng)前頁(yè)碼數(shù)的記錄列表
except PageNotAnInteger:
task_list = paginator.page(1)#如果輸入的頁(yè)數(shù)不是整數(shù)，則顯示第1頁(yè)內(nèi)容
except EmptyPage:
task_list =paginator.page(paginator.num_pages)#如果輸入的頁(yè)數(shù)不在系統(tǒng)的頁(yè)數(shù)中，# 則顯示最后一頁(yè)內(nèi)容
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() # 周期任務(wù)（如每隔1小時(shí)執(zhí)行1次）
crontab_list =CrontabSchedule.objects.all() # 定時(shí)任務(wù)（如某年某月某日的某時(shí)，每# 天的某時(shí)）
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中加入：
<tr> <td>
<li>
<a href="../periodic_task"target="mainFrame">
<iclass="glyphicon glyphicon-fire"></i>
任務(wù)計(jì)劃
</a>
</li>
<tr><td>&nbsp;</td></tr>
查看前端頁(yè)面效果，如圖11.3所示。
▲圖11.3
行動(dòng)吧，在路上總比一直觀望的要好，未來(lái)的你肯定會(huì)感 謝現(xiàn)在拼搏的自己！如果想學(xué)習(xí)提升找不到資料，沒人答疑解惑時(shí)，請(qǐng)及時(shí)加入扣群： 320231853，里面有各種軟件測(cè)試+開發(fā)資料和技術(shù)可以一起交流學(xué)習(xí)哦。
最后感謝每一個(gè)認(rèn)真閱讀我文章的人，禮尚往來(lái)總是要有的，雖然不是什么很值錢的東西，如果你用得到的話可以直接拿走：
這些資料，對(duì)于【軟件測(cè)試】的朋友來(lái)說應(yīng)該是最全面最完整的備戰(zhàn)倉(cāng)庫(kù)，這個(gè)倉(cāng)庫(kù)也陪伴上萬(wàn)個(gè)測(cè)試工程師們走過最艱難的路程，希望也能幫助到你！文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-851749.html
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-851749.html
到了這里，關(guān)于Django+Celery框架自動(dòng)化定時(shí)任務(wù)開發(fā)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
