正文

FasterRunner （httptunner+django）搭建以及小功能补充

IP云
IP云V管理员/12 阅读
11/29
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

配置 
下载地址
https://github.com/httprunner/FasterRunner

后端配置
https://www.jianshu.com/p/e26ccc21ddf2

前端配置
https://www.cnblogs.com/luopan/p/10250485.html

mac下安装RabbitMq
https://www.cnblogs.com/yihuihui/p/9095130.html

入口
http://localhost:8080/fastrunner/login
问题：
1. pycharm 创建的虚拟环境不能pip
https://www.jianshu.com/p/e46e36addf8d
2. pymysql/mysqlclient caching_sha2_password,
https://blog.csdn.net/weekdawn/article/details/81039382

django 1.8.2 mysql 加密方式 caching_sha2_password
django 2.0.3 mysql 加密方式 mysql_native_password

部分功能配置及补充

定时任务 

1. setting.py
djcelery.setup_loader()

CELERY_ENABLE_UTC = True

CELERY_TIMEZONE = 'Asia/Shanghai'

# BROKER_URL = 'amqp://username:password@IP:5672//'

BROKER_URL = 'amqp://guest:guest@127.0.0.1:5672//'

CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'

CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend'

CELERY_ACCEPT_CONTENT = ['application/json']

CELERY_TASK_SERIALIZER = 'json'

CELERY_RESULT_SERIALIZER = 'json'

CELERY_TASK_RESULT_EXPIRES = 7200

CELERYD_CONCURRENCY = 1 if DEBUG else 5

CELERYD_MAX_TASKS_PER_CHILD = 40
2.定时服务
cd /home/conan/conan-ta/FasterRunner/

nohup python3 manage.py celery beat -l info >> /Users/zd/Documents/FasterRunner/logs/beat.log 2>&1 &

cd /home/conan/conan-ta/FasterRunner/

celery multi start w1 -A FasterRunner -l info --logfile=/Users/zd/Documents/FasterRunner/logs/worker.log 2>&1 &
3.调试定时任务
更改定时相关逻辑时，需要关掉并重新启动celery beat，celery multi 才会生效，打印出来的调试信息在logs/worker.log下查看。
举例：定时任务，增加发送邮件的标题：
fastrunner/utils/task.py

fastrunner/task.py

发送邮件

setting.py 

# 发邮件

EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'

EMAIL_SEND_USERNAME = 'no-reply@fenbi.com'  # 定时任务报告发送邮箱，支持163,qq,sina,企业qq邮箱等，注意需要开通smtp服务

EMAIL_SEND_PASSWORD = ''     # 邮箱密码

EMAIL_PORT = 25

EMAIL_USE_TLS = True

fastrunner/utils/email.py 

import smtplib

from email.mime.text import MIMEText

from email.header import Header

from FasterRunner.settings import EMAIL_SEND_USERNAME, EMAIL_SEND_PASSWORD

def send_email_reports(receiver,save_summary,Cc=None,title=None):

    receiver = receiver.rstrip(';')

    all_receivers = receiver.split(';')

    if '@sina.com' in EMAIL_SEND_USERNAME:

        smtpserver = 'smtp.sina.com'

    elif '@163.com' in EMAIL_SEND_USERNAME:

        smtpserver = 'smtp.163.com'

    else:

        smtpserver = 'smtp.exmail.qq.com'

    if title:

        subject = "【%s】接口自动化测试报告"%title

    else:

        subject = "接口自动化测试报告"

    smtp = smtplib.SMTP_SSL(smtpserver, 465)

    smtp.login(EMAIL_SEND_USERNAME, EMAIL_SEND_PASSWORD)

    msg = MIMEText(save_summary, "html", "utf-8")

    msg["Subject"] = Header(subject, "utf-8")

    msg['From'] = Header('no-reply', 'utf-8')

    msg['To'] = receiver

    # 处理抄送

    if Cc:

        Cc = Cc.rstrip(';')

        msg['Cc'] = Cc

        all_receivers = receiver + ';' + Cc

        all_receivers = all_receivers.split(';')

    smtp.sendmail(EMAIL_SEND_USERNAME, all_receivers, msg.as_string())

