使用unittest和Django搭配写一个机遇unittest的接口测试的平台。 框架搭建 表结构设计
每个测试项目下面有多个测试用例。
需要实现：
-
对测试项目的
-
增
-
删
-
改
-
查，查看该测试项目下面所有的测试用例
-
为该测试项目批量导入，添加测试用例
-
-
对项目下的接口进行
-
增
-
删
-
改
-
查
-
单个用例的执行
-
批量执行选中的用例，并且将执行结果（html报告）下载到本地
-
-
数据可视化
-
接口项目相关数据进行统计
-
用例执行情况进行统计
-
-
定时任务
-
每个测试项目都有周期，在周期结束后，自动的将该项目中的所有用例，执行一遍，生成测试报告。
-
使用Django发邮件功能，将报告发送
-
相关功能截图：
接口项目列表
为指定的接口测试项目批量导入
为指定的接口测试项目添加用例
某个接口测试项目下的用例列表
配置settings
AdminLTE引入
-
将AdminLTE前端框架文件夹，拷贝到Django的static目录中。
-
将该AdminLTE框架中的starter.html文件拷贝到Django的templates目录中。
-
其他页面就可以继承该starter.html文件了。
配置starter.html文件的静态文件
表结构设计
接口项目一个表，项目下面有用例，用例也是一个表，那么用例表和接口项目表示一对多的关系。
接口项目表结构：
-
接口项目名称
-
接口项目描述
-
项目开始时间
-
项目结束时间
在实际开发中，如果发现还缺少字段，就手动添加
用例表结构设计
参考上图：
-
用例所属的接口项目，外键
-
用例名称
-
用例描述
-
用例的请求类型
-
用例的请求url
-
用例的请求参数
-
预期值
-
执行状态
-
已执行
-
未执行
-
-
通过状态
-
未通过
-
已通过
-
-
用例的执行结果报告
执行时间
# 平台分析
## 页面模板
adminlte
bootstrap
jquery
# Excel上传
```python
import xlrd
from django.shortcuts import render, redirect, HttpResponse
from django.db import transaction
def import_case(request, pk):
""" 为接口批量导入用例, pk:接口的pk """
if request.method == "POST":
try:
with transaction.atomic(): # 事务
# 读取Excel表格
book = xlrd.open_workbook(file_contents=request.FILES.get('excel').read())
sheet = book.sheet_by_index(0)
title = sheet.row_values(0)
'''
['cnodejs项目', 'get /topics 主题首页', 'https://cnodejs.org/api/v1/topics', 'get', '', '{"success":true}']
'''
for row in range(1, sheet.nrows):
row = sheet.row_values(row)
# break
models.Case.objects.create(
case_API_id=pk,
case_title=row[0],
case_desc=row[1],
case_url=row[2],
case_method=row[3],
case_params=row[4],
case_expect=row[-1],
)
return redirect('/case_list/{}'.format(pk))
except Exception as e:
return render(request, 'import_case.html', {"error": "文件格式不对,只能上传 xls/xlsx 的 {}".format(e)})
else:
return render(request, 'import_case.html',{"error": ""})
```
# 下载
```python
from utils.execute_case import execute
from django.http import JsonResponse
from django.http import FileResponse
def case_execute(request, pk):
""" 执行用例 PK:要执行用例的pk"""
if request.method == "POST":
pass
else:
# 1. 从前端获取要执行的用例id,然后从数据库取出这个用例对象
case_obj = models.Case.objects.filter(pk=pk).first()
# 2. 从用例对象中,提取相关字段,执行这个用例
file_path = execute(case_obj)
response = FileResponse(open(file_path, 'rb'))
response['Content-Type'] = 'application/octet-stream'
# filename的名称不能含有中文
response['Content-Disposition'] = 'attachment;filename="{}.html"'.format('report')
# print(111111, case_obj.case_title)
return response
# models 自定义字段
```python
from django.db import models
class API(models.Model):
""" 接口表 """
api_title = models.CharField(max_length=32, verbose_name='接口名称')
api_desc = models.CharField(max_length=128, verbose_name='接口描述')
def __str__(self):
return self.api_title
def xxoo(self):
if self.case_set.count():
a = "%s%% " % (self.case_set.filter(case_pass_status=1).count() / self.case_set.count() * 100)
return a
else:
return 0
```
前端调用:
```html
{% for foo in api_obj %}
<tr>
<td>{{ forloop.counter }}</td>
<td>{{ foo.api_title }}</td>
<td>{{ foo.api_desc }}</td>
<td>{{ foo.case_set.count }}</td> <!-- 接口下面有多少用例 -->
<td>{{ foo.xxoo }} </td> <!-- 接口下面的用例有多少通过的,计算公式: 通过/总数*100 -->
<td>
<a href="{% url 'api_edit' foo.pk %}" class="btn btn-success btn-sm">编辑接口</a>
<a href="{% url 'api_del' foo.pk %}" class="btn btn-success btn-sm">删除接口</a>
<a href="{% url 'case_add' foo.pk %}" class="btn btn-success btn-sm">添加用例</a>
<a href="{% url 'import_case' foo.pk %}" class="btn btn-success btn-sm">批量导入</a>
<a href="{% url 'case_list' foo.pk %}" class="btn btn-success btn-sm">查看用例</a>
</td>
</tr>
{% endfor %}
```
# 执行用例的思路
## 第一种,执行单个用例
1. 点击执行
2. 后台过滤出来当前的用例,将用例对象返回
3. unittest/pytest框架做
1. 从用例对象中,提取各个参数,发请求
2. 校验请求结果
3. 断言
4. 生成测试报告
5. 将该测试报告存储到数据库
1. 将HTML文件存储都某个目录下,数据库存储文件的路径
2. 直接将HTML文本存储数据的字段
6. 用例的执行状态,通过状态,都要改
4. 如何将测试报告返回给前端
1. 直接将报告下载到本地
## 批量执行(可选)
# 定时任务
定时任务,https://www.cnblogs.com/Neeo/p/10435059.html
发邮件:https://www.cnblogs.com/Neeo/articles/11199085.html
1. 每天检查接口表,检查有没有当天要结束的接口
2. 如果检查到有当天要结束的接口
3. 就把该接口下的所有用例执行一遍
4. 生成测试报告
## 实现步骤
1. 接口表,添加字段,接口的开始/结束时间
2. 每天固定的时间,读取接口表,判断结束时间是否为当天
3. 如果为当天要结束的,把该接口下的用例执行一遍
4. 生成测试报告,保存到数据库
5. 前端能查询该条执行记录
6. 下载这个记录的测试报告
# datetime
```python
import os
import datetime
import django
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "MB.settings")
django.setup()
from app01 import models
if __name__ == '__main__':
print(datetime.datetime.now())
today = datetime.date.today()
print(today.year, today.month)
```
实现可视化的一般方式：
-服务器端处理：pyecharts
-后端负责给数据，前端拿到数据，自己渲染：echarts
如何使用echarts
-官网：https://www.echartsis.com/zh/index.html
-cdn：https://www.bootcdn.cn/echarts
-django使用：
1.要引入echarts.mim.js
<script src="https://cdn.bootcss.com/echarts/3.0.1/echarts.min.js"></script>
2.从官网的实例 https://www.echartsis.com/examples/zh/index.html找一个合适的示例
3.将示例拷贝到我们前端页面，
before：页面中要定义一个标签，设置一下长宽
<div class="row">
<div ，'Sat'，'Sun]
yAxis:{
type:‘value'
series:【40
data:[820，932，901，934，1290，1330，1320]，
type:‘line'
}]
}；
var myChart = echarts.init(document.getElementById("p1"));
myChart.setOption(option);
}
4、后端将数据库传递给前端，前端将数据替换到option的相对位置
项目目录：
models.py：
from django.db import models
# Create your models here.
class API(models.Model):
"""接口表"""
api_title = models.CharField(max_length=32, verbose_name="接口名称")
api_desc = models.CharField(max_length=128, verbose_name="接口描述")
api_start_time = models.DateField(verbose_name="接口开始时间")
api_stop_time = models.DateField(verbose_name="接口结束时间")
def __str__(self):
return self.api_title
def xxoo(self):
if self.case_set.count():
a = "%.2f%%" % (self.case_set.filter(case_pass_status=1).count() / self.case_set.count() * 100)
return a
else:
return 0
class Case(models.Model):
"""用例表"""
case_API = models.ForeignKey(to="API", verbose_name="所属接口")
case_title = models.CharField(max_length=32, verbose_name="用例名称")
case_desc = models.CharField(max_length=128, verbose_name="用例描述")
case_expect = models.CharField(max_length=128, verbose_name="预期值")
case_url = models.CharField(max_length=256, verbose_name="请求URL")
case_params = models.CharField(max_length=256, verbose_name="请求参数", default="")
# case_data = models.CharField(max_length=256, verbose_name="请求参数", default="")
case_method = models.CharField(max_length=10, verbose_name="请求类型")
case_report = models.TextField(verbose_name="用例报告", default="")
case_execute_status_choices = (
(1, "已执行"),
(0, "未执行"),
)
case_execute_status = models.IntegerField(choices=case_execute_status_choices, default=0, verbose_name='执行状态')
case_pass_status_choices = (
(1, "已通过"),
(0, "未通过"),
)
case_pass_status = models.IntegerField(choices=case_pass_status_choices, default=0, verbose_name='通过状态')
def __str__(self):
return self.case_title
class CrontabLog(models.Model):
"""定时任务日志"""
log_api = models.ForeignKey(to="API", verbose_name="所属接口")
log_creat_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间")
log_report = models.TextField(verbose_name="报告")
class Meta:
ordering = ["-log_creat_time"]
views.py：
import xlrd
from django.db import transaction
from django.shortcuts import render, redirect, HttpResponse
from app01 import models
from utils import myForm
from django.http import FileResponse
from utils.execute_case import execute
from django.http import JsonResponse
from django.http import FileResponse
from utils import echartsMsg
# Create your views here.
def index(request):
"""主页功能"""
obj = models.API.objects.all()
return render(request, "index.html", {"api_obj": obj})
def api_add(request):
"""添加接口"""
if request.method == "POST":
form_obj = myForm.ApiForm(request.POST)
if form_obj.is_valid():
form_obj.save()
return redirect("/index/")
else:
return render(request, "api_add.html", {"form_obj": form_obj})
else:
form_obj = myForm.ApiForm()
return render(request, "api_add.html", {"form_obj": form_obj})
def api_edit(request, pk):
"""编辑接口 pk:接口的pk"""
api_obj = models.API.objects.filter(pk=pk).first()
if request.method == "POST":
form_obj = myForm.ApiForm(request.POST, instance=api_obj)
if form_obj.is_valid():
form_obj.save()
return redirect("/index/")
else:
return render(request, "api_edit.html", {"form_obj": form_obj})
else:
form_obj = myForm.ApiForm(instance=api_obj)
return render(request, "api_edit.html", {"form_obj": form_obj})
def api_del(request, pk):
"""删除接口"""
models.API.objects.filter(pk=pk).delete()
return redirect("/index/")
def case_list(request, pk):
"""接口下面的用例列表，pk：接口的pk"""
case_obj = models.Case.objects.filter(case_API_id=pk)
if request.method == "POST":
case_pk = request.POST.getlist("case_pk")
# print(111, case_pk)
if case_pk:
# 从数据库中，将包含case_pk的用例记录对象提取出来
case_list_obj = models.Case.objects.filter(pk__in=case_pk)
# print(222, case_list_obj)
# 交给execute模块去循环调用
f = execute(case_list_obj)
response = HttpResponse(f.getvalue())
response['Content-Type'] = 'application/octet-stream'
response['Content-Disposition'] = 'attachment;filename="{}.html"'.format('report')
# print(111, case_obj.case_title)
return response
else:
return render(request, "case_list.html", {"case_obj": case_obj, "pk": pk, "error": "至少勾选一个用例吧"})
else:
# case_obj = models.Case.objects.filter(case_API_id=pk)
return render(request, "case_list.html", {"case_obj": case_obj, "pk": pk, "error": ""})
def case_add(request, pk):
"""为接口添加用例"""
if request.method == "POST":
# a = dict(request.POST)
# b = a.setdefault("case_API_id", pk)
# print(a)
form_obj = myForm.CaseForm(request.POST)
if form_obj.is_valid():
form_obj.save()
return redirect("/case_list/{}".format(pk))
else:
return render(request, "case_add.html", {"form_obj": form_obj, "pk": pk})
else:
form_obj = myForm.CaseForm()
return render(request, "case_add.html", {"form_obj": form_obj, "pk": pk})
def case_edit(request, pk):
"""编辑用例,pk:用例的pk"""
case_obj = models.Case.objects.filter(pk=pk).first()
if request.method == "POST":
form_obj = myForm.CaseForm(request.POST, instance=case_obj)
if form_obj.is_valid():
form_obj.save()
obj1 = models.API.objects.filter(case__id=pk).first()
# 将用例执行状态和其他字段恢复到初始状态
models.Case.objects.filter(pk=pk).update(
case_execute_status=0,
case_pass_status=0,
case_report=""
)
return redirect("/case_list/{}".format(obj1.pk))
else:
return render(request, "case_edit.html", {"form_obj": form_obj})
else:
form_obj = myForm.CaseForm(instance=case_obj)
return render(request, "case_edit.html", {"form_obj": form_obj})
def case_del(request, pk):
"""删除用例，pk:用例的pk"""
# obj = models.Case.objects.filter(pk=pk).values("case_API_id")
# api_pk = list(obj)[0]["case_API_id"]
obj1 = models.API.objects.filter(case__id=pk).first()
# print(obj1)
# print(obj1.pk)
# print(obj[0]["case_API_id"])
models.Case.objects.filter(pk=pk).delete()
return redirect("/case_list/{}".format(obj1.pk))
# return HttpResponse("OK")
def case_execute(request, pk):
"""执行单个用例,pk:要执行用例的pk"""
if request.method == "POST":
pass
else:
# 1、从前端获取要执行的用例id、然后从数据库取出这个用例对象
case_obj = models.Case.objects.filter(pk=pk).first()
execute(case_obj)
# 2、从用例对象中，提取相关字段，执行这个用例
obj1 = models.API.objects.filter(case_id=pk).first()
return redirect("/case_list/{}".format(obj1.pk))
def download_case_report(request, pk):
"""下载报告"""
case_obj = models.Case.objects.filter(pk=pk).first()
# file_path = execute(case_obj)
response = FileResponse(case_obj.case_report)
response['Content-Type'] = 'application/octet-stream'
response['Content-Disposition'] = 'attachment;filename="{}.html"'.format('report')
# print(111, case_obj.case_title)
return response
def import_case(request, pk):
"""批量导入用例"""
if request.method == "POST":
try:
with transaction.atomic():
# 读取Excel表格信息：
book = xlrd.open_workbook(file_contents=request.FILES.get("excel").read())
sheet = book.sheet_by_index(0)
# print(sheet.nrows)
title = sheet.row_values(0)
l = []
for row in range(1, sheet.nrows):
# print(sheet.row_values(row))
# break
# l.append(dict(zip(title, sheet.row_values(row))))
# print(1111, l)
row = sheet.row_values(row)
models.Case.objects.create(
case_API_id=pk,
case_title=row[0],
case_desc=row[1],
case_url=row[2],
case_method=row[3],
case_params=row[4],
case_expect=row[-1],
)
return redirect("/case_list/{}".format(pk))
except Exception as e:
return render(request, "import_case.html", {"error": "文件格式不对，只能上传 xls 或者 xlsx 的{}".format(e)})
else:
return render(request, "import_case.html", {"error": ""})
def crontab_index(request):
"""定时任务主页"""
# obj = models.CrontabLog.objects.all().order_by("-log_creat_time")
obj = models.CrontabLog.objects.all()
return render(request, "crontab_index.html", {"obj": obj})
def show_crontablog_report(request, pk):
"""显示定时任务报告"""
obj = models.CrontabLog.objects.filter(pk=pk).first()
return render(request, "crontablog_report.html", {"f": obj.log_report, "pk": pk})
def download_crontab_log(request, pk):
"""下载定时任务测试报告、pk:log记录的pk"""
log_obj = models.CrontabLog.objects.filter(pk=pk).first()
# file_path = execute(case_obj)
response = FileResponse(log_obj.log_report)
response['Content-Type'] = 'application/octet-stream'
response['Content-Disposition'] = 'attachment;filename="{}.html"'.format('report')
# print(111, case_obj.case_title)
return response
def echarts_msg(request):
"""可视化信息"""
if request.method == "POST":
area_basic = echartsMsg.area_basic()
bar_y_category_stack = echartsMsg.bar_y_category_stack()[0]
area_stack = echartsMsg.bar_y_category_stack()[1]
pie_doughnut = echartsMsg.pie_doughnut()
pie_simple = echartsMsg.pie_simple()
return JsonResponse({
'area_basic': area_basic,
"bar_y_category_stack": bar_y_category_stack,
'pie_doughnut': pie_doughnut,
'pie_simple': pie_simple,
'area_stack': area_stack}
)
else:
return render(request, "echarts_msg.html")
settings.py：
"""
Django settings for MB project.
Generated by 'django-admin startproject' using Django 1.11.27.
For more information on this file, see
https://docs.djangoproject.com/en/1.11/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/1.11/ref/settings/
"""
import os
# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/1.11/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = 'hna_@ml)em@kidpae+d(ufxv=*52jknx_5alj1a(^roykctd2i'
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True
ALLOWED_HOSTS = []
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'app01.apps.App01Config',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'MB.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [os.path.join(BASE_DIR, 'templates')]
,
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'MB.wsgi.application'
# Database
# https://docs.djangoproject.com/en/1.11/ref/settings/#databases
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
}
}
# Password validation
# https://docs.djangoproject.com/en/1.11/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
# Internationalization
# https://docs.djangoproject.com/en/1.11/topics/i18n/
LANGUAGE_CODE = 'en-us'
# TIME_ZONE = 'UTC'
TIME_ZONE = "Asia/Shanghai"
USE_I18N = True
USE_L10N = True
USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/1.11/howto/static-files/
STATIC_URL = '/static/'
STATICFILES_DIRS = [os.path.join(BASE_DIR, "static_m")]
urls.py：
"""MB URL Configuration
The `urlpatterns` list routes URLs to views. For more information please see:
https://docs.djangoproject.com/en/1.11/topics/http/urls/
Examples:
Function views
1. Add an import: from my_app import views
2. Add a URL to urlpatterns: url(r'^$', views.home, name='home')
Class-based views
1. Add an import: from other_app.views import Home
2. Add a URL to urlpatterns: url(r'^$', Home.as_view(), name='home')
Including another URLconf
1. Import the include() function: from django.conf.urls import url, include
2. Add a URL to urlpatterns: url(r'^blog/', include('blog.urls'))
"""
from django.conf.urls import url
from django.contrib import admin
from app01 import views
urlpatterns = [
url(r'^admin/', admin.site.urls),
url(r'^index/', views.index, name="index"),
url(r'^api_add/', views.api_add, name="api_add"),
url(r'^api_edit/(?P<pk>d+)$', views.api_edit, name="api_edit"),
url(r'^api_del/(?P<pk>d+)$', views.api_del, name="api_del"),
url(r'^case_list/(?P<pk>d+)$', views.case_list, name="case_list"),
url(r'^case_add/(?P<pk>d+)$', views.case_add, name="case_add"),
url(r'^case_del/(?P<pk>d+)$', views.case_del, name="case_del"),
url(r'^case_edit/(?P<pk>d+)$', views.case_edit, name="case_edit"),
url(r'^case_execute/(?P<pk>d+)$', views.case_execute, name="case_execute"),
url(r'^import_case/(?P<pk>d+)$', views.import_case, name="import_case"),
url(r'^download_case_report/(?P<pk>d+)$', views.download_case_report, name="download_case_report"),
url(r'^download_crontab_log/(?P<pk>d+)$', views.download_crontab_log, name="download_crontab_log"),
url(r'^show_crontablog_report/(?P<pk>d+)$', views.show_crontablog_report, name="show_crontablog_report"),
url(r'^echarts_msg/', views.echarts_msg, name="echarts_msg"),
url(r'^crontab_index/', views.crontab_index, name="crontab_index"),
]
api_add.html：
{% extends "base.html" %}
{% block content %}
<div class="row">
<div class="col-md-8 offset-md-2"><a href="{% url 'index' %}">返回主页</a></div>
<div class="col-md-8 offset-md-2">
{% include 'form.html' %}
</div>
</div>
{% endblock %}
api_edit.html：
{% extends "base.html" %}
{% block content %}
<div class="row">
<div class="col-md-8 offset-md-2"><a href="{% url 'index' %}">返回主页</a></div>
<div class="col-md-8 offset-md-2">
{% include 'form.html' %}
</div>
</div>
{% endblock %}
base.html：
{% load static %}
<!DOCTYPE html>
<!--
This is a starter template page. Use this page to start your new project from
scratch. This page gets rid of all links and provides the needed markup only.
-->
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta http-equiv="x-ua-compatible" content="ie=edge">
<title>AdminLTE 3 | Starter</title>
<!-- Font Awesome Icons -->
<link rel="stylesheet" href="{% static 'AdminLTE-master/plugins/fontawesome-free/css/all.min.css' %}">
<!-- Theme style -->
<link rel="stylesheet" href="{% static 'AdminLTE-master/dist/css/adminlte.min.css' %}">
<!-- Google Font: Source Sans Pro -->
<link href="https://fonts.googleapis.com/css?family=Source+Sans+Pro:300,400,400i,700" rel="stylesheet">
</head>
<body class="hold-transition sidebar-mini">
<div class="wrapper">
<!-- Navbar -->
<nav class="main-header navbar navbar-expand navbar-white navbar-light">
<!-- Left navbar links -->
<ul class="navbar-nav">
<li class="nav-item">
<a class="nav-link" data-widget="pushmenu" href="#"><i class="fas fa-bars"></i></a>
</li>
<li class="nav-item d-none d-sm-inline-block">
<a href="#" class="nav-link">Home</a>
</li>
<li class="nav-item d-none d-sm-inline-block">
<a href="#" class="nav-link">Contact</a>
</li>
</ul>
<!-- SEARCH FORM -->
<form class="form-inline ml-3">
<div class="input-group input-group-sm">
<input class="form-control form-control-navbar" type="search" placeholder="Search" aria-label="Search">
<div class="input-group-append">
<button class="btn btn-navbar" type="submit">
<i class="fas fa-search"></i>
</button>
</div>
</div>
</form>
<!-- Right navbar links -->
<ul class="navbar-nav ml-auto">
<!-- Messages Dropdown Menu -->
<li class="nav-item dropdown">
<a class="nav-link" data-toggle="dropdown" href="#">
<i class="far fa-comments"></i>
<span class="badge badge-danger navbar-badge">3</span>
</a>
<div class="dropdown-menu dropdown-menu-lg dropdown-menu-right">
<a href="#" class="dropdown-item">
<!-- Message Start -->
<div class="media">
<img src="{% static 'AdminLTE-master/dist/img/user1-128x128.jpg' %}" alt="User Avatar" class="img-size-50 mr-3 img-circle">
<div class="media-body">
<h3 class="dropdown-item-title">
Brad Diesel
<span class="float-right text-sm text-danger"><i class="fas fa-star"></i></span>
</h3>
<p class="text-sm">Call me whenever you can...</p>
<p class="text-sm text-muted"><i class="far fa-clock mr-1"></i> 4 Hours Ago</p>
</div>
</div>
<!-- Message End -->
</a>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item">
<!-- Message Start -->
<div class="media">
<img src="{% static 'AdminLTE-master/dist/img/user8-128x128.jpg' %}" alt="User Avatar" class="img-size-50 img-circle mr-3">
<div class="media-body">
<h3 class="dropdown-item-title">
John Pierce
<span class="float-right text-sm text-muted"><i class="fas fa-star"></i></span>
</h3>
<p class="text-sm">I got your message bro</p>
<p class="text-sm text-muted"><i class="far fa-clock mr-1"></i> 4 Hours Ago</p>
</div>
</div>
<!-- Message End -->
</a>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item">
<!-- Message Start -->
<div class="media">
<img src="{% static 'AdminLTE-master/dist/img/user8-128x128.jpg' %}" alt="User Avatar" class="img-size-50 img-circle mr-3">
<div class="media-body">
<h3 class="dropdown-item-title">
Nora Silvester
<span class="float-right text-sm text-warning"><i class="fas fa-star"></i></span>
</h3>
<p class="text-sm">The subject goes here</p>
<p class="text-sm text-muted"><i class="far fa-clock mr-1"></i> 4 Hours Ago</p>
</div>
</div>
<!-- Message End -->
</a>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item dropdown-footer">See All Messages</a>
</div>
</li>
<!-- Notifications Dropdown Menu -->
<li class="nav-item dropdown">
<a class="nav-link" data-toggle="dropdown" href="#">
<i class="far fa-bell"></i>
<span class="badge badge-warning navbar-badge">15</span>
</a>
<div class="dropdown-menu dropdown-menu-lg dropdown-menu-right">
<span class="dropdown-header">15 Notifications</span>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item">
<i class="fas fa-envelope mr-2"></i> 4 new messages
<span class="float-right text-muted text-sm">3 mins</span>
</a>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item">
<i class="fas fa-users mr-2"></i> 8 friend requests
<span class="float-right text-muted text-sm">12 hours</span>
</a>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item">
<i class="fas fa-file mr-2"></i> 3 new reports
<span class="float-right text-muted text-sm">2 days</span>
</a>
<div class="dropdown-divider"></div>
<a href="#" class="dropdown-item dropdown-footer">See All Notifications</a>
</div>
</li>
<li class="nav-item">
<a class="nav-link" data-widget="control-sidebar" data-slide="true" href="#"><i
class="fas fa-th-large"></i></a>
</li>
</ul>
</nav>
<!-- /.navbar -->
<!-- Main Sidebar Container -->
<aside class="main-sidebar sidebar-dark-primary elevation-4">
<!-- Brand Logo -->
<a href="#" class="brand-link">
<img src="{% static 'AdminLTE-master/dist/img/AdminLTELogo.png' %}" alt="AdminLTE Logo" class="brand-image img-circle elevation-3"
style="opacity: .8">
<span class="brand-text font-weight-light">AdminLTE 3</span>
</a>
<!-- Sidebar -->
<div class="sidebar">
<!-- Sidebar user panel (optional) -->
<div class="user-panel mt-3 pb-3 mb-3 d-flex">
<div class="image">
<img src="{% static 'AdminLTE-master/dist/img/user2-160x160.jpg' %}" class="img-circle elevation-2" alt="User Image">
</div>
<div class="info">
<a href="#" class="d-block">Alexander Pierce</a>
</div>
</div>
<!-- Sidebar Menu -->
<nav class="mt-2">
<ul class="nav nav-pills nav-sidebar flex-column" data-widget="treeview" role="menu" data-accordion="false">
<!-- Add icons to the links using the .nav-icon class
with font-awesome or any other icon font library -->
{# <li class="nav-item has-treeview menu-open">#}
{# <a href="#" class="nav-link active">#}
{# <i class="nav-icon fas fa-tachometer-alt"></i>#}
{# <p>#}
{# Starter Pages#}
{# <i class="right fas fa-angle-left"></i>#}
{# </p>#}
{# </a>#}
{# <ul class="nav nav-treeview">#}
{# <li class="nav-item">#}
{# <a href="#" class="nav-link active">#}
{# <i class="far fa-circle nav-icon"></i>#}
{# <p>Active Page</p>#}
{# </a>#}
{# </li>#}
{# <li class="nav-item">#}
{# <a href="#" class="nav-link">#}
{# <i class="far fa-circle nav-icon"></i>#}
{# <p>Inactive Page</p>#}
{# </a>#}
{# </li>#}
{# </ul>#}
{# </li>#}
<li class="nav-item">
<a href="{% url 'index' %}" class="nav-link">
<i class="nav-icon fas fa-th"></i>
<p>
接口列表
{# <span class="right badge badge-danger">New</span>#}
</p>
</a>
</li>
<li class="nav-item">
<a href="{% url 'echarts_msg' %}" class="nav-link">
<i class="nav-icon fas fa-th"></i>
<p>
数据统计
{# <span class="right badge badge-danger">New</span>#}
</p>
</a>
</li>
<li class="nav-item">
<a href="{% url 'crontab_index' %}" class="nav-link">
<i class="nav-icon fas fa-th"></i>
<p>
测试报告
{# <span class="right badge badge-danger">New</span>#}
</p>
</a>
</li>
</ul>
</nav>
<!-- /.sidebar-menu -->
</div>
<!-- /.sidebar -->
</aside>
<!-- Content Wrapper. Contains page content -->
<div class="content-wrapper">
<div class="container">
{% block content %}
{% endblock %}
</div>
<!-- /.content -->
</div>
<!-- /.content-wrapper -->
<!-- Control Sidebar -->
<aside class="control-sidebar control-sidebar-dark">
<!-- Control sidebar content goes here -->
<div class="p-3">
<h5>Title</h5>
<p>Sidebar content</p>
</div>
</aside>
<!-- /.control-sidebar -->
<!-- Main Footer -->
<footer class="main-footer">
<!-- To the right -->
<div class="float-right d-none d-sm-inline">
Anything you want
</div>
<!-- Default to the left -->
<strong>Copyright © 2014-2019 <a href="https://adminlte.io">AdminLTE.io</a>.</strong> All rights reserved.
</footer>
</div>
<!-- ./wrapper -->
<!-- REQUIRED SCRIPTS -->
<!-- jQuery -->
<script src="{% static 'AdminLTE-master/plugins/jquery/jquery.min.js' %}"></script>
<!-- Bootstrap 4 -->
<script src="{% static 'AdminLTE-master/plugins/bootstrap/js/bootstrap.bundle.min.js' %}"></script>
<!-- AdminLTE App -->
<script src="{% static 'AdminLTE-master/dist/js/adminlte.min.js' %}"></script>
<script src="{% static 'echarts.mim.js' %}"></script>
</body>
{% block js %}
{% endblock %}
</html>
case_add.html：
{% extends "base.html" %}
{% block content %}
<div class="row">
<div class="col-md-8 offset-md-2"><a href="{% url 'index' %}">返回主页</a></div>
<div class="col-md-8 offset-md-2">
{% include 'form.html' %}
</div>
</div>
{% endblock %}
case_edit.html：
{% extends "base.html" %}
{% block content %}
<div class="row">
<div class="col-md-8 offset-md-2"><a href="{% url 'index' %}">返回主页</a></div>
<div class="col-md-8 offset-md-2">
{% include 'form.html' %}
</div>
</div>
{% endblock %}
case_list.html：
{% extends "base.html" %}
{% block content %}
<form action="" method="post">
{% csrf_token %}
<div class="row">
{# <div class="col-md-12">#}
{# <a href="{% url 'api_add' %}" class="btn btn-success">添加接口</a>#}
{# </div>#}
<div>
<nav class="nav">
<a href="{% url 'index' %}" class="nav-link">返回主页</a>
<a href="{% url 'case_add' pk %}" class="nav-link">去创建</a>
<input type="submit" value="批量执行" class="btn btn-sm btn-danger">
<span style="color: green">{{ error }}</span>
</nav>
</div>
<div class="">
{% if case_obj %}
<table class="table table-hover">
<thead>
<tr>
<th>选择</th>
<th>序号</th>
<th>名称</th>
{# <th>描述</th>#}
<th>所属接口</th>
<th>预期值</th>
{# <th>url</th>#}
{# <th>参数</th>#}
<th>报告</th>
<th>类型</th>
<th>执行</th>
<th>通过</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% for foo in case_obj %}
<tr>
<td>
<input type="checkbox" value="{{ foo.pk }}" name="case_pk">
</td>
<td>{{ forloop.counter }}</td>
<td>{{ foo.case_title }}</td>
{# <td>{{ foo.case_desc }}</td>#}
<td>{{ foo.case_API }}</td>
<td>{{ foo.case_expect }}</td>
{# <td>{{ foo.case_url }}</td>#}
{# <td>{{ foo.case_params }}</td>#}
{% if foo.case_report %}
<td><a href="{% url 'download_case_report' foo.pk %}">下载</a></td>
{% else %}
<td>无</td>
{% endif %}
<td>{{ foo.case_method }}</td>
<td>{{ foo.get_case_execute_status_display }}</td>
<td>{{ foo.get_case_pass_status_display }}</td>
{# <td>1</td><!-- 接口下面有多少用例 -- >#}
{# <td>1</td><!-- 接口下面的用例有多少通过的，计算公式：通过/总数*100 -->#}
<td>
<a href="{% url 'case_edit' foo.pk %}" class="btn btn-success btn-sm">编辑用例</a>
<a href="{% url 'case_del' foo.pk %}" class="btn btn-danger btn-sm">删除用例</a>
<a href="{% url 'case_execute' foo.pk %}" class="btn btn-warning btn-sm">执行</a>
{# <a href="" class="btn btn-success btn-sm">批量导入</a>#}
{# <a href="" class="btn btn-success btn-sm">查看用例</a>#}
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
没有数据,<a href="{% url 'case_add' pk %}">去创建</a>
{% endif %}
</div>
</div>
</form>
{% endblock %}
crontab_index.html：
{% extends "base.html" %}
{% block content %}
<div class="row">
<table class="table table-hover table-striped">
<thead>
<tr>
<th>序号</th>
<th>接口</th>
<th>创建时间</th>
<th>覆盖率</th>
<th>用例数</th>
<th>报告</th>
</tr>
</thead>
<tbody>
{% for log in obj %}
<tr>
<td>{{ forloop.counter }}</td>
<td>{{ log.log_api.api_title }}</td>
<td>{{ log.log_creat_time | date:"Y-m-d H:i:s"}}</td>
<td>{{ log.log_api.xxoo }}</td>
<td>{{ log.log_api.case_set.count }}</td>
<td>
<a href="{% url 'show_crontablog_report' log.pk %}">查看</a>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
{% endblock %}
crontablog_report.html：
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<div>
<a href="{% url 'crontab_index' %}">返回上一页</a>
<a href="{% url 'download_crontab_log' pk %}">下载报告</a>
</div>
<hr>
<div>
{{ f|safe }}
</div>
</body>
</html>
echarts_msg.html：
{% extends 'base.html' %}
{% block content %}
<div class="row">
<div class="page-header">
<h1>近一年每月接口数量走势图</h1>
<div ).val()},
success: function (data) {
console.log(data);
area_basic(data['area_basic']);
bar_y_category_stack(data['bar_y_category_stack']);
pie_doughnut(data['pie_doughnut']);
pie_simple(data['pie_simple']);
area_stack(data['area_stack']);
}
})
}
</script>
{% endblock %}
form.html：
<form action="" method="post" novalidate>
{% csrf_token %}
{% for foo in form_obj %}
<div>
<label for="">{{ foo.label }}</label>
{{ foo }}
<span style="color: red;">{{ foo.errors.0 }}</span>
</div>
{% endfor %}
<input type="submit" value="提交" class="btn-success btn">
</form>
import_case.html：
{% extends 'base.html' %}
{% block content %}
<div class="row">
<div class="col-md-8 offset-md-2">
<form action="" method="post" enctype="multipart/form-data">
{% csrf_token %}
<input type="file" name="excel" class="form-control">
<input type="submit" value="提交" class="btn btn-success">
<span style="color: red">{{ error }}</span>
</form>
</div>
</div>
{% endblock %}
index.html：
{% extends "base.html" %}
{% block content %}
<div class="row">
<div class="col-md-12">
<a href="{% url 'api_add' %}" class="btn btn-success">添加接口</a>
</div>
<div class="col-md-12">
{% if api_obj %}
<table class="table table-hover">
<thead>
<tr>
<th>序号</th>
<th>接口名称</th>
{# <th>接口描述</th>#}
<th>开始时间</th>
<th>结束时间</th>
<th>数量</th>
<th>覆盖率</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% for foo in api_obj %}
<tr>
<td>{{ forloop.counter }}</td>
<td>{{ foo.api_title }}</td>
<td>{{ foo.api_start_time | date:"Y-m-d"}}</td>
<td>{{ foo.api_stop_time | date:"Y-m-d"}}</td>
<td>{{ foo.case_set.count }}</td> <!-- 接口下面有多少用例 -->
<td>{{ foo.xxoo }}</td> <!-- 接口下面的用例有多少通过的，计算公式：通过/总数*100 -->
<td>
<a href="{% url 'api_edit' foo.pk %}" class="btn btn-success btn-sm">编辑接口</a>
<a href="{% url 'api_del' foo.pk %}" class="btn btn-success btn-sm">删除接口</a>
<a href="{% url 'case_add' foo.pk %}" class="btn btn-success btn-sm">添加用例</a>
<a href="{% url 'import_case' foo.pk %}" class="btn btn-success btn-sm">批量导入</a>
<a href="{% url 'case_list' foo.pk %}" class="btn btn-success btn-sm">查看用例</a>
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
没有数据,<a href="">去创建</a>
{% endif %}
</div>
</div>
{% endblock %}
crontab.py：
"""
执行定时任务
"""
import os
import datetime
import django
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "MB.settings")
django.setup()
from app01 import models
from utils.execute_case import crontab_execute
from apscheduler.schedulers.blocking import BlockingScheduler
def task():
l = []
today = datetime.date.today()
api_obj = models.API.objects.all()
for i in api_obj:
# print(i.api_stop_time)
if today == i.api_stop_time:
l.append(i)
print(111, l, datetime.datetime.now())
crontab_execute(l)
# 每几秒执行一次：
# def job1():
# """定时任务开始"""
# print('job1', datetime.datetime.now())
# scheduler = BlockingScheduler()
# scheduler.add_job(job1, 'interval', seconds=5, id='job1') # 每隔5秒执行一次
# scheduler.start()
# 每天某一时刻执行一次：
# sc = BlockingScheduler()
# f = open('t1.text', 'a', encoding='utf8')
# @sc.scheduled_job('cron', day_of_week='*', hour=1, minute='30', second='50')
# def check_db():
# print(111111111111)
# 每隔几分钟执行一次：
# def job1():
# print('job1', datetime.datetime.now())
# scheduler = BlockingScheduler()
# # 每隔2分钟执行一次， */1：每隔1分钟执行一次
# scheduler.add_job(job1, 'cron', minute="*/2", id='job1')
# scheduler.start()
if __name__ == '__main__':
# print(datetime.datetime.now())
# today = datetime.date.today()
# print(today.year, today.month)
# foo()
# obj = models.CrontabLog.objects.all()
# print(obj)
scheduler = BlockingScheduler()
# 每隔2分钟执行一次， */1：每隔1分钟执行一次
scheduler.add_job(task, 'cron', minute="*/2", id='task')
scheduler.start()
echartsMsg.py：
import datetime
from dateutil.relativedelta import relativedelta
from django.db.models import Count
from app01 import models
def get_bar_option():
return {
"tooltip": {
"trigger": 'axis',
"axisPointer": {
"type": 'shadow'
}
},
"legend": {
"data": ['已通过', '未通过']
},
"grid": {
"left": '3%',
"right": '4%',
"bottom": '3%',
"containLabel": "true"
},
"xAxis": {
"type": 'value'
},
"yAxis": {
"type": 'category',
"data": ['周一', '周二', '周三', '周四', '周五', '周六', '周日']
},
"series": [
{
"name": '已通过',
"type": 'bar',
"stack": '总量',
"label": {
"show": True,
"position": 'insideRight'
},
"data": [320, 302, 301, 334, 390, 330, 320]
},
{
"name": '未通过',
"type": 'bar',
"stack": '总量',
"label": {
"show": True,
"position": 'insideRight'
},
"data": [120, 132, 101, 134, 90, 230, 210]
},
# {
# "name": '联盟广告',
# "type": 'bar',
# "stack": '总量',
# "label": {
# "show": True,
# "position": 'insideRight'
# },
# "data": [220, 182, 191, 234, 290, 330, 310]
# },
# {
# "name": '视频广告',
# "type": 'bar',
# "stack": '总量',
# "label": {
# "show": True,
# "position": 'insideRight'
# },
# "data": [150, 212, 201, 154, 190, 330, 410]
# },
# {
# "name": '搜索引擎',
# "type": 'bar',
# "stack": '总量',
# "label": {
# "show": True,
# "position": 'insideRight'
# },
# "data": [820, 832, 901, 934, 1290, 1330, 1320]
# }
]
}
def bar_y_category_stack():
# 一年前的今天
start_time = datetime.date.today() - relativedelta(months=12)
# print(start_time)
# 当前时间
now_time = datetime.date.today()
# print(now_time)
# 获取近一年的数据
data = models.API.objects.filter(api_start_time__range=(start_time, now_time))
# print(data)
res = data.extra(select={'api_start_time': "strftime('%%Y-%%m',api_start_time)"}).values('api_start_time').order_by(
'api_start_time')
# print(res)
d = {
'case_execute_status': [],
'date': [],
'executed': [],
'un_executed': [],
'pass': [],
'un_pass': []
}
# 已执行/未执行
for i in models.Case.case_execute_status_choices:
d['case_execute_status'].append(i[1])
for i in models.Case.case_pass_status_choices:
d['case_execute_status'].append(i[1])
for i in res:
d['date'].append(i['api_start_time'])
for i in data:
d['executed'].append(i.case_set.filter(case_execute_status=1).count())
d['un_executed'].append(i.case_set.filter(case_execute_status=0).count())
d['pass'].append(i.case_set.filter(case_pass_status=1).count())
d['un_pass'].append(i.case_set.filter(case_pass_status=0).count())
a = {
"tooltip": {
"trigger": 'axis',
"axisPointer": {
"type": 'shadow'
}
},
"legend": {
"data": d['case_execute_status']
},
"grid": {
"left": '3%',
"right": '4%',
"bottom": '3%',
"containLabel": "true"
},
"xAxis": {
"type": 'value'
},
"yAxis": {
"type": 'category',
"data": d['date']
},
"series": [
{
"name": '已通过',
"type": 'bar',
"stack": '总量',
"label": {
"show": True,
"position": 'insideRight'
},
"data": d['pass']
},
{
"name": '未通过',
"type": 'bar',
"stack": '总量',
"label": {
"show": True,
"position": 'insideRight'
},
"data": d['un_pass']
},
{
"name": '已执行',
"type": 'bar',
"stack": '总量',
"label": {
"show": True,
"position": 'insideRight'
},
"data": d['executed']
},
{
"name": '未执行',
"type": 'bar',
"stack": '总量',
"label": {
"show": True,
"position": 'insideRight'
},
"data": d['un_executed']
},
]
}
print(d)
return a, d
def area_basic():
# 一年前的今天
start_time = datetime.date.today() - relativedelta(months=12)
# print(start_time)
# 当前时间
now_time = datetime.date.today()
# print(now_time)
# 获取近一年的数据
data = models.API.objects.filter(api_start_time__range=(start_time, now_time))
# print(data)
res = data.extra(select={'api_start_time': "strftime('%%Y-%%m',api_start_time)"}).values('api_start_time').annotate(
count=Count('api_start_time')).order_by(
'api_start_time')
# print(res)
l = [[], []]
for i in res:
l[0].append(i['api_start_time'])
l[1].append(i['count'])
print(l)
return l
def pie_doughnut():
# 统计用例总数
'''
[
{value: 335, name: '直接访问'},
{value: 310, name: '邮件营销'},
{value: 234, name: '联盟广告'},
{value: 135, name: '视频广告'},
{value: 1548, name: '搜索引擎'}
]
:return:
'''
l = [{"value": 0, "name": "已执行"}, {"value": 0, "name": "未执行"}, ], ["已执行", '未执行']
api_obj = models.API.objects.all()
for api in api_obj:
l[0][0]['value'] += api.case_set.filter(case_execute_status=1).count()
l[0][1]['value'] += api.case_set.filter(case_execute_status=0).count()
# for api in api_obj:
# print(api.case_set.filter(case_pass_status=1).count(), api.case_set.filter(case_pass_status=0).count())
print(l)
return l
def pie_simple():
# 统计用例总数
'''
[
{value: 335, name: '直接访问'},
{value: 310, name: '邮件营销'},
{value: 234, name: '联盟广告'},
{value: 135, name: '视频广告'},
{value: 1548, name: '搜索引擎'}
]
:return:
'''
l = [{"value": 0, "name": "已通过"}, {"value": 0, "name": "未通过"}, ], ["已通过", '未通过']
api_obj = models.API.objects.all()
for api in api_obj:
l[0][0]['value'] += api.case_set.filter(case_pass_status=1).count()
l[0][1]['value'] += api.case_set.filter(case_pass_status=0).count()
# for api in api_obj:
# print(api.case_set.filter(case_pass_status=1).count(), api.case_set.filter(case_pass_status=0).count())
print(l)
return l
execute_case.py：
"""
获取用例对象
执行用例并返回
当获取到了对象之后
-发请求
-校验
-生成测试报告
-修改数据库字段
-下载
"""
import os
import requests
import json
import unittest
from io import BytesIO, StringIO
from bs4 import BeautifulSoup
from HTMLTestRunner import HTMLTestRunner
from MB.settings import BASE_DIR
from app01 import models
class MyTestCase(unittest.TestCase):
def test_case(self):
self.assertEqual(self.f, self.s)
class Worker(object):
def __init__(self, case_obj=None, case_list=[]):
self.case_obj = case_obj
self.case_list = case_list
def handler(self):
"""操作函数"""
# 1、发请求
response = self.send_msg()
# 2、校验
result = self.check_response(response)
# 3、断言
test_case_obj = self.assert_msg(result)
# 4、生成测试报告
self.create_report(test_case_obj)
# 5、修改数据库字段
self.update_db(result)
def update_db(self, result):
"""修改数据库字段
1.通过状态
2.执行状态
3.用例报告
"""
# print(111, result, " ", file_path)
if result[-1]["status"]: # 断言成功
models.Case.objects.filter(pk=self.case_obj.pk).update(
case_execute_status=1,
case_pass_status=1,
case_report=self.f.getvalue()
)
else:
models.Case.objects.filter(pk=self.case_obj).update(
case_execute_status=1,
case_pass_status=0,
case_report=self.f.getvalue()
)
# print(111, "修改数据库成功")
def assert_msg(self, result):
"""获取请求的校验结果，使用unittest做断言"""
test_case_obj = MyTestCase(methodName="test_case")
test_case_obj.f, test_case_obj.s, status = result
self.case_list.append(test_case_obj)
return test_case_obj
def generate_single_report(self, test_case_obj):
"""生成单个测试报告"""
suite = unittest.TestSuite()
suite.addTest(test_case_obj)
self.create_report(suite, title=self.case_obj.case_title, desc=self.case_obj.case_desc)
def generate_batch_report(self):
"""生成多个测试报告"""
suite = unittest.TestSuite()
suite.addTests(self.case_list)
self.create_report(suite, title="批量测试集合", desc="本次执行了{}个用例".format(suite.countTestCases()))
return self.f
def create_report(self, suite, title=None, desc=None):
"""生成测试报告"""
self.f = BytesIO()
HTMLTestRunner(
title=title,
description=desc,
stream=self.f
).run(suite)
def send_msg(self):
"""发请求"""
response = requests.request(
method=self.case_obj.case_method,
url=self.case_obj.case_url,
params=self._check_params(),
data=self._check_data(),
)
return response
def _check_text_response(self, response):
"""校验文本类型的返回值"""
# 期望值
expect = self.case_obj.case_expect
# 响应结果
text = response.text
# 使用bs4做校验
soup = BeautifulSoup(text, "html.parser")
title = soup.find(name="title").text
# print(222, title, expect)
if title != expect: # 断言失败
return title, expect, {"status": 0}
else:
return title, expect, {"status": 1}
def _check_application_response(self, response):
"""校验json类型的返回值"""
# 期望值
expect = json.loads(self.case_obj.case_expect)
# 响应结果
response = response.json()
# 校验
# print(expect, type(expect), json.loads(expect))
for k in expect:
if expect[k] != response[k]: # 断言失败
return {k: expect[k]}, {k: response[k]}, {"status": 0}
else: # 断言成功
return {k: expect[k]}, {k: response[k]}, {"status": 1}
def _check_data(self):
"""处理请求携带的data数据"""
"""处理data逻辑"""
return {}
def _check_params(self):
"""处理请求的参数"""
if self.case_obj.case_params:
return json.loads(self.case_obj.case_params)
else:
return {}
def check_response(self, response):
"""校验请求结果"""
# print(111, response.json())
"""
两种响应结果：
码云
111 text/javascript; charset=utf-8
v2EX项目
111 application/json;charset=UTF-8
"""
# 切一次取0
header = response.headers["Content-Type"].split("/", 1)[0]
# print(111, header)
if hasattr(self, "_check_{}_response".format(header)):
result = getattr(self, "_check_{}_response".format(header))(response)
else:
raise "不支持的响应类型"
print(111, result)
return result
def save_crontablog(self, api_obj):
"""将接口的测试报告写crontablog中"""
models.CrontabLog.objects.create(
log_api_id=api_obj.pk,
log_report=self.f.getvalue()
)
def execute(case_list_obj):
"""case_obj:从前端获取来的用例"""
# print(case_obj)
l = []
# file_path = Worker(case_obj).handler()
# return file_path
for case in case_list_obj:
Worker(case_obj=case, case_list=l).handler()
print(111, case)
# print(222, 1)
# 生成测试报告集
temp_f = Worker(case_list=l).generate_batch_report()
print(222, l)
return temp_f
def crontab_execute(api_list):
"""定时任务"""
for api in api_list:
# print(api.case_set.all())
l = []
for case in api.case_set.all():
w = Worker(case_obj=case, case_list=l)
w.handler()
# print(333, l)
# 1、为当前的接口生成测试报告
w1 = Worker(case_list=l) # 返回self.f
w1.generate_batch_report()
# 2、将测试报告存储到数据库：CrontabLog
w1.save_crontablog(api)
myForm.py：
from django import forms
from django.forms import widgets as wid
from app01 import models
class ApiForm(forms.ModelForm):
class Meta:
model = models.API
fields = "__all__"
# exclude = ["project_status"]
# labels = {
# "project_name": "项目名称",
# "project_detail": "项目描述",
# }
error_messages = {
"api_title": {"required": "不能为空"},
"api_desc": {"required": "不能为空"},
}
widgets = {
"api_title": wid.TextInput(attrs={"class": "form-control"}),
"api_desc": wid.Textarea(attrs={"class": "form-control"}),
"api_start_time": wid.DateInput(attrs={"class": "form-control", "type": "date"}),
"api_stop_time": wid.DateInput(attrs={"class": "form-control", "type": "date"}),
}
class CaseForm(forms.ModelForm):
class Meta:
model = models.Case
fields = "__all__"
exclude = ["case_report", "case_execute_status", "case_pass_status"]
# labels = {
# "project_name": "项目名称",
# "project_detail": "项目描述",
# }
error_messages = {
"case_title": {"required": "不能为空"},
"case_desc": {"required": "不能为空"},
"case_expect": {"required": "不能为空"},
"case_url": {"required": "不能为空"},
"case_method": {"required": "不能为空"},
"case_params": {"required": "不能为空"},
}
widgets = {
"case_API": wid.Select(attrs={"class": "form-control"}),
"case_title": wid.TextInput(attrs={"class": "form-control"}),
"case_desc": wid.TextInput(attrs={"class": "form-control"}),
"case_expect": wid.TextInput(attrs={"class": "form-control"}),
"case_url": wid.TextInput(attrs={"class": "form-control"}),
"case_method": wid.TextInput(attrs={"class": "form-control"}),
"case_params": wid.TextInput(attrs={"class": "form-control"}),
}
效果如下：
# 项目总结
-- unittest框架
- HTMLTestRunner:该插件用来生成unittest测试报告,且该插件暂时无法使用pip等方式下载,从网上sou现成的脚本,并且
该插件受Python的版本影响,2/3不兼容
下载地址:https://www.cnblogs.com/Neeo/articles/7942613.html
然后保存成py文件,放到Python解释器的第三方库目录中: PythonLibsite-packagesHTMLTestRunner.py
from HTMLTestRunner import HTMLTestRunner
- 参数化不太好使
-- 上传下载
- 上传
如果使用的是form提交的文件,别忘了form的属性enctype="multipart/form-data"
<form action="" method="post" enctype="multipart/form-data">
{% csrf_token %}
<input type="file" name="excel" class="form-control">
<input type="submit" value="提交" class="btn btn-success">
<span style="color: red;">{{ error }}</span>
</form>
后台使用 request.FILES.get("excel")
完事使用xlrd读取文件
- 下载
from django.http import FileResponse
from django.http import HttpResponse
from django.http import StreamingHttpResponse
重点:
log_obj = models.CrontabLog.objects.filter(pk=pk).first()
response = FileResponse(log_obj.log_report) # HttpResponse(log_obj.log_report)
response['Content-Type'] = 'application/octet-stream'
# filename的名称不能含有中文
response['Content-Disposition'] = 'attachment;filename="{}.html"'.format('report')
# print(111111, case_obj.case_title)
return response
-- modelform
from django import forms
from django.forms import widgets as wid
from app01 import models
class ApiForm(forms.ModelForm):
class Meta:
model = models.API
fields = "__all__"
error_messages = {
"api_title": {"required": "不能为空"},
"api_desc": {"required": "不能为空"}
}
widgets = {
"api_title": wid.TextInput(attrs={"class": "form-control"}),
"api_desc": wid.Textarea(attrs={"class": "form-control"}),
"api_start_time": wid.DateInput(attrs={"class": "form-control", 'type':"date"}),
"api_stop_time": wid.DateInput(attrs={"class": "form-control", 'type':"date"}),
}
-- modules.py中,
- 自定义字段
def xxoo(self):
if self.case_set.count():
a = "%.2f%% " % (self.case_set.filter(case_pass_status=1).count() / self.case_set.count() * 100)
return a
else:
return 0
- 自定义排序,根据字段排序
class CrontabLog(models.Model):
log_api = models.ForeignKey(to='API', verbose_name='所属接口')
log_creat_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
log_report = models.TextField(verbose_name='报告')
class Meta:
# 加 "-" 降序排序, 否则升序排序
ordering = ['-log_creat_time']
-- echarts: 如何安装和使用,参照上面的笔记
- 个人建议,首先在官网的示例中,先把你的数据代入进去,看效果,能行,再在后端处理数据格式
- 后端只负责数据的传递,前端渲染
- 建议使用ajax
- 跨域: https://www.cnblogs.com/Neeo/articles/11455271.html
-- 定时任务,参照上面的笔记
- 定时任务,https://www.cnblogs.com/Neeo/p/10435059.html
- 发邮件:https://www.cnblogs.com/Neeo/articles/11199085.html