fastrunner/tasks.py 

 from fastrunner.utils.emails import send_email_reports

 import time

 @shared_task

 def schedule_debug_suite(*args, **kwargs):

     """定时任务

     """

     print("定时任务start.....")

     project = kwargs["project"]

     receiver = kwargs["receiver"]

     Cc = kwargs["copy"]

     title = kwargs["name"]

     print("receiver****:%s"%receiver)

     print("args****:%s"%args)

     print("kwargs****:%s"%kwargs)

     print("定时任务end.....")

     receiver = receiver.strip()

     Cc = Cc.strip()

     suite = []

     test_sets = []

     config_list = []

     for pk in args:

         try:

             name = models.Case.objects.get(id=pk).name

             suite.append({

                 "name": name,

                 "id": pk

             })

         except ObjectDoesNotExist:

             pass

     for content in suite:

         test_list = models.CaseStep.objects. \

             filter(case__id=content["id"]).order_by("step").values("body")

         testcase_list = []

         config = None

         for content in test_list:

             body = eval(content["body"])

             if "base_url" in body["request"].keys():

                 config = eval(models.Config.objects.get(name=body["name"], project__id=project).body)

                 continue

             testcase_list.append(body)

         config_list.append(config)

         test_sets.append(testcase_list)

     summary = debug_suite(test_sets, project, suite, config_list, save=False)

     save_summary("", summary, project, type=3)

     # 整理数据

     testTime = summary['time']['start_at']

     testTime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(testTime))

     durTime = str(summary['time']['duration'])[:8]

     totalApi = summary['stat']['testsRun']

     successApi = summary['stat']['successes']

     FailApi = summary['stat']['failures']

     errorApi = summary['stat']['errors']

     skipApi = summary['stat']['skipped']

     htmll = """

     <table border="1" cellpadding="0" cellspacing="0" width="700px">

         <tr style="background-color: #f8f8fa">

             <th>测试时间</th>

             <th>持续时间</th>

             <th>Total</th>

             <th>Success</th>

             <th>Failed</th>

             <th>Error</th>

             <th>Skipped</th>

         </tr>

         <tr>

             <td>%s</td>

             <td>%s 秒</td>

             <td>%s</td>

             <td>%s</td>

             <td>%s</td>

             <td>%s</td>

             <td>%s</td>

         </tr>

     </table>

     <div style="height: 30px"></div>

     <table border="1" cellpadding="0" cellspacing="0" width="700px">

         <tr style="background-color: #f8f8fa">

             <th>名称</th>

             <th>请求地址</th>

             <th>请求方法</th>

             <th>响应时间（ms）</th>

             <th>测试结果</th>

         </tr>

     """ % (

         testTime, durTime, totalApi, successApi, FailApi, errorApi, skipApi

     )

     # 名称/请求地址/请求方法/响应时间/测试结果

     for i in summary['details']:  # [{},{}]

         detail = i['records']  # 列表

         for d in detail:

             name = d['name']

             uurl = d['meta_data']['request']['url']

             method = d['meta_data']['request']['method']

             responseTime = d['meta_data']['response']['response_time_ms']

             iresult = d['status']

             htmll += """

             <tr>

             <td>%s</td>

             <td>%s</td>

             <td>%s</td>

             <td>%s (ms)</td>

             <td>%s</td>

          </tr>

             """ % (name, uurl, method, responseTime, iresult)

     htmll = htmll + '</table>'

     if Cc:

         send_email_reports(receiver, htmll, Cc=Cc,title=title)

     else:

         send_email_reports(receiver, htmll,title=title)

启动/关闭 shell

start.sh 

#!/bin/bash

#启动 FasterWeb

echo -e "启动 FasterWeb"

cd /home/conan/conan-ta/FasterWeb/

nohup npm run build >> /home/shared/log/npm.log 2>&1 &

# 启动 FasterRunner

echo -e "启动 FasterRunner"

cd /home/conan/conan-ta/FasterRunner/

nohup python3 manage.py runserver 0.0.0.0:9000 >> /home/shared/log/django.log 2>&1 &

# 使用默认的celery.py启动

echo -e "启动celery beat"

cd /home/conan/conan-ta/FasterRunner/

nohup python3 manage.py celery beat -l info >> /Users/zd/Documents/FasterRunner/logs/beat.log 2>&1 &

# 使用默认的celery.py启动

echo -e "启动 celery work"

cd /home/conan/conan-ta/FasterRunner/

celery multi start w1 -A FasterRunner -l info --logfile=/Users/zd/Documents/FasterRunner/logs/worker.log 2>&1 &

stop.sh 

#!/bin/bash

# kill django pid

echo -e "shutting down django pid"

pids=$(ps aux | grep "python" | grep "runserver" | awk '{print $2}')

for pid in $pids

do

        kill -9 $pid

done

# kill celery beat pid

echo -e "shutting down celery beat pid"

pids=$(ps aux | grep "celery" | grep "FasterRunner" | awk '{print $2}')

for pid in $pids

do

        kill -9 $pid

done

相关阅读

目录[+]