Django 接口自动化测试平台

阅读原文时间：2021年07月21日阅读：1

本项目工程 github 地址：https://github.com/juno3550/InterfaceAutoTestPlatform

0. 引言

本接口自动化测试平台的整体功能如下图所示：

本项目环境如下：

python 3.6.5
pip install django==2.2.4
pip install redis==2.10.6
pip install eventlet==0.25.2
pip install celery==3.1.26.post2

对于 Django 的基础使用教程，可以参考本博客的《Django》系列博文。

1）创建本项目工程

# 方式一
django-admin startproject InterfaceAutoTest

方式二

python -m django startproject interface_test_platform

此时工程结构如下所示：

2）创建应用

在项目目录下，使用如下命令创建应用：

django-admin startapp interfacetestplatform

此时工程结构目录如下：

3）注册应用

在项目目录 InterfaceAutoTest/settings.py 中，找到 INSTALLED_APPS 配置项，在列表末尾添加新建的应用名称“interfacetestplatform”：

INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'interfacetestplatform,'
]

新建的应用需要在 INSTALLED_APPS 配置项中注册，Django 会根据该配置项到应用下查找模板等信息。

4）使用 Bootstrap

Bootstrap 是流行的 HTML、CSS 和 JS 框架，用于开发响应式布局、移动设备优先的 Web 项目。

下载 bootstrap：https://v3.bootcss.com/getting-started/#download
下载 jquery：https://jquery.com/download/

在项目目录下新建一个 static 目录，将解压后的如 bootstrap-3.3.7-dist 目录整体拷贝到 static 目录中，并将文件名改为 bootstrap。

由于 bootstrap 依赖 jQuery，我们需要提前下载并引入 jQuery。在 static 目录下，新建 css 和 js 目录，作为后面的样式文件和 js 文件的存放地，将我们的 jQuery 文件拷贝到 static/js/ 目录下。

ECharts 是一个使用 JavaScript 实现的开源可视化库，在本项目中用于用例/集合的执行结果统计可视化。于是我们引入 echarts 文件到 static/js/ 目录下。

此时 static 目录结构如下图所示：

在项目目录下的 settings.py 文件的末尾添加如下配置，用于指定静态文件的搜索目录：

STATICFILES_DIRS = [
os.path.join(BASE_DIR, "static"),
]

通过 STATICFILES_DIRS 属性，可以指定静态资源的路径，此处配置的是项目根目录下的 static 文件夹下。

默认情况下，Django 只能识别项目应用目录下的 static 目录的静态资源，不会识别到项目目录下的 static目录，因此通过 STATICFILES_DIR 属性可以解决这个问题。

5）创建应用的模板目录

在应用目录下新建一个 templates 目录，专门存放本应用所有的模板文件。

默认情况下，Django 只能识别项目应用目录下的 templates 目录的静态资源。本工程因仅涉及一个应用，因此选用此方案。

若应用较多，为了易于维护，可将各应用的模板文件进行统一处理，则在项目目录下新建 templates 目录，并在 settings.py 中新增配置：

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',
],
},
},
]

6）启动 Django 的 Web 服务

在项目目录下，运行如下命令启动端口号为 8000 的 Django 的 Web 服务：

python manage.py runsever 8000 # 端口号可自定义，不写则默认8000

启动日志如下：

You have 18 unapplied migration(s). Your project may not work properly until you apply the migrations for app(s): admin, auth, contenttypes, sessions.
Run 'python manage.py migrate' to apply them.
July 13, 2021 - 10:03:23
Django version 3.2.5, using settings 'InterfaceAutoTest.settings'
Starting development server at http://127.0.0.1:8000/
Quit the server with CTRL-BREAK.

启动成功后，用浏览器访问 127.0.0.1:8000，可以看到 Django 默认的页面，如下所示即代表服务启动成功：

7）配置 Mysql 数据库

修改项目目录下的 settings.py 文件，将 DATABASES 配置项改为如下配置：

DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'NAME': 'interface_platform', # 已存在的库名
'USER': 'root', # 数据库账号
'PASSWORD': '123', # 数据库密码
'HOST': '127.0.0.1', # 数据库IP
'PORT': '3306', # 数据库端口
}
}

若使用的是 Python 3 的 pymysql，则需要在项目的 __init__.py 文件中添加如下内容，避免报错：

import pymysql
pymysql.version_info = (1, 4, 13, "final", 0) # 指定版本。在出现“mysqlclient 1.4.0 or newer is required; you have 0.9.3.”报错时加上此行
pymysql.install_as_MySQLdb()

8）创建 Admin 站点超级用户

首先在项目目录下进行数据迁移，生成 auth_user 等基础表：

python manage.py migrate

在工程目录下，执行以下命令创建超级用户，用于登录 Django 的 Admin 站点：

D:\InterfaceAutoTest>python manage.py createsuperuser
Username (leave blank to use 'administrator'): admin # 用户名
Email address: admin@123.com # 符合邮箱格式即可
Password: # 密码
Password (again): # 二次确认密码
The password is too similar to the username.
This password is too common.
Bypass password validation and create user anyway? [y/N]: y
Superuser created successfully.

创建好 superuser 后，访问 127.0.0.1:8000/admin，用注册的 admin 账户进行登录（数据参考的就是 auth_user 表，并且必须是管理员用户才能进入）。

配置 admin 语言和时区

在 settings.py 中，找到 MIDDLEWARE 属性，添加中间件。

MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.locale.LocaleMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

同时，在后续添加项目信息表数据时，可以发现数据的创建时间、更新时间字段的时间不是本地时间，而是 UTC 时间，那么怎么展示为国内本地时间呢？

也是 settings.py 中，修改 TIME_ZONE 和 USE_TZ 的配置项：

TIME_ZONE = 'Asia/Shanghai'

USE_TZ = False

这样配置后，admin 页面中数据表的时间字段以及前端模板中的时间字段都是本地时区的时间。

当 USE_TZ 为 False 时，TIME_ZONE值是 Django 存储所有日期时间的时区。
当 USE_TZ 为 True 时，TIME_ZONE值是 Django 在模板中显示日期时间和解释表单中输入的日期时间的默认时区。简单起见，USE_TZ 直接设为 False。

1. 登录功能

预期效果如下：

1）路由定义规则说明

路由的基本信息包含路由地址和视图函数，路由地址即访问网址，视图函数即客户端在访问指定网址后，服务器端所对应的处理逻辑。

在应用目录（interfacetestplatform 目录）下新建 urls.py，将所有属于该应用的路由都写入该文件中，这样更容易管理和区分每个应用的路由地址，而项目目录（InterfaceAutoTest）下的 urls.py 是将每个应用的 urls.py 进行统一管理。

这种路由设计模式是 Django 常用的，其工作原理如下：

运行 InterfaceAutoTest 项目时，Django 从 interfacetestplatform （应用）目录的 urls.py 找到对应应用所定义的路由信息，生成完整的路由表。
当用户在浏览器上访问某个路由地址时，Django 服务端就会收到该用户的请求信息。Django 从当前请求信息获取路由地址，首先匹配项目目录下的 urls.py 的路由列表。转发到指定应用的 urls.py 的路由列表。
再执行应用下的路由信息所指向的视图函数，从而完成整个请求响应过程。

2）路由配置

配置项目 urls.py：InterfaceAutoTest/urls.py

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
path('admin/', admin.site.urls), # 指向内置Admin后台系统的路由文件sites.py
path('', include('interfacetestplatform.urls')), # 指向interfacetestplatform应用的urls.py
]

由于默认地址分发给了 interfacetestplatform（应用）的 urls.py 进行处理，因此需要对 interfacetestplatform/urls.py 编写路由信息，代码如下：

from django.urls import path
from . import views

urlpatterns = [
path('', views.index),
path('login/', views.login),
path('logout/', views.logout),
]

如路由信息 path('', views.index) 的 views.index 是指专门处理网站默认页的用户请求和响应过程的视图函数名称 index，其他路由规则原理一样。

1）定义 Form 表单类

在定义视图函数之前，我们先定义 Django 提供的表单模型类，来代替原生的前端 Form 表单。

在应用目录下，新建 form.py：

1 from django import forms
2
3
4 class UserForm(forms.Form):
5 username = forms.CharField(label="用户名", max_length=128, widget=forms.TextInput(attrs={'class': 'form-control'}))
6 password = forms.CharField(label="密码", max_length=256, widget=forms.PasswordInput(attrs={'class': 'form-control'}))

2）定义视图函数

在应用的 interfacepestplatform/views.py （视图模块）中添加如下代码：

1 from django.shortcuts import render, redirect
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from .form import UserForm
5 import traceback
6
7
8 # Create your views here.
9 # 默认页的视图函数
10 @login_required
11 def index(request):
12 return render(request, 'index.html')
13
14
15 # 登录页的视图函数
16 def login(request):
17 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
18 if request.session.get('is_login', None):
19 return redirect('/')
20 # 如果是表单提交行为，则进行登录校验
21 if request.method == "POST":
22 login_form = UserForm(request.POST)
23 message = "请检查填写的内容！"
24 if login_form.is_valid():
25 username = login_form.cleaned_data['username']
26 password = login_form.cleaned_data['password']
27 try:
28 # 使用django提供的身份验证功能
29 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配成功则返回用户对象；反之返回None
30 if user is not None:
31 print("用户【%s】登录成功" % username)
32 auth.login(request, user)
33 request.session['is_login'] = True
34 # 登录成功，跳转主页
35 return redirect('/')
36 else:
37 message = "用户名不存在或者密码不正确！"
38 except:
39 traceback.print_exc()
40 message = "登录程序出现异常"
41 # 用户名或密码为空，返回登录页和错误提示信息
42 else:
43 return render(request, 'login.html', locals())
44 # 不是表单提交，代表只是访问登录页
45 else:
46 login_form = UserForm()
47 return render(request, 'login.html', locals())
48
49
50 # 注册页的视图函数
51 def register(request):
52 return render(request, 'register.html')
53
54
55 # 登出的视图函数：重定向至login视图函数
56 @login_required
57 def logout(request):
58 auth.logout(request)
59 request.session.flush()
60 return redirect("/login/")

render 方法

用来生成网页内容并返回给客户端，其两个必须参数表示：第一个参数是浏览器想服务器发送的请求对象；第二个参数是模板名称，用于生成网页内容。

auth 模块

auth 模块是 cookie 和 session 的升级版。

auth 模块是对登录认证方法的一种封装，之前我们获取用户输入的用户名及密码后需要自己从 user 表里查询有没有用户名和密码符合的对象，而有了 auth 模块之后就可以很轻松地去验证用户的登录信息是否存在于数据库（auth_user 表）中。

除此之外，auth 还对 session 做了一些封装，方便我们校验用户是否已登录。

login 方法

对于非 POST 方法的请求（如 GET 请求），直接返回空的表单，让用户可以填入数据。

对于 POST 方法请求，接收表单数据，并验证：

使用表单类自带的 is_valid() 方法进一步完成数据验证工作；
验证成功后可以从表单对象的 cleand_data 数据字典中获取表单的具体值；
如果验证不通过，则返回一个包含先前数据的表单给前端页面，方便用户修改。也就是说，它会帮你保留先前填写的数据内容，而不是返回一个空表。
另外，这里使用了一个小技巧，python 内置了一个 locals() 函数，它返回当前所有的本地变量字典，我们可以偷懒的将这作为 render 函数的数据字典参数值，就不用费劲去构造一个形如 {'message':message, 'login_form':login_form} 的字典了。这样做的好处是大大方便了我们；但同时也可能往模板传入一些多余的变量数据，造成数据冗余降低效率。

@login_required

为函数增加装饰器 @login_required，这种方式可以实现未登录禁止访问首页的功能。

此种方式需要在项目 settings.py 中添加如下配置：

LOGIN_URL = '/login/'

通过 LOGIN_URL 告诉 Django 在用户没登录的情况下，重定向的登录地址；如果没配置该属性，Django 会重定向到默认地址。

如果不用装饰器方式，也可以在函数内部判断用户状态，并实现重定向。如下所示，效果是一样的：

def index(request):
print("request.user.is_authenticated: {}".format(request.user.is_authenticated))
if not request.user.is_authenticated:
return redirect("/login/")
return render(request,'index.html')

1）定义 base 基础模板，供其他模板继承

新建 interfacetestplatform/templates/base.html，用作平台前端的基础模板，代码如下：

1
2 3 {% load static %} 4 5 6 7 8 9 {% block title %}base{% endblock %} 10 11 12 13 14 15 16 20 {% block css %}{% endblock %} 21 22 23

25 26 36 37 51

53 54 {% block content %}{% endblock %} 55 56 57 58 59 60 61 62

在 base 模板中，通过 if 判断，当登录成功时，显示当前用户名和登出按钮；未登录时，则显示登录按钮。

注意，request 这个变量是默认被传入模板中的，可以通过圆点的调用方式获取对象中的属性，如 {{ reques.user.username }} 表示当前登录的用户名。

首先在 static/css/ 目录中，新建一个 login.css 样式文件，配置一些简单样式：

1 body {
2 background-color: #eee;
3 }
4 .form-login {
5 max-width: 330px;
6 padding: 15px;
7 margin: 0 auto;
8 }
9 .form-login .form-control {
10 position: relative;
11 height: auto;
12 -webkit-box-sizing: border-box;
13 -moz-box-sizing: border-box;
14 box-sizing: border-box;
15 padding: 10px;
16 font-size: 16px;
17 }
18 .form-login .form-control:focus {
19 z-index: 2;
20 }
21 .form-login input[type="text"] {
22 margin-bottom: -1px;
23 border-bottom-right-radius: 0;
24 border-bottom-left-radius: 0;
25 }
26 .form-login input[type="password"] {
27 margin-bottom: 10px;
28 border-top-left-radius: 0;
29 border-top-right-radius: 0;
30 }

然后新建 templates/login.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}登录{% endblock %}
4 {% block css %}
5
6 {% endblock %}
7
8 {% block content %}
9

11
33

35 {% endblock %}

通过 “{% extends 'base.html' %}” 继承了“base.html”模板的内容。
通过 “{% block title %} 登录 {% endblock %}” 设置了专门的 title。
若页面提示 CSRF 验证失败，我们需要在前端页面的 form 表单内添加一个 {% csrf_token %} 标签，CSRF（Cross-site request forgery）跨站请求伪造，是一种常见的网络攻击手段。为此，Django 自带了许多常见攻击手段的防御机制，CSRF 就是其中一种，还有防 XSS、SQL 注入等。
通过 “{% block css %}”引入了针对性的 login.css 样式文件。
主体内容定义在“{% block content %}”中。
使用 {{ login_form.name_of_field }} 方式分别获取每一个字段，然后分别进行渲染。

3）定义 index 首页模板

新增 templates/index.html：

1 {% extends 'base.html' %}
2 {% block title %}主页{% endblock %}
3 {% block content %}
4 {% if request.user.is_authenticated %}
5

你好，{{ request.user.username }}！欢迎回来！

6 {% endif %}
7 {% endblock %}

根据登录状态的不同，显示不同的内容。

2. 项目

预期效果如下：

Django 的模型类提供以面向对象的方式管理数据库数据，即 ORM（关系对象映射）的思想。

1）编写模型类

打开应用目录下的 models.py，定义项目信息的模型类：

字段说明：

name：项目名称，必填，最长不超过 128 个字符，并且唯一，也就是不能有相同项目名称。
proj_owner：项目负责人，必填，最长不超过 20 个字符（实际可能不需要这么长）。
test_owner：测试负责人。必填，最长不超过 20 个字符。
dev_owner：开发负责人，必填，最长不超过 20 个字符。
desc ：项目描述，选填，最长不超过 100 个字符。
create_time：创建时间，必填，默认使用当前时间。
update_time：更新时间，选填，默认为当前时间。
使用 __str__ 更友好地打印对象信息。
Meta 类属性 verbose_name 指定了后台 Admin 页面展示的表名称。

2）数据迁移

在项目目录下，执行以下两个命令进行数据迁移（将模型类转换成数据库表）：

python manage.py makemigrations # 生成迁移文件（模型类的信息）
python manage.py migrate # 执行开始迁移（将模型类信息转换成数据库表）

从数据库迁移过程来看，迁移文件的内容是 Django 与 Mysql 之间的沟通语言，Django 通过 ORM（对象关系映射）功能，把模型类生成的迁移文件，映射到 Mysql 中，Mysql 通过迁移文件内容，创建对应的数据库表信息。

ORM 的好处就是，我们不需要编写 SQL 语句，而是通过 Django 的模型类，就可以在 Mysql 中创建和操作想要的表对象及其数据。

除了模型类中定义的表，初次执行迁移命令也会生成 Django 内置的表，这些表是服务于 Django 内置的功能。

2）Admin 站点增加项目数据

在应用目录下的 admin.py 中，添加如下代码：

from django.contrib import admin
from .import models

class ProjectAdmin(admin.ModelAdmin):
list_display = ("id", "name", "proj_owner", "test_owner", "dev_owner", "desc", "create_time", "update_time")

admin.site.register(models.Project, ProjectAdmin)

admin.site.register() 方法把模型类 Project 和 ProjectAdmin 进行绑定，并注册到 Admin 系统。通过 ProjectAdmin 类中的 list_display 字段，定义了后台 Admin 页面展示 Project 表的字段列表。

进入 Admin 页面并添加数据：

可以看到，在应用名称下展示了“项目信息表”，接着可以依次点击“项目信息表”—“ADD 项目信息表 +”—填写所需新增的项目信息—“SAVE”，完成项目信息的数据添加。

添加项目管理页面的路由信息：

from django.urls import path
from . import views

urlpatterns = [
path('',views.index),
path('login/', views.login),
path('logout/', views.logout),
path('project/', views.project, name="project"),
]

1 from django.shortcuts import render, redirect
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from .form import UserForm
5 import traceback
6 from .models import Project
7
8
9 # 项目管理页的视图函数
10 @login_required
11 def project(request):
12 print("request.user.is_authenticated: ", request.user.is_authenticated) # 打印用户是否已登录
13 projects = Project.objects.filter().order_by('-id') # 使用负id是为了倒序取出项目数据
14 print("projects:", projects) # 打印项目名称
15 return render(request, 'project.html', {'projects': projects})
16
17
18 # 默认页的视图函数
19 @login_required
20 def index(request):
21 return render(request, 'index.html')
22
23
24 # 登录页的视图函数
25 def login(request):
26 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
27 if request.session.get('is_login', None):
28 return redirect('/')
29 # 如果是表单提交行为，则进行登录校验
30 if request.method == "POST":
31 login_form = UserForm(request.POST)
32 message = "请检查填写的内容！"
33 if login_form.is_valid():
34 username = login_form.cleaned_data['username']
35 password = login_form.cleaned_data['password']
36 try:
37 # 使用django提供的身份验证功能
38 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
39 if user is not None:
40 print("用户【%s】登录成功" % username)
41 auth.login(request, user)
42 request.session['is_login'] = True
43 # 登录成功，跳转主页
44 return redirect('/')
45 else:
46 message = "用户名不存在或者密码不正确！"
47 except:
48 traceback.print_exc()
49 message = "登录程序出现异常"
50 # 用户名或密码为空，返回登录页和错误提示信息
51 else:
52 return render(request, 'login.html', locals())
53 # 不是表单提交，代表只是访问登录页
54 else:
55 login_form = UserForm()
56 return render(request, 'login.html', locals())
57
58
59 # 注册页的视图函数
60 def register(request):
61 return render(request, 'register.html')
62
63
64 # 登出的视图函数：重定向至login视图函数
65 @login_required
66 def logout(request):
67 auth.logout(request)
68 request.session.flush()
69 return redirect("/login/")

1）新增 templates/project.html 模板，用于展示项目信息：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}项目{% endblock %}
4
5 {% block content %}
6

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for project in projects %} 24 25 26 27 28 29 30 31 32 33 34 35 {% endfor %} 36 37

id	项目名称	项目负责人	测试负责人	开发负责人	简要描述	创建时间	更新时间	测试结果统计
{{ project.id }}	{{ project.name }}	{{ project.proj_owner }}	{{ project.test_owner }}	{{ project.dev_owner }}	{{ project.desc }}	{{ project.create_time\|date:"Y-n-d H:i" }}	{{ project.update_time\|date:"Y-n-d H:i" }}	查看

39 {% endblock %}

2）调整 base.html 模板：导航栏增加项目菜单，把“主页”菜单改为“项目”，同时把“自动化测试平台”菜单连接调整为”/”，即首页页面。

1
2 3 {% load static %} 4 5 6 7 8 9 {% block title %}base{% endblock %} 10 11 12 13 14 15 16 17 21 {% block css %}{% endblock %} 22 23 24

26 27 37 38 52

54 55 {% block content %}{% endblock %} 56 57 58 59 60 61 62 63

1）新增封装分页处理对象的视图函数：

2）修改 project.html 模板，新增分页处理内容：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}主页{% endblock %}
4
5 {% block content %}
6

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for project in projects %} 24 25 26 27 28 29 30 31 32 33 34 35 {% endfor %} 36 37

id	项目名称	项目负责人	测试负责人	开发负责人	简要描述	创建时间	更新时间	测试结果统计
{{ project.id }}	{{ project.name }}	{{ project.proj_owner }}	{{ project.test_owner }}	{{ project.dev_owner }}	{{ project.desc }}	{{ project.create_time\|date:"Y-n-d H:i" }}	{{ project.update_time\|date:"Y-n-d H:i" }}	查看

39
40 {# 实现分页标签的代码 #}
41 {# 这里使用 bootstrap 渲染页面 #}
42

43 59

60 {% endblock %}

3. 模块

预期效果如下：

模块管理的实现思路与项目管理相似，我们依然遵循模型类 —> admin 增加数据 —> 路由 —> 视图函数 —> 模板文件的步骤来实现。

1）应用的 models.py 中增加 Module 模型类：

1 from django.db import models
2
3
4 class Project(models.Model):
5 id = models.AutoField(primary_key=True)
6 name = models.CharField('项目名称', max_length=50, unique=True, null=False)
7 proj_owner = models.CharField('项目负责人', max_length=20, null=False)
8 test_owner = models.CharField('测试负责人', max_length=20, null=False)
9 dev_owner = models.CharField('开发负责人', max_length=20, null=False)
10 desc = models.CharField('项目描述', max_length=100, null=True)
11 create_time = models.DateTimeField('项目创建时间', auto_now_add=True)
12 update_time = models.DateTimeField('项目更新时间', auto_now=True, null=True)
13
14 # 打印时返回项目名称
15 def __str__(self):
16 return self.name
17
18 class Meta:
19 verbose_name = '项目信息表'
20 verbose_name_plural = '项目信息表'
21
22
23 class Module(models.Model):
24 id = models.AutoField(primary_key=True)
25 name = models.CharField('模块名称', max_length=50, null=False)
26 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE)
27 test_owner = models.CharField('测试负责人', max_length=50, null=False)
28 desc = models.CharField('简要描述', max_length=100, null=True)
29 create_time = models.DateTimeField('创建时间', auto_now_add=True)
30 update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
31
32 def __str__(self):
33 return self.name
34
35 class Meta:
36 verbose_name = '模块信息表'
37 verbose_name_plural = '模块信息表'

2）数据迁移

在项目目录下，执行以下两个命令进行数据迁移（将模型类转换成数据库表）：

python manage.py makemigrations # 生成迁移文件（模型类的信息）
python manage.py migrate # 执行开始迁移（将模型类信息转换成数据库表）

1）注册模型类到 admin：

2）登录 admin 系统，添加模块数据

访问：http://127.0.0.1:8000/admin/，进入模块信息表，添加数据。

新增应用 urls.py 的路由配置：

from django.urls import path
from . import views

urlpatterns = [
path('',views.index),
path('login/', views.login),
path('logout/', views.logout),
path('project/', views.project, name="project"),
path('module/', views.module, name="module"),
]

路由地址“module”对应的视图函数，指向 views.py 中的 module 方法，下一步我们添加下该方法的处理。

1）新增 templates/module.html 模板：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}模块{% endblock %}
4
5 {% block content %}
6

11
12

13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 {% for module in modules %} 30 31 32 33 34 35 36 37 38 39 40 {% endfor %} 41 42 43

id	模块名称	所属项目	测试负责人	模块描述	创建时间	更新时间	测试结果统计
{{ module.id }}	{{ module.name }}	{{ module.belong_project.name }}	{{ module.test_owner }}	{{ module.desc }}	{{ module.create_time\|date:"Y-n-d H:i" }}	{{ module.update_time\|date:"Y-n-d H:i" }}	查看

45
46 {# 实现分页标签的代码 #}
47 {# 这里使用 bootstrap 渲染页面 #}
48

49 65

66 {% endblock %}

2）修改 base.html 模板，新增模块菜单栏：

1
2 3 {% load static %} 4 5 6 7 8 9 {% block title %}base{% endblock %} 10 11 12 13 14 15 16 17 21 {% block css %}{% endblock %} 22 23 24

26 27 37 38 53

55 56 {% block content %}{% endblock %} 57 58 59 60 61 62 63 64

4. 测试用例

1）在应用 models.py 中增加 TestCase 模型类：

1 from django.db import models
2 from smart_selects.db_fields import GroupedForeignKey # 后台级联选择
3 from django.contrib.auth.models import User
4
5
6 class Project(models.Model):
7 id = models.AutoField(primary_key=True)
8 name = models.CharField('项目名称', max_length=50, unique=True, null=False)
9 proj_owner = models.CharField('项目负责人', max_length=20, null=False)
10 test_owner = models.CharField('测试负责人', max_length=20, null=False)
11 dev_owner = models.CharField('开发负责人', max_length=20, null=False)
12 desc = models.CharField('项目描述', max_length=100, null=True)
13 create_time = models.DateTimeField('项目创建时间', auto_now_add=True)
14 update_time = models.DateTimeField('项目更新时间', auto_now=True, null=True)
15
16 def __str__(self):
17 return self.name
18
19 class Meta:
20 verbose_name = '项目信息表'
21 verbose_name_plural = '项目信息表'
22
23
24 class Module(models.Model):
25 id = models.AutoField(primary_key=True)
26 name = models.CharField('模块名称', max_length=50, null=False)
27 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE)
28 test_owner = models.CharField('测试负责人', max_length=50, null=False)
29 desc = models.CharField('简要描述', max_length=100, null=True)
30 create_time = models.DateTimeField('创建时间', auto_now_add=True)
31 update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
32
33 def __str__(self):
34 return self.name
35
36 class Meta:
37 verbose_name = '模块信息表'
38 verbose_name_plural = '模块信息表'
39
40
41 class TestCase(models.Model):
42 id = models.AutoField(primary_key=True)
43 case_name = models.CharField('用例名称', max_length=50, null=False) # 如 register
44 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE, verbose_name='所属项目')
45 belong_module = GroupedForeignKey(Module, "belong_project", on_delete=models.CASCADE, verbose_name='所属模块')
46 request_data = models.CharField('请求数据', max_length=1024, null=False, default='')
47 uri = models.CharField('接口地址', max_length=1024, null=False, default='')
48 assert_key = models.CharField('断言内容', max_length=1024, null=True)
49 maintainer = models.CharField('编写人员', max_length=1024, null=False, default='')
50 extract_var = models.CharField('提取变量表达式', max_length=1024, null=True) # 示例：userid||userid": (\d+)
51 request_method = models.CharField('请求方式', max_length=1024, null=True)
52 status = models.IntegerField(null=True, help_text="0：表示有效，1：表示无效，用于软删除")
53 created_time = models.DateTimeField('创建时间', auto_now_add=True)
54 updated_time = models.DateTimeField('更新时间', auto_now=True, null=True)
55 user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name='责任人', null=True)
56
57 def __str__(self):
58 return self.case_name
59
60 class Meta:
61 verbose_name = '测试用例表'
62 verbose_name_plural = '测试用例表'

GroupedForeignKey 可以支持在 admin 新增数据时，展示该模型类的关联表数据。（需提前安装：pip install django-smart-selects）

2）数据迁移

在项目目录下，执行以下两个命令进行数据迁移（将模型类转换成数据库表）：

python manage.py makemigrations # 生成迁移文件（模型类的信息）
python manage.py migrate # 执行开始迁移（将模型类信息转换成数据库表）

1）注册模型类到 admin

应用 admin.py 文件中增加如下代码：注册 TestCase 模型类到 admin 后台系统。

2）登录 admin 系统，添加用例数据

访问 http://127.0.0.1:8000/admin/，进入测试用例表，添加数据：

添加用例数据时，页面如下：

所属项目和所属模块下拉选项是根据模型类中的 GroupedForeignKey 属性生成的，方便我们正确的关联数据。
请求数据、断言内容、提取变量表达式等字段的定义，需要根据接口业务逻辑，以及后续运行逻辑的设计来输入。

添加数据后如下所示：

应用 urls.py：

from django.urls import path
from . import views

1）新增 templates/test_case.html 模板：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试用例{% endblock %}
4
5 {% block content %}
6

7 {% csrf_token %} 8

9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 {% for test_case in test_cases %} 25 26 27 28 29 30 31 32 33 34 35 {% endfor %} 36 37

用例名称	所属项目	所属模块	接口地址	请求方式	请求数据	断言key	提取变量表达式
{{ test_case.case_name }}	{{ test_case.belong_project.name }}	{{ test_case.belong_module.name }}	{{ test_case.uri }}	{{ test_case.request_method }}	{{ test_case.request_data }}	{{ test_case.assert_key }}	{{ test_case.extract_var }}

38 39

41 {# 实现分页标签的代码 #}
42 {# 这里使用 bootstrap 渲染页面 #}
43

44 60

61 {% endblock %}

2）修改 base.html 模板：增加测试用例菜单。

1
2 3 {% load static %} 4 5 6 7 8 9 {% block title %}base{% endblock %} 10 11 12 13 14 15 16 17 21 {% block css %}{% endblock %} 22 23 24

26 27 37 38 54

56 57 {% block content %}{% endblock %} 58 59 60 61 62 63 64 65

目前用例列表中的字段仅包含用例的基本信息，下面继续加一下（点击用例名称跳转）用例详情页面，用于展示用例的全部字段信息，如创建时间、更新时间、维护人、创建人。

1）新增用例详情的路由配置

from django.urls import path, re_path
from . import views

由于用例详情路由地址需要传路由变量“test_case_id”，该变量需要通过正则表达式进行匹配，在 Django2.0 后，路由地址用到正则时需要用到 re_path 来解析，便于 Django 正确的匹配视图函数，以及在浏览器地址栏正确的展示 url 地址。

2）增加视图函数

3）新增 templates/test_case_detail.html 模板

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例详情{% endblock %}
4
5 {% block content %}
6

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42

id	接口名称	所属项目	所属模块	接口地址	请求数据	断言内容	编写人员	提取变量表达式	维护人	创建人	创建时间	更新时间
{{ test_case.id }}	{{ test_case.case_name }}	{{ test_case.belong_project }}	{{ test_case.belong_module }}	{{ test_case.uri }}	{{ test_case.request_data }}	{{ test_case.assert_key }}	{{ test_case.maintainer }}	{{ test_case.extract_var }}	{{ test_case.maintainer }}	{{ test_case.user.username }}	{{ test_case.created_time\|date:"Y-n-d H:i" }}	{{ test_case.updated_time\|date:"Y-n-d H:i" }}

44 {% endblock %}

4）修改 test_case.html 模板

添加用例名称的链接为用例详情路由地址：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试用例{% endblock %}
4
5 {% block content %}
6

7 {% csrf_token %} 8

9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 {% for test_case in test_cases %} 25 26 27 28 29 30 31 32 33 34 35 {% endfor %} 36 37

用例名称	所属项目	所属模块	接口地址	请求方式	请求数据	断言key	提取变量表达式
{{ test_case.case_name }}	{{ test_case.belong_project.name }}	{{ test_case.belong_module.name }}	{{ test_case.uri }}	{{ test_case.request_method }}	{{ test_case.request_data }}	{{ test_case.assert_key }}	{{ test_case.extract_var }}

38 39

41 {# 实现分页标签的代码 #}
42 {# 这里使用 bootstrap 渲染页面 #}
43

44 60

61 {% endblock %}

1）新增模块页面的用例路由配置：

from django.urls import path, re_path
from . import views

2）新增视图函数：

3）修改模块的模板文件：在模块名称链接中，添加对应测试用例的路由地址。

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}模块{% endblock %}
4
5 {% block content %}
6

11
12

13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 {% for module in modules %} 30 31 32 33 34 35 36 37 38 39 40 {% endfor %} 41 42 43

id	模块名称	所属项目	测试负责人	模块描述	创建时间	更新时间	测试结果统计
{{ module.id }}	{{ module.name }}	{{ module.belong_project.name }}	{{ module.test_owner }}	{{ module.desc }}	{{ module.create_time\|date:"Y-n-d H:i" }}	{{ module.update_time\|date:"Y-n-d H:i" }}	查看

45
46 {# 实现分页标签的代码 #}
47 {# 这里使用 bootstrap 渲染页面 #}
48

49 65

66 {% endblock %}

5. 用例集合

预期效果如下：

1）models.py 中新增 case_suite 模型类：

2）数据迁移

在项目目录下，执行以下两个命令进行数据迁移（将模型类转换成数据库表）：

python manage.py makemigrations # 生成迁移文件（模型类的信息）
python manage.py migrate # 执行开始迁移（将模型类信息转换成数据库表）

1）注册模型类到 admin.py：

1 from django.contrib import admin
2 from .import models
3
4
5 class ProjectAdmin(admin.ModelAdmin):
6 list_display = ("id", "name", "proj_owner", "test_owner", "dev_owner", "desc", "create_time", "update_time")
7
8 admin.site.register(models.Project, ProjectAdmin)
9
10
11 class ModuleAdmin(admin.ModelAdmin):
12 list_display = ("id", "name", "belong_project", "test_owner", "desc", "create_time", "update_time")
13
14 admin.site.register(models.Module, ModuleAdmin)
15
16
17 class TestCaseAdmin(admin.ModelAdmin):
18 list_display = (
19 "id", "case_name", "belong_project", "belong_module", "request_data", "uri", "assert_key", "maintainer",
20 "extract_var", "request_method", "status", "created_time", "updated_time", "user")
21
22 admin.site.register(models.TestCase, TestCaseAdmin)
23
24
25 class CaseSuiteAdmin(admin.ModelAdmin):
26 list_display = ("id", "suite_desc", "creator", "create_time")
27
28 admin.site.register(models.CaseSuite, CaseSuiteAdmin)

2）登录 admin 系统，进入用例集合表，添加数据：

from django.urls import path, re_path
from . import views

新增 templates/case_suite.html 模板：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试集合{% endblock %}
4 {% block content %}
5

6 {% csrf_token %} 7 8

9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for case_suite in case_suites %} 24 25 26 27 28 29 30 31 32 33 {% endfor %} 34 35

id	测试集合名称	创建者	创建时间	查看/删除测试用例	添加测试用例	用例集合执行结果
{{ case_suite.id }}	{{ case_suite.suite_desc }}	{{ case_suite.creator }}	{{ case_suite.create_time\|date:"Y-n-d H:i" }}	查看/删除测试用例	添加测试用例	查看用例集合执行结果

38
39 {# 实现分页标签的代码 #}
40 {# 这里使用 bootstrap 渲染页面 #}
41

42 58

59 {% endblock %}

2）修改 base 模板：菜单栏新增“用例集合”。

1
2 3 {% load static %} 4 5 6 7 8 9 {% block title %}base{% endblock %} 10 11 12 13 14 15 16 17 21 {% block css %}{% endblock %} 22 23 24

26 27 37 38 55

57 58 {% block content %}{% endblock %} 59 60 61 62 63 64 65 66

6. 用例集合添加测试用例

预期效果如下：

1）在 models.py 中，增加模型类 SuiteCase，记录用例集合所关联的用例。

2）数据迁移

在项目目录下，执行以下两个命令进行数据迁移（将模型类转换成数据库表）：

python manage.py makemigrations # 生成迁移文件（模型类的信息）
python manage.py migrate # 执行开始迁移（将模型类信息转换成数据库表）

from django.urls import path, re_path
from . import views

1）新增添加测试用例页的模板文件 templates/add_case_in_suite.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}管理测试集合{% endblock %}
4 {% block content %}
5
6
42

43 {% csrf_token %} 44 45

46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 {% for test_case in test_cases %} 61 62 63 64 65 66 67 68 69 70 71 {% endfor %} 72 73

id	用例名称	所属项目	所属模块	编写人员	创建时间	更新时间	创建用例用户名
{{ test_case.id }}	{{ test_case.case_name }}	{{ test_case.belong_project.name }}	{{ test_case.belong_module.name }}	{{ test_case.maintainer }}	{{ test_case.created_time\|date:"Y-n-d H:i" }}	{{ test_case.updated_time\|date:"Y-n-d H:i" }}	{{ test_case.user.username }}

76 {# 实现分页标签的代码 #}
77 {# 这里使用 bootstrap 渲染页面 #}
78

79 93

94 {% endblock %}

2）修改用例集合模板文件 templates/case_suite.html：修改“添加测试用例”的链接地址。

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例集合{% endblock %}
4 {% block content %}
5

6 {% csrf_token %} 7 8

9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for case_suite in case_suites %} 24 25 26 27 28 29 30 31 32 33 {% endfor %} 34 35

id	测试集合名称	创建者	创建时间	查看/删除测试用例	添加测试用例	用例集合执行结果
{{ case_suite.id }}	{{ case_suite.suite_desc }}	{{ case_suite.creator }}	{{ case_suite.create_time\|date:"Y-n-d H:i" }}	查看/删除测试用例	添加测试用例	查看用例集合执行结果

38
39 {# 实现分页标签的代码 #}
40 {# 这里使用 bootstrap 渲染页面 #}
41

42 58

59 {% endblock %}

7. 用例集合查看/删除测试用例

预期效果如下：

from django.urls import path, re_path
from . import views

1）新建 templates/show_and_delete_case_in_suite.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}查看/删除测试用例{% endblock %}
4 {% block content %}
5
6
45
46

测试集合名称：{{case_suite.suite_desc}}

49 {% csrf_token %} 50 51

52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 {% for test_case in test_cases %} 69 70 71 72 73 74 75 76 77 78 79 80 {% endfor %} 81 82

全选	用例序号	用例名称	所属项目	所属模块	编写人员	创建时间	更新时间	创建用例用户名
	{{ test_case.test_case.id }}	{{ test_case.test_case.case_name }}	{{ test_case.test_case.belong_project.name }}	{{ test_case.test_case.belong_module.name }}	{{ test_case.test_case.maintainer }}	{{ test_case.test_case.created_time\|date:"Y-n-d H:i" }}	{{ test_case.test_case.updated_time\|date:"Y-n-d H:i" }}	{{ test_case.test_case.user.username }}

85
86 {# 实现分页标签的代码 #}
87 {# 这里使用 bootstrap 渲染页面 #}
88

92 {% if test_cases.has_previous %} 93 上一页 94 {% endif %} 95 96 97 第 {{ test_cases.number }} 页 / 共 {{ test_cases.paginator.num_pages }} 页 98 99 {% if test_cases.has_next %} 100 下一页 101 {% endif %} 102

104

105

106

107

108 {% endblock %}

2）修改 templates/case_suite.html：增加“查看/删除测试用例”链接

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例集合{% endblock %}
4 {% block content %}
5

6 {% csrf_token %} 7 8

9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for case_suite in case_suites %} 24 25 26 27 28 29 30 31 32 33 {% endfor %} 34 35

id	测试集合名称	创建者	创建时间	查看/删除测试用例	添加测试用例	用例集合执行结果
{{ case_suite.id }}	{{ case_suite.suite_desc }}	{{ case_suite.creator }}	{{ case_suite.create_time\|date:"Y-n-d H:i" }}	查看/删除测试用例	添加测试用例	查看用例集合执行结果

38
39 {# 实现分页标签的代码 #}
40 {# 这里使用 bootstrap 渲染页面 #}
41

42 58

59 {% endblock %}

8. 测试用例执行

预期效果如下：

用例执行逻辑如下：

前端提交用例 id 列表到后台，后台获取每一条用例的信息；
后台获取域名信息、用例 id 列表；
对用例的请求数据进行变量的参数化、函数化等预处理操作；
根据先后顺序进行接口请求，并对响应数据进行断言；
根据用例中的提取变量表达式，从断言成功的响应数据中提取关联变量值用于后续用例使用。

templates/test_case.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试用例{% endblock %}
4
5 {% block content %}
6
42
43

44 {% csrf_token %} 45 46 运行环境： 47 51

52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 {% for test_case in test_cases %} 69 70 71 72 73 74 75 76 77 78 79 80 {% endfor %} 81 82

全选	用例名称	所属项目	所属模块	接口地址	请求方式	请求数据	断言key	提取变量表达式
{{ test_case.id }}	{{ test_case.case_name }}	{{ test_case.belong_project.name }}	{{ test_case.belong_module.name }}	{{ test_case.uri }}	{{ test_case.request_method }}	{{ test_case.request_data }}	{{ test_case.assert_key }}	{{ test_case.extract_var }}

83 84

86 {# 实现分页标签的代码 #}
87 {# 这里使用 bootstrap 渲染页面 #}
88

89 105

106 {% endblock %}

models.py：

执行数据迁移：

python manage.py makemigrations
python manage.py migrate

admin.py：

登录 admin 系统，添加地址配置数据：

1）Redis 持久化递增唯一数

在本工程中，我们使用 Redis 来维护一个每次调用函数来就会递增的数值，供注册接口的注册用户名拼接使用，避免注册接口请求数据重复使用问题。

1.1）Redis 持久化配置

修改 redis.windows.conf：

appendonly yes # 每次更新操作后进行日志记录
appendfsync everysec # 每秒同步一次（默认值）

1.2）启动 Redis 服务端：

redis-server.exe redis.windows.conf

2）请求/响应数据处理

在应用目录下新建 utils 包，用于封装接口请求的相关函数。

data_process.py

该模块实现了对接口请求的所需工具函数，如获取递增唯一数（供注册用户名使用）、md5 加密（用于登录密码加密）、请求数据预处理、响应数据断言等功能。

get_unique_num_value()：用于获取每次递增的唯一数
- 该函数的目标是解决注册用户名重复的问题。
- 虽然可以在赋值注册用户名变量时，采用前缀字符串拼接随机数的方式，但是用随机数的方式仍然是有可能出现用户名重复的情况。因此，可以在单独的一个文件中维护一个数字，每次请求注册接口之前，先读取该文件中的数字，拼接用户名前缀字符串。读取完之后，再把这个数字进行加一的操作并保存，即每读取一次这个数字之后，就做一次修改，进而保证每次拼接的用户名都是唯一的，避免出现因为用户名重复导致用例执行失败的情况。
data_preprocess()：对请求数据进行预处理：参数化及函数化。
data_postprocess()：将响应数据需要关联的参数保存进全局变量，供后续接口使用。
assert_result()：对响应数据进行关键字断言。

1 import re
2 import hashlib
3 import os
4 import json
5 import traceback
6 import redis
7 from InterfaceAutoTest.settings import redis_port
8
9
10 # 连接redis
11 pool = redis.ConnectionPool(host='localhost', port=redis_port, decode_responses=True)
12 redis_obj = redis.Redis(connection_pool=pool)
13
14
15 # 初始化框架工程中的全局变量，存储在测试数据中的唯一值数据
16 # 框架工程中若要使用字典中的任意一个变量，则每次使用后，均需要将字典中的value值进行加1操作。
17 def get_unique_number_value(unique_number):
18 data = None
19 try:
20 redis_value = redis_obj.get(unique_number) # {"unique_number": 666}
21 if redis_value:
22 data = redis_value
23 print("全局唯一数当前生成的值是：%s" % data)
24 # 把redis中key为unique_number的值进行加一操作，以便下提取时保持唯一
25 redis_obj.set(unique_number, int(redis_value) + 1)
26 else:
27 data = 1000 # 初始化递增数值
28 redis_obj.set(unique_number, data)
29 except Exception as e:
30 print("获取全局唯一数变量值失败，请求的全局唯一数变量是%s,异常原因如下：%s" % (unique_number, traceback.format_exc()))
31 data = None
32 finally:
33 return data
34
35
36 def md5(s):
37 m5 = hashlib.md5()
38 m5.update(s.encode("utf-8"))
39 md5_value = m5.hexdigest()
40 return md5_value
41
42
43 # 请求数据预处理：参数化、函数化
44 # 将请求数据中包含的${变量名}的字符串部分，替换为唯一数或者全局变量字典中对应的全局变量
45 def data_preprocess(global_key, requestData):
46 try:
47 # 匹配注册用户名参数，即"${unique_num…}"的格式，并取出本次请求的随机数供后续接口的用户名参数使用
48 if re.search(r"\$\{unique_num\d+\}", requestData):
49 var_name = re.search(r"\$\{(unique_num\d+)\}", requestData).group(1) # 获取用户名参数
50 print("用户名变量:%s" % var_name)
51 var_value = get_unique_number_value(var_name)
52 print("用户名变量值: %s" % var_value)
53 requestData = re.sub(r"\$\{unique_num\d+\}", str(var_value), requestData)
54 var_name = var_name.split("_")[1]
55 print("关联的用户名变量: %s" % var_name)
56 # "xxxkey" : "{'var_name': var_value}"
57 global_var = json.loads(os.environ[global_key])
58 global_var[var_name] = var_value
59 os.environ[global_key] = json.dumps(global_var)
60 print("用户名唯一数参数化后的全局变量【os.environ[global_key]】: {}".format(os.environ[global_key]))
61 # 函数化，如密码加密"${md5(…)}"的格式
62 if re.search(r"\$\{\w+$.+$\}", requestData):
63 var_pass = re.search(r"\$\{(\w+$.+$)\}", requestData).group(1) # 获取密码参数
64 print("需要函数化的变量: %s" % var_pass)
65 print("函数化后的结果: %s" % eval(var_pass))
66 requestData = re.sub(r"\$\{\w+$.+$\}", eval(var_pass), requestData) # 将requestBody里面的参数内容通过eval修改为实际变量值
67 print("函数化后的请求数据: %s" % requestData) # requestBody是拿到的请求时发送的数据
68 # 其余变量参数化
69 if re.search(r"\$\{(\w+)\}", requestData):
70 print("需要参数化的变量: %s" % (re.findall(r"\$\{(\w+)\}", requestData)))
71 for var_name in re.findall(r"\$\{(\w+)\}", requestData):
72 requestData = re.sub(r"\$\{%s\}" % var_name, str(json.loads(os.environ[global_key])[var_name]), requestData)
73 print("变量参数化后的最终请求数据: %s" % requestData)
74 print("数据参数后的最终全局变量【os.environ[global_key]】: {}".format(os.environ[global_key]))
75 return 0, requestData, ""
76 except Exception as e:
77 print("请求数据预处理发生异常，error：{}".format(traceback.format_exc()))
78 return 1, {}, traceback.format_exc()
79
80
81 # 响应数据提取关联参数
82 def data_postprocess(global_key, response_data, extract_var):
83 print("需提取的关联变量：%s" % extract_var)
84 var_name = extract_var.split("||")[0]
85 print("关联变量名：%s" % var_name)
86 regx_exp = extract_var.split("||")[1]
87 print("关联变量正则：%s" % regx_exp)
88 if re.search(regx_exp, response_data):
89 global_vars = json.loads(os.environ[global_key])
90 print("关联前的全局变量：{}".format(global_vars))
91 global_vars[var_name] = re.search(regx_exp, response_data).group(1)
92 os.environ[global_key] = json.dumps(global_vars)
93 print("关联前的全局变量：{}".format(os.environ[global_key]))
94 return
95
96
97 # 响应数据断言处理
98 def assert_result(response_obj, key_word):
99 try:
100 # 多个断言关键字
101 if '&&' in key_word:
102 key_word_list = key_word.split('&&')
103 print("断言关键字列表：%s" % key_word_list)
104 # 断言结果标识符
105 flag = True
106 exception_info = ''
107 # 遍历分隔出来的断言关键词列表
108 for key_word in key_word_list:
109 # 如果断言词非空，则进行断言
110 if key_word:
111 # 没查到断言词则认为是断言失败
112 if not (key_word in json.dumps(response_obj.json(), ensure_ascii=False)):
113 print("断言关键字【{}】匹配失败".format(key_word))
114 flag = False # 只要有一个断言词匹配失败，则整个接口断言失败
115 exception_info = "keyword: {} not matched from response, assert failed".format(key_word)
116 else:
117 print("断言关键字【{}】匹配成功".format(key_word))
118 if flag:
119 print("接口断言成功！")
120 else:
121 print("接口断言失败！")
122 return flag, exception_info
123 # 单个断言关键字
124 else:
125 if key_word in json.dumps(response_obj.json(), ensure_ascii=False):
126 print("接口断言【{}】匹配成功！".format(key_word))
127 return True, ''
128 else:
129 print("接口断言【{}】匹配失败！".format(key_word))
130 return False, ''
131 except Exception as e:
132 return False, traceback.format_exc()
133
134
135 # 测试代码
136 if __name__ == "__main__":
137 print(get_unique_number_value("unique_num1"))

request_process.py

该模块实现了对接口请求的封装。

1 import requests
2 import json
3 # from Util.Log import logger
4
5
6 # 此函数封装了get请求、post和put请求的方法
7 def request_process(url, request_method, request_content):
8 print("-------- 开始调用接口 --------")
9 if request_method == "get":
10 try:
11 if isinstance(request_content, dict):
12 print("接口地址：%s" % url)
13 print("请求数据：%s" % request_content)
14 r = requests.get(url, params=json.dumps(request_content))
15 else:
16 r = requests.get(url+str(request_content))
17 print("接口地址：%s" % r.url)
18 print("请求数据：%s" % request_content)
19
20 except Exception as e:
21 print("get方法请求发生异常：请求的url是%s, 请求的内容是%s\n发生的异常信息如下：%s" % (url, request_content, e))
22 r = None
23 return r
24 elif request_method == "post":
25 try:
26 if isinstance(request_content, dict):
27 print("接口地址：%s" % url)
28 print("请求数据：%s" % json.dumps(request_content))
29 r = requests.post(url, data=json.dumps(request_content))
30 else:
31 raise ValueError
32 except ValueError as e:
33 print("post方法请求发生异常：请求的url是%s, 请求的内容是%s\n发生的异常信息如下：%s" % (url, request_content, "请求参数不是字典类型"))
34 r = None
35 except Exception as e:
36 print("post方法请求发生异常：请求的url是%s, 请求的内容是%s\n发生的异常信息如下：%s" % (url, request_content, e))
37 r = None
38 return r
39 elif request_method == "put":
40 try:
41 if isinstance(request_content, dict):
42 print("接口地址：%s" % url)
43 print("请求数据：%s" % json.dumps(request_content))
44 r = requests.put(url, data=json.dumps(request_content))
45 else:
46 raise ValueError
47 except ValueError as e:
48 print("put方法请求发生异常：请求的url是%s, 请求的内容是%s\n发生的异常信息如下：%s" % (url, request_content, "请求参数不是字典类型"))
49 r = None
50 except Exception as e:
51 print("put方法请求发生异常：请求的url是%s, 请求的内容是%s\n发生的异常信息如下：%s" % (url, request_content, e))
52 r = None
53 return r

3）封装接口用例执行方法

在应用目录下新建 task.py：

4）修改测试用例视图函数

9. 用例执行结果展示

1）models.py 中增加 TestCaseExecuteResult 模型类，用于记录用例执行结果。

1 from django.db import models
2 from smart_selects.db_fields import GroupedForeignKey # pip install django-smart-selects：后台级联选择
3 from django.contrib.auth.models import User
4
5
6 class Project(models.Model):
7 id = models.AutoField(primary_key=True)
8 name = models.CharField('项目名称', max_length=50, unique=True, null=False)
9 proj_owner = models.CharField('项目负责人', max_length=20, null=False)
10 test_owner = models.CharField('测试负责人', max_length=20, null=False)
11 dev_owner = models.CharField('开发负责人', max_length=20, null=False)
12 desc = models.CharField('项目描述', max_length=100, null=True)
13 create_time = models.DateTimeField('项目创建时间', auto_now_add=True)
14 update_time = models.DateTimeField('项目更新时间', auto_now=True, null=True)
15
16 def __str__(self):
17 return self.name
18
19 class Meta:
20 verbose_name = '项目信息表'
21 verbose_name_plural = '项目信息表'
22
23
24 class Module(models.Model):
25 id = models.AutoField(primary_key=True)
26 name = models.CharField('模块名称', max_length=50, null=False)
27 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE)
28 test_owner = models.CharField('测试负责人', max_length=50, null=False)
29 desc = models.CharField('简要描述', max_length=100, null=True)
30 create_time = models.DateTimeField('创建时间', auto_now_add=True)
31 update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
32
33 def __str__(self):
34 return self.name
35
36 class Meta:
37 verbose_name = '模块信息表'
38 verbose_name_plural = '模块信息表'
39
40
41 class TestCase(models.Model):
42 id = models.AutoField(primary_key=True)
43 case_name = models.CharField('用例名称', max_length=50, null=False) # 如 register
44 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE, verbose_name='所属项目')
45 belong_module = GroupedForeignKey(Module, "belong_project", on_delete=models.CASCADE, verbose_name='所属模块')
46 request_data = models.CharField('请求数据', max_length=1024, null=False, default='')
47 uri = models.CharField('接口地址', max_length=1024, null=False, default='')
48 assert_key = models.CharField('断言内容', max_length=1024, null=True)
49 maintainer = models.CharField('编写人员', max_length=1024, null=False, default='')
50 extract_var = models.CharField('提取变量表达式', max_length=1024, null=True) # 示例：userid||userid": (\d+)
51 request_method = models.CharField('请求方式', max_length=1024, null=True)
52 status = models.IntegerField(null=True, help_text="0：表示有效，1：表示无效，用于软删除")
53 created_time = models.DateTimeField('创建时间', auto_now_add=True)
54 updated_time = models.DateTimeField('更新时间', auto_now=True, null=True)
55 user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name='责任人', null=True)
56
57 def __str__(self):
58 return self.case_name
59
60 class Meta:
61 verbose_name = '测试用例表'
62 verbose_name_plural = '测试用例表'
63
64
65 class CaseSuite(models.Model):
66 id = models.AutoField(primary_key=True)
67 suite_desc = models.CharField('用例集合描述', max_length=100, blank=True, null=True)
68 if_execute = models.IntegerField(verbose_name='是否执行', null=False, default=0, help_text='0：执行；1：不执行')
69 test_case_model = models.CharField('测试执行模式', max_length=100, blank=True, null=True, help_text='data/keyword')
70 creator = models.CharField(max_length=50, blank=True, null=True)
71 create_time = models.DateTimeField('创建时间', auto_now=True) # 创建时间-自动获取当前时间
72
73 class Meta:
74 verbose_name = "用例集合表"
75 verbose_name_plural = '用例集合表'
76
77
78 class SuiteCase(models.Model):
79 id = models.AutoField(primary_key=True)
80 case_suite = models.ForeignKey(CaseSuite, on_delete=models.CASCADE, verbose_name='用例集合')
81 test_case = models.ForeignKey(TestCase, on_delete=models.CASCADE, verbose_name='测试用例')
82 status = models.IntegerField(verbose_name='是否有效', null=False, default=1, help_text='0：有效，1：无效')
83 create_time = models.DateTimeField('创建时间', auto_now=True) # 创建时间-自动获取当前时间
84
85
86 class InterfaceServer(models.Model):
87 id = models.AutoField(primary_key=True)
88 env = models.CharField('环境', max_length=50, null=False, default='')
89 ip = models.CharField('ip', max_length=50, null=False, default='')
90 port = models.CharField('端口', max_length=100, null=False, default='')
91 remark = models.CharField('备注', max_length=100, null=True)
92 create_time = models.DateTimeField('创建时间', auto_now_add=True)
93 update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
94
95 def __str__(self):
96 return self.env
97
98 class Meta:
99 verbose_name = '接口地址配置表'
100 verbose_name_plural = '接口地址配置表'
101
102
103 class TestCaseExecuteResult(models.Model):
104 id = models.AutoField(primary_key=True)
105 belong_test_case = GroupedForeignKey(TestCase, "belong_test_case", on_delete=models.CASCADE, verbose_name='所属用例')
106 status = models.IntegerField(null=True, help_text="0：表示未执行，1：表示已执行")
107 exception_info = models.CharField(max_length=2048, blank=True, null=True)
108 request_data = models.CharField('请求体', max_length=1024, null=True) # {"code": "00", "userid": 22889}
109 response_data = models.CharField('响应字符串', max_length=1024, null=True) # {"code": "00", "userid": 22889}
110 execute_result = models.CharField('执行结果', max_length=1024, null=True) # 成功/失败
111 extract_var = models.CharField('关联参数', max_length=1024, null=True) # 响应成功后提取变量
112 last_time_response_data = models.CharField('上一次响应字符串', max_length=1024, null=True) # {"code": "00", "userid": 22889}
113 execute_total_time = models.CharField('执行耗时', max_length=1024, null=True)
114 execute_start_time = models.CharField('执行开始时间', max_length=300, blank=True, null=True)
115 execute_end_time = models.CharField('执行结束时间', max_length=300, blank=True, null=True)
116 created_time = models.DateTimeField('创建时间', auto_now_add=True)
117 updated_time = models.DateTimeField('更新时间', auto_now=True, null=True)
118
119 def __str__(self):
120 return str(self.id)
121
122 class Meta:
123 verbose_name = '用例执行结果记录表'
124 verbose_name_plural = '用例执行结果记录表'

2）数据迁移

python manage.py makemigrations
python manage.py migrate

修改应用目录下 task.py：

前端执行测试用例，查看用例执行结果表数据：

在前面已经获取到用例结果数据并保存，下面处理一下用例结果展示。

from django.urls import path, re_path
from . import views

1）新增”测试执行记录“模板文件：templates/test_case_execute_records.html

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例执行记录{% endblock %}
4 {% block content %}
5
6

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 {% for testrecord in test_case_execute_records %} 26 27 28 29 30 31 32 33 34 {% if testrecord.exception_info %} 35 36 {% else %} 37 38 {% endif %} 39 40 41 42 43 44 {% endfor %} 45 46 47

id	名称	请求数据	执行返回结果	操作	断言内容	执行结果	异常信息	请求后提取变量	开始时间	执行耗时(ms)
{{ testrecord.id }}	{{ testrecord.belong_test_case.case_name }}	{{ testrecord.request_data }}	{{ testrecord.response_data }}	对比差异	{{ testrecord.belong_test_case.assert_key }}	{{ testrecord.execute_result\|default_if_none:"" }}	显示异常信息	无	{{ testrecord.extract_var }}	{{ testrecord.execute_start_time }}	{{ testrecord.execute_total_time }}

48 49 {# 实现分页标签的代码 #} 50 {# 这里使用 bootstrap 渲染页面 #} 51

55 {% if test_case_execute_records.has_previous %} 56 上一页 57 {% endif %} 58 59 60 第 {{ test_case_execute_records.number }} 页 / 共 {{ test_case_execute_records.paginator.num_pages }} 页 61 62 {% if test_case_execute_records.has_next %} 63 下一页 64 {% endif %} 65

70 {% endblock %}

2）修改 base.html：新增“用例执行结果”菜单项

1
2 3 {% load static %} 4 5 6 7 8 9 {% block title %}base{% endblock %} 10 11 12 13 14 15 16 17 21 {% block css %}{% endblock %} 22 23 24

26 27 37 38

39 46 55

58 59 {% block content %}{% endblock %} 60 61 62 63 64 65 66 67

页面效果如下：

9.6 结果对比差异

在用例执行结果页面，可以看到在“操作”列，有“对比差异”链接，该功能用于对比当前用例上一次的执行结果与当前的执行结果，便于查看结果的差异。

由于在前面用例执行时，已经在结果记录环节获取到当前用例上一次的结果并记录到当前用例记录数据中，下面来处理一下这个页面的展示。

1）定义路由

from django.urls import path, re_path
from . import views

2）定义视图函数

3）定义模板

新增 case_result_diff.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}结果对比差异{% endblock %}
4
5 {% block content %}
6

7 8 9 10 11 12 13 14 15 20 23 24 25

上次执行结果	本次执行结果
16 17 {{ last_time_execute_response \| safe }} 18 19	21 {{ present_response \| safe }} 22

26
27 {% endblock %}

修改 test_case_execute_records.html：增加“对比差异”链接

{% extends 'base.html' %}
{% load static %}
{% block title %}用例执行记录{% endblock %}
{% block content %}

{% for testrecord in test\_case\_execute\_records %} {% if testrecord.exception\_info %} {% else %} {% endif %} {% endfor %}

id	名称	请求数据	执行返回结果	操作	断言内容	执行结果	异常信息	请求后提取变量	开始时间	执行耗时(ms)
{{ testrecord.id }}	{{ testrecord.belong\_test\_case.case\_name }}	{{ testrecord.request\_data }}	{{ testrecord.response\_data }}	对比差异	{{ testrecord.belong\_test\_case.assert\_key }}	{{ testrecord.execute\_result\|default\_if\_none:"" }}	显示异常信息	无	{{ testrecord.extract\_var }}	{{ testrecord.execute\_start\_time }}	{{ testrecord.execute\_total\_time }}

{# 实现分页标签的代码 #} {# 这里使用 bootstrap 渲染页面 #}

{% if test\_case\_execute\_records.has\_previous %} 上一页 {% endif %} 第 {{ test\_case\_execute\_records.number }} 页 / 共 {{ test\_case\_execute\_records.paginator.num\_pages }} 页 {% if test\_case\_execute\_records.has\_next %} 下一页 {% endif %}

{% endblock %}

1）定义路由

from django.urls import path, re_path
from . import views

2）定义视图函数

3）定义模板

新增异常信息展示模板：show_exception.html

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}异常信息{% endblock %}
4 {% block content %}
5
6

异常信息如下：

8
9 {% endblock %}

修改用例执行记录模板 test_case_execute_records.html：增加异常信息展示链接

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例执行记录{% endblock %}
4 {% block content %}
5
6

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 {% for testrecord in test_case_execute_records %} 26 27 28 29 30 31 32 33 34 {% ifequal testrecord.execute_result '成功' %} 35 36 {% else %} 37 38 {% endifequal %} 39 40 {% if testrecord.exception_info %} 41 42 {% else %} 43 44 {% endif %} 45 46 47 48 49 50 {% endfor %} 51 52 53

id	名称	请求数据	执行返回结果	操作	断言内容	执行结果	异常信息	请求后提取变量	开始时间	执行耗时(ms)
{{ testrecord.id }}	{{ testrecord.belong_test_case.case_name }}	{{ testrecord.request_data }}	{{ testrecord.response_data }}	对比差异	{{ testrecord.belong_test_case.assert_key }}	{{ testrecord.execute_result}}	{{ testrecord.execute_result}}	显示异常信息	无	{{ testrecord.extract_var }}	{{ testrecord.execute_start_time }}	{{ testrecord.execute_total_time }}

54 55 {# 实现分页标签的代码 #} 56 {# 这里使用 bootstrap 渲染页面 #} 57

61 {% if test_case_execute_records.has_previous %} 62 上一页 63 {% endif %} 64 65 66 第 {{ test_case_execute_records.number }} 页 / 共 {{ test_case_execute_records.paginator.num_pages }} 页 67 68 {% if test_case_execute_records.has_next %} 69 下一页 70 {% endif %} 71

76 {% endblock %}

10. 测试集合执行

在前面的测试用例执行步骤，我们已经把请求数据处理、接口请求、断言、变量提取等环节做了处理。那么测试集合的执行也可以复用前面的逻辑，前端在提交测试集合到后端时，后端获取到集合中包含的用例，调用用例执行的方法即可，然后记录测试集合执行相关的结果记录。

models.py：

1 from django.db import models
2 from smart_selects.db_fields import GroupedForeignKey # pip install django-smart-selects：后台级联选择
3 from django.contrib.auth.models import User
4
5
6 # 项目
7 class Project(models.Model):
8 id = models.AutoField(primary_key=True)
9 name = models.CharField('项目名称', max_length=50, unique=True, null=False)
10 proj_owner = models.CharField('项目负责人', max_length=20, null=False)
11 test_owner = models.CharField('测试负责人', max_length=20, null=False)
12 dev_owner = models.CharField('开发负责人', max_length=20, null=False)
13 desc = models.CharField('项目描述', max_length=100, null=True)
14 create_time = models.DateTimeField('项目创建时间', auto_now_add=True)
15 update_time = models.DateTimeField('项目更新时间', auto_now=True, null=True)
16
17 def __str__(self):
18 return self.name
19
20 class Meta:
21 verbose_name = '项目信息表'
22 verbose_name_plural = '项目信息表'
23
24
25 # 模块
26 class Module(models.Model):
27 id = models.AutoField(primary_key=True)
28 name = models.CharField('模块名称', max_length=50, null=False)
29 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE)
30 test_owner = models.CharField('测试负责人', max_length=50, null=False)
31 desc = models.CharField('简要描述', max_length=100, null=True)
32 create_time = models.DateTimeField('创建时间', auto_now_add=True)
33 update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
34
35 def __str__(self):
36 return self.name
37
38 class Meta:
39 verbose_name = '模块信息表'
40 verbose_name_plural = '模块信息表'
41
42
43 # 测试用例
44 class TestCase(models.Model):
45 id = models.AutoField(primary_key=True)
46 case_name = models.CharField('用例名称', max_length=50, null=False) # 如 register
47 belong_project = models.ForeignKey(Project, on_delete=models.CASCADE, verbose_name='所属项目')
48 belong_module = GroupedForeignKey(Module, "belong_project", on_delete=models.CASCADE, verbose_name='所属模块')
49 request_data = models.CharField('请求数据', max_length=1024, null=False, default='')
50 uri = models.CharField('接口地址', max_length=1024, null=False, default='')
51 assert_key = models.CharField('断言内容', max_length=1024, null=True)
52 maintainer = models.CharField('编写人员', max_length=1024, null=False, default='')
53 extract_var = models.CharField('提取变量表达式', max_length=1024, null=True) # 示例：userid||userid": (\d+)
54 request_method = models.CharField('请求方式', max_length=1024, null=True)
55 status = models.IntegerField(null=True, help_text="0：表示有效，1：表示无效，用于软删除")
56 created_time = models.DateTimeField('创建时间', auto_now_add=True)
57 updated_time = models.DateTimeField('更新时间', auto_now=True, null=True)
58 user = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name='责任人', null=True)
59
60 def __str__(self):
61 return self.case_name
62
63 class Meta:
64 verbose_name = '测试用例表'
65 verbose_name_plural = '测试用例表'
66
67
68 # 用例集合
69 class CaseSuite(models.Model):
70 id = models.AutoField(primary_key=True)
71 suite_desc = models.CharField('用例集合描述', max_length=100, blank=True, null=True)
72 if_execute = models.IntegerField(verbose_name='是否执行', null=False, default=0, help_text='0：执行；1：不执行')
73 test_case_model = models.CharField('测试执行模式', max_length=100, blank=True, null=True, help_text='data/keyword')
74 creator = models.CharField(max_length=50, blank=True, null=True)
75 create_time = models.DateTimeField('创建时间', auto_now=True) # 创建时间-自动获取当前时间
76
77 class Meta:
78 verbose_name = "用例集合表"
79 verbose_name_plural = '用例集合表'
80
81
82 # 用例集合关联用例
83 class SuiteCase(models.Model):
84 id = models.AutoField(primary_key=True)
85 case_suite = models.ForeignKey(CaseSuite, on_delete=models.CASCADE, verbose_name='用例集合')
86 test_case = models.ForeignKey(TestCase, on_delete=models.CASCADE, verbose_name='测试用例')
87 status = models.IntegerField(verbose_name='是否有效', null=False, default=1, help_text='0：有效，1：无效')
88 create_time = models.DateTimeField('创建时间', auto_now=True) # 创建时间-自动获取当前时间
89
90
91 # 接口服务器配置
92 class InterfaceServer(models.Model):
93 id = models.AutoField(primary_key=True)
94 env = models.CharField('环境', max_length=50, null=False, default='')
95 ip = models.CharField('ip', max_length=50, null=False, default='')
96 port = models.CharField('端口', max_length=100, null=False, default='')
97 remark = models.CharField('备注', max_length=100, null=True)
98 create_time = models.DateTimeField('创建时间', auto_now_add=True)
99 update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
100
101 def __str__(self):
102 return self.env
103
104 class Meta:
105 verbose_name = '接口地址配置表'
106 verbose_name_plural = '接口地址配置表'
107
108
109 # 测试用例执行记录
110 class TestCaseExecuteResult(models.Model):
111 id = models.AutoField(primary_key=True)
112 belong_test_case = GroupedForeignKey(TestCase, "belong_test_case", on_delete=models.CASCADE, verbose_name='所属用例')
113 status = models.IntegerField(null=True, help_text="0：表示未执行，1：表示已执行")
114 exception_info = models.CharField(max_length=2048, blank=True, null=True)
115 request_data = models.CharField('请求体', max_length=1024, null=True) # {"code": "00", "userid": 22889}
116 response_data = models.CharField('响应字符串', max_length=1024, null=True) # {"code": "00", "userid": 22889}
117 execute_result = models.CharField('执行结果', max_length=1024, null=True) # 成功/失败
118 extract_var = models.CharField('关联参数', max_length=1024, null=True) # 响应成功后提取变量
119 last_time_response_data = models.CharField('上一次响应字符串', max_length=1024, null=True) # {"code": "00", "userid": 22889}
120 execute_total_time = models.CharField('执行耗时', max_length=1024, null=True)
121 execute_start_time = models.CharField('执行开始时间', max_length=300, blank=True, null=True)
122 execute_end_time = models.CharField('执行结束时间', max_length=300, blank=True, null=True)
123 created_time = models.DateTimeField('创建时间', auto_now_add=True)
124 updated_time = models.DateTimeField('更新时间', auto_now=True, null=True)
125
126 def __str__(self):
127 return str(self.id)
128
129 class Meta:
130 verbose_name = '用例执行结果记录表'
131 verbose_name_plural = '用例执行结果记录表'
132
133
134 # 用例集合的执行记录
135 class CaseSuiteExecuteRecord(models.Model):
136 id = models.AutoField(primary_key=True)
137 case_suite = models.ForeignKey(CaseSuite, on_delete=models.CASCADE, verbose_name='测试集合')
138 run_time_interval = models.IntegerField(verbose_name='延迟时间', null=True, default=0)
139 status = models.IntegerField(verbose_name='执行状态', null=True, default=0)
140 test_result = models.CharField(max_length=50, blank=True, null=True)
141 creator = models.CharField(max_length=50, blank=True, null=True)
142 create_time = models.DateTimeField('创建时间', auto_now=True) # 创建时间-自动获取当前时间
143 execute_start_time = models.CharField('执行开始时间', max_length=300, blank=True, null=True)
144
145
146 # 用例集合下的用例执行记录
147 class CaseSuiteTestCaseExecuteRecord(models.Model):
148 id = models.AutoField(primary_key=True)
149 case_suite_record = models.ForeignKey(CaseSuiteExecuteRecord, on_delete=models.CASCADE, verbose_name='测试集合执行记录')
150 test_case = models.ForeignKey(TestCase, on_delete=models.CASCADE, verbose_name='测试用例')
151 status = models.IntegerField(verbose_name='执行状态', null=True, default=0)
152 exception_info = models.CharField(max_length=2048, blank=True, null=True)
153 request_data = models.CharField('请求体', max_length=1024, null=True) # {"code": "00", "userid": 22889}
154 response_data = models.CharField('响应字符串', max_length=1024, null=True) # {"code": "00", "userid": 22889}
155 execute_result = models.CharField('执行结果', max_length=1024, null=True) # 成功/失败
156 extract_var = models.CharField('关联参数', max_length=1024, null=True) # 响应成功后提取变量
157 last_time_response_data = models.CharField('上一次响应字符串', max_length=1024,
158 null=True) # {"code": "00", "userid": 22889}
159 execute_total_time = models.CharField('执行耗时', max_length=1024, null=True)
160 execute_start_time = models.CharField('执行开始时间', max_length=300, blank=True, null=True)
161 execute_end_time = models.CharField('执行结束时间', max_length=300, blank=True, null=True)

执行数据迁移：

python manage.py makemigrations
python manage.py migrate

1）修改 case_suite.html 模板，增加提交测试集合相关组件。

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例集合{% endblock %}
4 {% block content %}
5
40
41

42 {% csrf_token %} 43 延迟执行的时间（单位：秒）: 44 45 运行环境： 46 50 51 52

53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 {% for case_suite in case_suites %} 69 70 71 72 73 74 75 76 77 78 79 {% endfor %} 80 81

全选	id	测试集合名称	创建者	创建时间	查看/删除测试用例	添加测试用例	用例集合执行结果
	{{ case_suite.id }}	{{ case_suite.suite_desc }}	{{ case_suite.creator }}	{{ case_suite.create_time\|date:"Y-n-d H:i" }}	查看/删除测试用例	添加测试用例	查看用例集合执行结果

84
85 {# 实现分页标签的代码 #}
86 {# 这里使用 bootstrap 渲染页面 #}
87

88 104

105 {% endblock %}

1）在应用 task.py 中增加用例集合执行的任务函数：

1 import time
2 import os
3 import traceback
4 import json
5 from . import models
6 from .utils.data_process import data_preprocess, assert_result, data_postprocess
7 from .utils.request_process import request_process
8
9
10 # 测试用例执行
11 def case_task(test_case_id_list, server_address):
12 global_key = 'case'+ str(int(time.time() * 100000))
13 os.environ[global_key] = '{}'
14 print()
15 print("全局变量标识符【global_key】: {}".format(global_key))
16 print("全局变量内容【os.environ[global_key]】: {}".format(os.environ[global_key]))
17 for test_case_id in test_case_id_list:
18
19 test_case = models.TestCase.objects.filter(id=int(test_case_id))[0]
20 last_execute_record_data = models.TestCaseExecuteResult.objects.filter(
21 belong_test_case_id=test_case_id).order_by('-id')
22 if last_execute_record_data:
23 last_time_execute_response_data = last_execute_record_data[0].response_data
24 else:
25 last_time_execute_response_data = ''
26 print("上一次响应结果: {}".format(last_execute_record_data))
27 print("上一次响应时间: {}".format(last_time_execute_response_data))
28 execute_record = models.TestCaseExecuteResult.objects.create(belong_test_case=test_case)
29 execute_record.last_time_response_data = last_time_execute_response_data
30 # 获取当前用例上一次执行结果
31 execute_record.save()
32
33 test_case = models.TestCase.objects.filter(id=int(test_case_id))[0]
34 print("\n######### 开始执行用例【{}】 #########".format(test_case))
35 execute_start_time = time.time() # 记录时间戳，便于计算总耗时（毫秒）
36 execute_record.execute_start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_start_time))
37
38 request_data = test_case.request_data
39 extract_var = test_case.extract_var
40 assert_key = test_case.assert_key
41 interface_name = test_case.uri
42 belong_project = test_case.belong_project
43 belong_module = test_case.belong_module
44 maintainer = test_case.maintainer
45 request_method = test_case.request_method
46 print("初始请求数据: {}".format(request_data))
47 print("关联参数: {}".format(extract_var))
48 print("断言关键字: {}".format(assert_key))
49 print("接口名称: {}".format(interface_name))
50 print("所属项目: {}".format(belong_project))
51 print("所属模块: {}".format(belong_module))
52 print("用例维护人: {}".format(maintainer))
53 print("请求方法: {}".format(request_method))
54 url = "{}{}".format(server_address, interface_name)
55 print("接口地址: {}".format(url))
56 code, request_data, error_msg = data_preprocess(global_key, str(request_data))
57 # 请求数据预处理异常，结束用例执行
58 if code != 0:
59 print("数据处理异常，error: {}".format(error_msg))
60 execute_record.execute_result = "失败"
61 execute_record.status = 1
62 execute_record.exception_info = error_msg
63 execute_end_time = time.time()
64 execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_end_time))
65 execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
66 execute_record.save()
67 return
68 # 记录请求预处理结果
69 else:
70 execute_record.request_data = request_data
71 # 调用接口
72 try:
73 res_data = request_process(url, request_method, json.loads(request_data))
74 print("响应数据: {}".format(json.dumps(res_data.json(), ensure_ascii=False))) # ensure_ascii：兼容中文
75 result_flag, exception_info = assert_result(res_data, assert_key)
76 # 结果记录保存
77 if result_flag:
78 print("用例【%s】执行成功！" % test_case)
79 execute_record.execute_result = "成功"
80 if extract_var.strip() != "None":
81 var_value = data_postprocess(global_key, json.dumps(res_data.json(), ensure_ascii=False), extract_var)
82 execute_record.extract_var = var_value
83 else:
84 print("用例【%s】执行失败！" % test_case)
85 execute_record.execute_result = "失败"
86 execute_record.exception_info = exception_info
87 execute_record.response_data = json.dumps(res_data.json(), ensure_ascii=False)
88 execute_record.status = 1
89 execute_end_time = time.time()
90 execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_end_time))
91 print("执行结果结束时间: {}".format(execute_record.execute_end_time))
92 execute_record.execute_total_time = int((execute_end_time - execute_start_time) * 1000)
93 print("用例执行耗时: {}".format(execute_record.execute_total_time))
94 execute_record.save()
95 except Exception as e:
96 print("接口请求异常，error: {}".format(traceback.format_exc()))
97 execute_record.execute_result = "失败"
98 execute_record.exception_info = traceback.format_exc()
99 execute_record.status = 1
100 execute_end_time = time.time()
101 execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_end_time))
102 print("执行结果结束时间: {}".format(execute_record.execute_end_time))
103 execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
104 print("用例执行耗时: {} 毫秒".format(execute_record.execute_total_time))
105 execute_record.save()
106
107
108 # 用例集合执行
109 def suite_task(case_suite_record, case_suite, server_address):
110 global_key = case_suite.suite_desc + str(int(time.time() * 100000))
111 # global_vars = {"{}".format(global_key): {}}
112 os.environ[global_key] = '{}'
113 print("global_key: {}".format(global_key))
114 print("os.environ[global_key]: {}".format(os.environ[global_key]))
115 case_suite_test_cases = models.SuiteCase.objects.filter(case_suite=case_suite).order_by('id')
116 print("用例集合的测试用例列表: {}".format(case_suite_test_cases))
117 case_suite_record.test_result = "成功"
118 case_suite_record.execute_start_time = time.strftime("%Y-%m-%d %H:%M:%S")
119
120 for case_suite_test_case in case_suite_test_cases:
121 test_case = case_suite_test_case.test_case
122 print("\n######### 开始执行用例【{}】 #########".format(test_case))
123 last_execute_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.filter(
124 test_case_id=test_case.id).order_by('-id')
125 if last_execute_record_data:
126 last_time_execute_response_data = last_execute_record_data[0].response_data
127 else:
128 last_time_execute_response_data = ''
129 print("上一次响应结果: {}".format(last_execute_record_data))
130 print("上一次响应时间: {}".format(last_time_execute_response_data))
131 suite_case_execute_record = models.CaseSuiteTestCaseExecuteRecord.objects.create(case_suite_record=case_suite_record,
132 test_case=test_case)
133 execute_start_time = time.time() # 记录时间戳，便于计算总耗时（毫秒）
134 suite_case_execute_record.execute_start_time = time.strftime("%Y-%m-%d %H:%M:%S",
135 time.localtime(execute_start_time))
136 print("用例集合开始执行时间: {}".format(suite_case_execute_record.execute_start_time))
137 suite_case_execute_record.last_time_response_data = last_time_execute_response_data
138 suite_case_execute_record.save()
139 request_data = test_case.request_data
140 extract_var = test_case.extract_var
141 assert_key = test_case.assert_key
142 interface_name = test_case.uri
143 belong_project = test_case.belong_project
144 belong_module = test_case.belong_module
145 maintainer = test_case.maintainer
146 request_method = test_case.request_method
147 print("初始请求数据: {}".format(request_data))
148 print("关联参数: {}".format(extract_var))
149 print("断言关键字: {}".format(assert_key))
150 print("接口名称: {}".format(interface_name))
151 print("所属项目: {}".format(belong_project))
152 print("所属模块: {}".format(belong_module))
153 print("用例维护人: {}".format(maintainer))
154 print("请求方法: {}".format(request_method))
155 url = "{}{}".format(server_address, interface_name)
156 print("接口地址: {}".format(url))
157 # 请求数据预处理
158 code, request_data, error_msg = data_preprocess(global_key, str(request_data))
159 # 请求数据预处理异常，结束用例执行
160 if code != 0:
161 print("数据处理异常，error: {}".format(error_msg))
162 suite_case_execute_record.execute_result = "失败"
163 suite_case_execute_record.status = 1
164 suite_case_execute_record.exception_info = error_msg
165 execute_end_time = time.time()
166 suite_case_execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S",
167 time.localtime(execute_end_time))
168 suite_case_execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
169 suite_case_execute_record.save()
170 case_suite_record.test_result = "失败"
171 # 记录请求预处理的结果
172 suite_case_execute_record.request_data = request_data
173 try:
174 # 调用接口
175 res_data = request_process(url, request_method, json.loads(request_data))
176 print("响应数据: {}".format(json.dumps(res_data.json(), ensure_ascii=False)))
177
178 result_flag, exception_info = assert_result(res_data, assert_key)
179 # 结果记录保存
180 if result_flag:
181 print("用例【%s】执行成功！" % test_case)
182 suite_case_execute_record.execute_result = "成功"
183 if extract_var.strip() != "None":
184 var_value = data_postprocess(global_key, json.dumps(res_data.json(), ensure_ascii=False),
185 extract_var)
186 suite_case_execute_record.extract_var = var_value
187 else:
188 print("用例【%s】执行失败！" % test_case)
189 suite_case_execute_record.execute_result = "失败"
190 suite_case_execute_record.exception_info = exception_info
191 case_suite_record.test_result = "失败"
192 suite_case_execute_record.response_data = json.dumps(res_data.json(), ensure_ascii=False)
193 suite_case_execute_record.status = 1
194 execute_end_time = time.time()
195 suite_case_execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S",
196 time.localtime(execute_end_time))
197 suite_case_execute_record.execute_total_time = int((execute_end_time - execute_start_time) * 1000)
198 print("用例执行耗时: {} 毫秒".format(
199 suite_case_execute_record.execute_total_time))
200 suite_case_execute_record.save()
201 except Exception as e:
202 print("接口请求异常，error: {}".format(e))
203 suite_case_execute_record.execute_result = "失败"
204 suite_case_execute_record.exception_info = traceback.format_exc()
205 suite_case_execute_record.status = 1
206 execute_end_time = time.time()
207 suite_case_execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S",
208 time.localtime(execute_end_time))
209 suite_case_execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
210 print("用例集合执行总耗时: {} 毫秒".format(suite_case_execute_record.execute_total_time))
211 suite_case_execute_record.save()
212 case_suite_record.test_result = "失败"
213
214 case_suite_record.status = 1 # 执行完毕
215 case_suite_record.save()

2）修改视图函数

前端页面勾选测试集合，提交运行，生成表数据如下所示：

11. 用例集合执行结果展示

from django.urls import path, re_path
from . import views

1）新增“用例集合执行结果”模板：case_suite_execute_record.html

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试集合执行结果{% endblock %}
4
5 {% block content %}
6
7

id	测试集合名称	延迟执行时间	执行状态	测试结果	创建者	创建时间
{{ case_suite_execute_record.id }}	{{ case_suite_execute_record.case_suite.suite_desc }}	{{ case_suite_execute_record.run_time_interval }}	执行完毕	待执行	{{ case_suite_execute_record.test_result\|default_if_none:"" }}	{{ case_suite_execute_record.creator }}	{{ case_suite_execute_record.create_time\|date:"Y-n-d H:i" }}

42
43 {# 实现分页标签的代码 #}
44 {# 这里使用 bootstrap 渲染页面 #}
45

49 {% if case_suite_execute_records.has_previous %} 50 上一页 51 {% endif %} 52 53 54 第 {{ case_suite_execute_records.number }} 页 / 共 {{ case_suite_execute_records.paginator.num_pages }} 页 55 56 {% if case_suite_execute_records.has_next %} 57 下一页 58 {% endif %} 59

63 {% endblock %}

此时，页面中展示了测试集合维度的执行结果，并没有展示其包含的用例的结果，下面处理一下测试集合包含用例的结果展示。

该页面的实现逻辑与用例执行结果模块类似。

1）定义路由

from django.urls import path, re_path
from . import views

urlpatterns = [
path('', views.index),
path('login/', views.login),
path('logout/', views.logout),
path('project/', views.project, name='project'),
path('module/', views.module, name='module'),
path('test_case/', views.test_case, name="test_case"),
re_path('test_case_detail/(?P[0-9]+)', views.test_case_detail, name="test_case_detail"),
re_path('module_test_cases/(?P[0-9]+)/$', views.module_test_cases, name="module_test_cases"),
path('case_suite/', views.case_suite, name="case_suite"),
re_path('add_case_in_suite/(?P[0-9]+)', views.add_case_in_suite, name="add_case_in_suite"),
re_path('show_and_delete_case_in_suite/(?P[0-9]+)', views.show_and_delete_case_in_suite, name="show_and_delete_case_in_suite"),
path('test_case_execute_record/', views.test_case_execute_record, name="test_case_execute_record"),
re_path('case_result_diff/(?P[0-9]+)', views.case_result_diff, name="case_result_diff"),
re_path('show_exception/(?P[0-9]+)$', views.show_exception, name="show_exception"),
path('case_suite_execute_record/', views.case_suite_execute_record, name="case_suite_execute_record"),
re_path('suite_case_execute_record/(?P[0-9]+)', views.suite_case_execute_record, name="suite_case_execute_record"),
re_path('suite_case_result_diff/(?P[0-9]+)', views.suite_case_result_diff, name="suite_case_result_diff"),
re_path('suite_case_exception/(?P[0-9]+)', views.suite_case_exception, name="suite_case_exception"),
]

2）定义视图

1 from django.shortcuts import render, redirect, HttpResponse
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage
5 from .form import UserForm
6 import traceback
7 import json
8 from . import models
9 from .task import case_task, suite_task
10
11
12 # 封装分页处理
13 def get_paginator(request, data):
14 paginator = Paginator(data, 10) # 默认每页展示10条数据
15 # 获取 url 后面的 page 参数的值, 首页不显示 page 参数, 默认值是 1
16 page = request.GET.get('page')
17 try:
18 paginator_pages = paginator.page(page)
19 except PageNotAnInteger:
20 # 如果请求的页数不是整数, 返回第一页。
21 paginator_pages = paginator.page(1)
22 except InvalidPage:
23 # 如果请求的页数不存在, 重定向页面
24 return HttpResponse('找不到页面的内容')
25 return paginator_pages
26
27
28 # 项目菜单项
29 @login_required
30 def project(request):
31 print("request.user.is_authenticated: ", request.user.is_authenticated)
32 projects = models.Project.objects.filter().order_by('-id')
33 print("projects:", projects)
34 return render(request, 'project.html', {'projects': get_paginator(request, projects)})
35
36
37 # 模块菜单项
38 @login_required
39 def module(request):
40 if request.method == "GET": # 请求get时候，id倒序查询所有的模块数据
41 modules = models.Module.objects.filter().order_by('-id')
42 return render(request, 'module.html', {'modules': get_paginator(request, modules)})
43 else: # 否则就是Post请求，会根据输入内容，使用模糊的方式查找所有的项目
44 proj_name = request.POST['proj_name']
45 projects = models.Project.objects.filter(name__contains=proj_name.strip())
46 projs = [proj.id for proj in projects]
47 modules = models.Module.objects.filter(belong_project__in=projs) # 把项目中所有的模块都找出来
48 return render(request, 'module.html', {'modules': get_paginator(request, modules), 'proj_name': proj_name})
49
50
51 # 获取测试用例执行的接口地址
52 def get_server_address(env):
53 if env: # 环境处理
54 env_data = models.InterfaceServer.objects.filter(env=env[0])
55 print("env_data: {}".format(env_data))
56 if env_data:
57 ip = env_data[0].ip
58 port = env_data[0].port
59 print("ip: {}, port: {}".format(ip, port))
60 server_address = "http://{}:{}".format(ip, port)
61 print("server_address: {}".format(server_address))
62 return server_address
63 else:
64 return ""
65 else:
66 return ""
67
68
69 # 测试用例菜单项
70 @login_required
71 def test_case(request):
72 print("request.session['is_login']: {}".format(request.session['is_login']))
73 test_cases = ""
74 if request.method == "GET":
75 test_cases = models.TestCase.objects.filter().order_by('id')
76 print("testcases: {}".format(test_cases))
77 elif request.method == "POST":
78 print("request.POST: {}".format(request.POST))
79 test_case_id_list = request.POST.getlist('test_cases_list')
80 env = request.POST.getlist('env')
81 print("env: {}".format(env))
82 server_address = get_server_address(env)
83 if not server_address:
84 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
85 if test_case_id_list:
86 test_case_id_list.sort()
87 print("test_case_id_list: {}".format(test_case_id_list))
88 print("获取到用例，开始用例执行")
89 case_task(test_case_id_list, server_address)
90 else:
91 print("运行测试用例失败")
92 return HttpResponse("提交的运行测试用例为空，请选择用例后在提交！")
93 test_cases = models.TestCase.objects.filter().order_by('id')
94 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
95
96
97 # 用例详情页
98 @login_required
99 def test_case_detail(request, test_case_id):
100 test_case_id = int(test_case_id)
101 test_case = models.TestCase.objects.get(id=test_case_id)
102 print("test_case: {}".format(test_case))
103 print("test_case.id: {}".format(test_case.id))
104 print("test_case.belong_project: {}".format(test_case.belong_project))
105
106 return render(request, 'test_case_detail.html', {'test_case': test_case})
107
108
109 # 模块页展示测试用例
110 @login_required
111 def module_test_cases(request, module_id):
112 module = ""
113 if module_id: # 访问的时候，会从url中提取模块的id，根据模块id查询到模块数据，在模板中展现
114 module = models.Module.objects.get(id=int(module_id))
115 test_cases = models.TestCase.objects.filter(belong_module=module).order_by('-id')
116 print("test_case in module_test_cases: {}".format(test_cases))
117 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
118
119
120 # 用例集合菜单项
121 @login_required
122 def case_suite(request):
123 if request.method == "POST":
124 count_down_time = 0
125 if request.POST['delay_time']:
126 print("输入的延迟时间是: {}".format(request.POST['delay_time']))
127 try:
128 count_down_time = int(request.POST['delay_time'])
129 except:
130 print("输入的延迟时间是非数字！")
131 else:
132 print("没有输入延迟时间")
133 env = request.POST.getlist('env')
134 print("env: {}".format(env))
135 server_address = get_server_address(env)
136 if not server_address:
137 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
138 case_suite_list = request.POST.getlist('case_suite_list')
139 if case_suite_list:
140 print("所需执行的用例集合列表：", case_suite_list)
141 for suite_id in case_suite_list:
142 test_suite = models.CaseSuite.objects.get(id=int(suite_id))
143 print("所需执行的用例集合: {}".format(test_suite))
144 username = request.user.username
145 test_suite_record = models.CaseSuiteExecuteRecord.objects.create(case_suite=test_suite,
146 run_time_interval=count_down_time,
147 creator=username)
148 suite_task(test_suite_record, test_suite, server_address)
149 else:
150 print("运行测试集合用例失败")
151 return HttpResponse("运行的测试集合为空，请选择测试集合后再运行！")
152 case_suites = models.CaseSuite.objects.filter()
153 return render(request, 'case_suite.html', {'case_suites': get_paginator(request, case_suites)})
154
155
156 # 用例集合-添加测试用例页
157 @login_required
158 def add_case_in_suite(request, suite_id):
159 # 查询指定的用例集合
160 case_suite = models.CaseSuite.objects.get(id=suite_id)
161 # 根据id号查询所有的用例
162 test_cases = models.TestCase.objects.filter().order_by('id')
163 if request.method == "GET":
164 print("test cases:", test_cases)
165 elif request.method == "POST":
166 test_cases_list = request.POST.getlist('testcases_list')
167 # 如果页面勾选了用例
168 if test_cases_list:
169 print("勾选用例id：", test_cases_list)
170 # 根据页面勾选的用例与查询出的所有用例一一比较
171 for test_case in test_cases_list:
172 test_case = models.TestCase.objects.get(id=int(test_case))
173 # 匹配成功则添加用例
174 models.SuiteCase.objects.create(case_suite=case_suite, test_case=test_case)
175 # 未勾选用例
176 else:
177 print("添加测试用例失败")
178 return HttpResponse("添加的测试用例为空，请选择用例后再添加！")
179 return render(request, 'add_case_in_suite.html',
180 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
181
182
183 # 用例集合页-查看/删除用例
184 @login_required
185 def show_and_delete_case_in_suite(request, suite_id):
186 case_suite = models.CaseSuite.objects.get(id=suite_id)
187 test_cases = models.SuiteCase.objects.filter(case_suite=case_suite)
188 if request.method == "POST":
189 test_cases_list = request.POST.getlist('test_cases_list')
190 if test_cases_list:
191 print("勾选用例：", test_cases_list)
192 for test_case in test_cases_list:
193 test_case = models.TestCase.objects.get(id=int(test_case))
194 models.SuiteCase.objects.filter(case_suite=case_suite, test_case=test_case).first().delete()
195 else:
196 print("测试用例删除失败")
197 return HttpResponse("所选测试用例为空，请选择用例后再进行删除！")
198 case_suite = models.CaseSuite.objects.get(id=suite_id)
199 return render(request, 'show_and_delete_case_in_suite.html',
200 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
201
202
203 # 用例执行结果-菜单项
204 @login_required
205 def test_case_execute_record(request):
206 test_case_execute_records = models.TestCaseExecuteResult.objects.filter().order_by('-id')
207 return render(request, 'test_case_execute_records.html', {'test_case_execute_records':
208 get_paginator(request, test_case_execute_records)})
209
210
211 # 用例执行结果-对比差异
212 @login_required
213 def case_result_diff(request, test_record_id):
214 test_record_data = models.TestCaseExecuteResult.objects.get(id=test_record_id)
215 print("用例执行结果记录: {}".format(test_record_data))
216 present_response = test_record_data.response_data
217 if present_response:
218 present_response = json.dumps(json.loads(present_response), sort_keys=True, indent=4,
219 ensure_ascii=False) # 中文字符不转ascii编码
220 print("当前响应结果: {}".format(present_response))
221 last_time_execute_response = test_record_data.last_time_response_data
222 if last_time_execute_response:
223 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True, indent=4,
224 ensure_ascii=False)
225 print("上一次响应结果: {}".format(last_time_execute_response))
226 return render(request, 'case_result_diff.html', locals())
227
228
229 # 用例执行结果-异常信息展示
230 @login_required
231 def show_exception(request, execute_id):
232 test_record = models.TestCaseExecuteResult.objects.get(id=execute_id)
233 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
234
235
236 # 用例集合执行结果
237 @login_required
238 def case_suite_execute_record(request):
239 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.filter().order_by('-id')
240 return render(request, 'case_suite_execute_record.html',
241 {'case_suite_execute_records': get_paginator(request, case_suite_execute_record)})
242
243
244 # 用例集合执行结果-包含用例结果展示
245 @login_required
246 def suite_case_execute_record(request, suite_record_id):
247 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.get(id=suite_record_id)
248 suite_case_execute_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=case_suite_execute_record)
249 return render(request, 'suite_case_execute_record.html',
250 {'suite_case_execute_records': get_paginator(request, suite_case_execute_records)})
251
252
253 # 用例集合执行结果-包含用例结果展示-差异比对
254 @login_required
255 def suite_case_result_diff(request, suite_case_record_id):
256 suite_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
257 present_response = suite_record_data.response_data
258 if present_response:
259 present_response = json.dumps(json.loads(present_response),sort_keys=True, indent=4, ensure_ascii=False)
260 print("当前响应: {}".format(present_response))
261 last_time_execute_response = suite_record_data.last_time_response_data
262 if last_time_execute_response:
263 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True,
264 indent=4, ensure_ascii=False)
265 print("上一次响应: {}".format(last_time_execute_response))
266 return render(request, 'case_result_diff.html', locals())
267
268
269 # 用例集合执行结果-包含用例结果展示-异常信息展示
270 @login_required
271 def suite_case_exception(request, suite_case_record_id):
272 test_record = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
273 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
274
275
276 # 默认页的视图函数
277 @login_required
278 def index(request):
279 return render(request, 'index.html')
280
281
282 # 登录页的视图函数
283 def login(request):
284 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
285 if request.session.get('is_login', None):
286 return redirect('/')
287 # 如果是表单提交行为，则进行登录校验
288 if request.method == "POST":
289 login_form = UserForm(request.POST)
290 message = "请检查填写的内容！"
291 if login_form.is_valid():
292 username = login_form.cleaned_data['username']
293 password = login_form.cleaned_data['password']
294 try:
295 # 使用django提供的身份验证功能
296 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
297 if user is not None:
298 print("用户【%s】登录成功" % username)
299 auth.login(request, user)
300 request.session['is_login'] = True
301 # 登录成功，跳转主页
302 return redirect('/')
303 else:
304 message = "用户名不存在或者密码不正确！"
305 except:
306 traceback.print_exc()
307 message = "登录程序出现异常"
308 # 用户名或密码为空，返回登录页和错误提示信息
309 else:
310 return render(request, 'login.html', locals())
311 # 不是表单提交，代表只是访问登录页
312 else:
313 login_form = UserForm()
314 return render(request, 'login.html', locals())
315
316
317 # 注册页的视图函数
318 def register(request):
319 return render(request, 'register.html')
320
321
322 # 登出的视图函数：重定向至login视图函数
323 @login_required
324 def logout(request):
325 auth.logout(request)
326 request.session.flush()
327 return redirect("/login/")

3）定义模板

修改“测试集合执行结果”模板中，测试集合名称链接：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试集合执行结果{% endblock %}
4
5 {% block content %}
6
7

8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 {% for case_suite_execute_record in case_suite_execute_records %} 23 24 25 26 27 {% if case_suite_execute_record.status %} 28 29 {% else %} 30 31 {% endif %} 32 33 {% ifequal case_suite_execute_record.test_result '成功' %} 34 35 {% else %} 36 37 {% endifequal %} 38 39 40 41 42 43 {% endfor %} 44 45 46 47

id	测试集合名称	延迟执行时间	执行状态	测试结果	创建者	创建时间
{{ case_suite_execute_record.id }}	{{ case_suite_execute_record.case_suite.suite_desc }}	{{ case_suite_execute_record.run_time_interval }}	执行完毕	待执行	{{ case_suite_execute_record.test_result}}	{{ case_suite_execute_record.test_result}}	{{ case_suite_execute_record.creator }}	{{ case_suite_execute_record.create_time\|date:"Y-n-d H:i" }}

49
50 {# 实现分页标签的代码 #}
51 {# 这里使用 bootstrap 渲染页面 #}
52

56 {% if case_suite_execute_records.has_previous %} 57 上一页 58 {% endif %} 59 60 61 第 {{ case_suite_execute_records.number }} 页 / 共 {{ case_suite_execute_records.paginator.num_pages }} 页 62 63 {% if case_suite_execute_records.has_next %} 64 下一页 65 {% endif %} 66

70 {% endblock %}

新增用例集合关联的测试用例执行结果页：templates/suite_case_execute_record.html

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例集合用例执行结果{% endblock %}
4
5 {% block content %}
6
7

8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 {% for case_execute_record in suite_case_execute_records %} 29 30 31 32 33 {% if case_execute_record.status %} 34 35 {% else %} 36 37 {% endif %} 38 39 40 41 42 43 {% ifequal case_execute_record.execute_result '成功' %} 44 45 {% else %} 46 47 {% endifequal %} 48 49 <！----!> 50 {% if case_execute_record.exception_info %} 51 52 {% else %} 53 54 {% endif %} 55 56 57 58 59 60 {% endfor %} 61 62 63 64

执行id	集合名称	用例名称	状态	请求数据	执行返回结果	操作	断言内容	执行结果	异常信息	请求后提取变量	开始时间	执行耗时(ms)
{{ case_execute_record.id }}	{{ case_execute_record.case_suite_record.case_suite.suite_desc }}	{{ case_execute_record.test_case.case_name }}	执行完毕	待执行	{{ case_execute_record.request_data }}	{{ case_execute_record.response_data }}	对比差异	{{ case_execute_record.test_case.assert_key }}	{{ case_execute_record.execute_result}}	{{ case_execute_record.execute_result}}	{{ case_execute_record.execute_result\|default_if_none:"" }}	显示异常	无	{{ case_execute_record.extract_var }}	{{ case_execute_record.execute_start_time }}	{{ case_execute_record.execute_total_time }}

66
67 {# 实现分页标签的代码 #}
68 {# 这里使用 bootstrap 渲染页面 #}
69

73 {% if suite_case_execute_records.has_previous %} 74 上一页 75 {% endif %} 76 77 78 第 {{ suite_case_execute_records.number }} 页 / 共 {{ suite_case_execute_records.paginator.num_pages }} 页 79 80 {% if suite_case_execute_records.has_next %} 81 下一页 82 {% endif %} 83

84 {% endblock %}

12. 用例集合历史执行结果统计

from django.urls import path, re_path
from . import views

1 from django.shortcuts import render, redirect, HttpResponse
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage
5 from .form import UserForm
6 import traceback
7 import json
8 from . import models
9 from .task import case_task, suite_task
10
11
12 # 封装分页处理
13 def get_paginator(request, data):
14 paginator = Paginator(data, 10) # 默认每页展示10条数据
15 # 获取 url 后面的 page 参数的值, 首页不显示 page 参数, 默认值是 1
16 page = request.GET.get('page')
17 try:
18 paginator_pages = paginator.page(page)
19 except PageNotAnInteger:
20 # 如果请求的页数不是整数, 返回第一页。
21 paginator_pages = paginator.page(1)
22 except InvalidPage:
23 # 如果请求的页数不存在, 重定向页面
24 return HttpResponse('找不到页面的内容')
25 return paginator_pages
26
27
28 # 项目菜单项
29 @login_required
30 def project(request):
31 print("request.user.is_authenticated: ", request.user.is_authenticated)
32 projects = models.Project.objects.filter().order_by('-id')
33 print("projects:", projects)
34 return render(request, 'project.html', {'projects': get_paginator(request, projects)})
35
36
37 # 模块菜单项
38 @login_required
39 def module(request):
40 if request.method == "GET": # 请求get时候，id倒序查询所有的模块数据
41 modules = models.Module.objects.filter().order_by('-id')
42 return render(request, 'module.html', {'modules': get_paginator(request, modules)})
43 else: # 否则就是Post请求，会根据输入内容，使用模糊的方式查找所有的项目
44 proj_name = request.POST['proj_name']
45 projects = models.Project.objects.filter(name__contains=proj_name.strip())
46 projs = [proj.id for proj in projects]
47 modules = models.Module.objects.filter(belong_project__in=projs) # 把项目中所有的模块都找出来
48 return render(request, 'module.html', {'modules': get_paginator(request, modules), 'proj_name': proj_name})
49
50
51 # 获取测试用例执行的接口地址
52 def get_server_address(env):
53 if env: # 环境处理
54 env_data = models.InterfaceServer.objects.filter(env=env[0])
55 print("env_data: {}".format(env_data))
56 if env_data:
57 ip = env_data[0].ip
58 port = env_data[0].port
59 print("ip: {}, port: {}".format(ip, port))
60 server_address = "http://{}:{}".format(ip, port)
61 print("server_address: {}".format(server_address))
62 return server_address
63 else:
64 return ""
65 else:
66 return ""
67
68
69 # 测试用例菜单项
70 @login_required
71 def test_case(request):
72 print("request.session['is_login']: {}".format(request.session['is_login']))
73 test_cases = ""
74 if request.method == "GET":
75 test_cases = models.TestCase.objects.filter().order_by('id')
76 print("testcases: {}".format(test_cases))
77 elif request.method == "POST":
78 print("request.POST: {}".format(request.POST))
79 test_case_id_list = request.POST.getlist('test_cases_list')
80 env = request.POST.getlist('env')
81 print("env: {}".format(env))
82 server_address = get_server_address(env)
83 if not server_address:
84 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
85 if test_case_id_list:
86 test_case_id_list.sort()
87 print("test_case_id_list: {}".format(test_case_id_list))
88 print("获取到用例，开始用例执行")
89 case_task(test_case_id_list, server_address)
90 else:
91 print("运行测试用例失败")
92 return HttpResponse("提交的运行测试用例为空，请选择用例后在提交！")
93 test_cases = models.TestCase.objects.filter().order_by('id')
94 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
95
96
97 # 用例详情页
98 @login_required
99 def test_case_detail(request, test_case_id):
100 test_case_id = int(test_case_id)
101 test_case = models.TestCase.objects.get(id=test_case_id)
102 print("test_case: {}".format(test_case))
103 print("test_case.id: {}".format(test_case.id))
104 print("test_case.belong_project: {}".format(test_case.belong_project))
105
106 return render(request, 'test_case_detail.html', {'test_case': test_case})
107
108
109 # 模块页展示测试用例
110 @login_required
111 def module_test_cases(request, module_id):
112 module = ""
113 if module_id: # 访问的时候，会从url中提取模块的id，根据模块id查询到模块数据，在模板中展现
114 module = models.Module.objects.get(id=int(module_id))
115 test_cases = models.TestCase.objects.filter(belong_module=module).order_by('-id')
116 print("test_case in module_test_cases: {}".format(test_cases))
117 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
118
119
120 # 用例集合菜单项
121 @login_required
122 def case_suite(request):
123 if request.method == "POST":
124 count_down_time = 0
125 if request.POST['delay_time']:
126 print("输入的延迟时间是: {}".format(request.POST['delay_time']))
127 try:
128 count_down_time = int(request.POST['delay_time'])
129 except:
130 print("输入的延迟时间是非数字！")
131 else:
132 print("没有输入延迟时间")
133 env = request.POST.getlist('env')
134 print("env: {}".format(env))
135 server_address = get_server_address(env)
136 if not server_address:
137 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
138 case_suite_list = request.POST.getlist('case_suite_list')
139 if case_suite_list:
140 print("所需执行的用例集合列表：", case_suite_list)
141 for suite_id in case_suite_list:
142 test_suite = models.CaseSuite.objects.get(id=int(suite_id))
143 print("所需执行的用例集合: {}".format(test_suite))
144 username = request.user.username
145 test_suite_record = models.CaseSuiteExecuteRecord.objects.create(case_suite=test_suite,
146 run_time_interval=count_down_time,
147 creator=username)
148 suite_task(test_suite_record, test_suite, server_address)
149 else:
150 print("运行测试集合用例失败")
151 return HttpResponse("运行的测试集合为空，请选择测试集合后再运行！")
152 case_suites = models.CaseSuite.objects.filter()
153 return render(request, 'case_suite.html', {'case_suites': get_paginator(request, case_suites)})
154
155
156 # 用例集合-添加测试用例页
157 @login_required
158 def add_case_in_suite(request, suite_id):
159 # 查询指定的用例集合
160 case_suite = models.CaseSuite.objects.get(id=suite_id)
161 # 根据id号查询所有的用例
162 test_cases = models.TestCase.objects.filter().order_by('id')
163 if request.method == "GET":
164 print("test cases:", test_cases)
165 elif request.method == "POST":
166 test_cases_list = request.POST.getlist('testcases_list')
167 # 如果页面勾选了用例
168 if test_cases_list:
169 print("勾选用例id：", test_cases_list)
170 # 根据页面勾选的用例与查询出的所有用例一一比较
171 for test_case in test_cases_list:
172 test_case = models.TestCase.objects.get(id=int(test_case))
173 # 匹配成功则添加用例
174 models.SuiteCase.objects.create(case_suite=case_suite, test_case=test_case)
175 # 未勾选用例
176 else:
177 print("添加测试用例失败")
178 return HttpResponse("添加的测试用例为空，请选择用例后再添加！")
179 return render(request, 'add_case_in_suite.html',
180 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
181
182
183 # 用例集合页-查看/删除用例
184 @login_required
185 def show_and_delete_case_in_suite(request, suite_id):
186 case_suite = models.CaseSuite.objects.get(id=suite_id)
187 test_cases = models.SuiteCase.objects.filter(case_suite=case_suite)
188 if request.method == "POST":
189 test_cases_list = request.POST.getlist('test_cases_list')
190 if test_cases_list:
191 print("勾选用例：", test_cases_list)
192 for test_case in test_cases_list:
193 test_case = models.TestCase.objects.get(id=int(test_case))
194 models.SuiteCase.objects.filter(case_suite=case_suite, test_case=test_case).first().delete()
195 else:
196 print("测试用例删除失败")
197 return HttpResponse("所选测试用例为空，请选择用例后再进行删除！")
198 case_suite = models.CaseSuite.objects.get(id=suite_id)
199 return render(request, 'show_and_delete_case_in_suite.html',
200 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
201
202
203 # 用例执行结果-菜单项
204 @login_required
205 def test_case_execute_record(request):
206 test_case_execute_records = models.TestCaseExecuteResult.objects.filter().order_by('-id')
207 return render(request, 'test_case_execute_records.html', {'test_case_execute_records':
208 get_paginator(request, test_case_execute_records)})
209
210
211 # 用例执行结果-对比差异
212 @login_required
213 def case_result_diff(request, test_record_id):
214 test_record_data = models.TestCaseExecuteResult.objects.get(id=test_record_id)
215 print("用例执行结果记录: {}".format(test_record_data))
216 present_response = test_record_data.response_data
217 if present_response:
218 present_response = json.dumps(json.loads(present_response), sort_keys=True, indent=4,
219 ensure_ascii=False) # 中文字符不转ascii编码
220 print("当前响应结果: {}".format(present_response))
221 last_time_execute_response = test_record_data.last_time_response_data
222 if last_time_execute_response:
223 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True, indent=4,
224 ensure_ascii=False)
225 print("上一次响应结果: {}".format(last_time_execute_response))
226 return render(request, 'case_result_diff.html', locals())
227
228
229 # 用例执行结果-异常信息展示
230 @login_required
231 def show_exception(request, execute_id):
232 test_record = models.TestCaseExecuteResult.objects.get(id=execute_id)
233 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
234
235
236 # 用例集合执行结果
237 @login_required
238 def case_suite_execute_record(request):
239 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.filter().order_by('-id')
240 return render(request, 'case_suite_execute_record.html',
241 {'case_suite_execute_records': get_paginator(request, case_suite_execute_record)})
242
243
244 # 用例集合执行结果-包含用例结果展示
245 @login_required
246 def suite_case_execute_record(request, suite_record_id):
247 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.get(id=suite_record_id)
248 suite_case_execute_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=case_suite_execute_record)
249 return render(request, 'suite_case_execute_record.html',
250 {'suite_case_execute_records': get_paginator(request, suite_case_execute_records)})
251
252
253 # 用例集合执行结果-包含用例结果展示-差异比对
254 @login_required
255 def suite_case_result_diff(request, suite_case_record_id):
256 suite_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
257 present_response = suite_record_data.response_data
258 if present_response:
259 present_response = json.dumps(json.loads(present_response),sort_keys=True, indent=4, ensure_ascii=False)
260 print("当前响应: {}".format(present_response))
261 last_time_execute_response = suite_record_data.last_time_response_data
262 if last_time_execute_response:
263 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True,
264 indent=4, ensure_ascii=False)
265 print("上一次响应: {}".format(last_time_execute_response))
266 return render(request, 'case_result_diff.html', locals())
267
268
269 # 用例集合执行结果-包含用例结果展示-异常信息展示
270 @login_required
271 def suite_case_exception(request, suite_case_record_id):
272 test_record = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
273 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
274
275
276 # 用例集合执行结果统计
277 def case_suite_statistics(request, suite_id):
278 case_suite = models.CaseSuite.objects.get(id=suite_id)
279 success_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="成功"))
280 fail_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="失败"))
281 case_suite_records = models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite).order_by('-id')
282 return render(request, 'case_suite_statistics.html',
283 {'case_suite_records': get_paginator(request, case_suite_records), 'success_num': success_num,
284 'fail_num': fail_num})
285
286
287 # 默认页的视图函数
288 @login_required
289 def index(request):
290 return render(request, 'index.html')
291
292
293 # 登录页的视图函数
294 def login(request):
295 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
296 if request.session.get('is_login', None):
297 return redirect('/')
298 # 如果是表单提交行为，则进行登录校验
299 if request.method == "POST":
300 login_form = UserForm(request.POST)
301 message = "请检查填写的内容！"
302 if login_form.is_valid():
303 username = login_form.cleaned_data['username']
304 password = login_form.cleaned_data['password']
305 try:
306 # 使用django提供的身份验证功能
307 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
308 if user is not None:
309 print("用户【%s】登录成功" % username)
310 auth.login(request, user)
311 request.session['is_login'] = True
312 # 登录成功，跳转主页
313 return redirect('/')
314 else:
315 message = "用户名不存在或者密码不正确！"
316 except:
317 traceback.print_exc()
318 message = "登录程序出现异常"
319 # 用户名或密码为空，返回登录页和错误提示信息
320 else:
321 return render(request, 'login.html', locals())
322 # 不是表单提交，代表只是访问登录页
323 else:
324 login_form = UserForm()
325 return render(request, 'login.html', locals())
326
327
328 # 注册页的视图函数
329 def register(request):
330 return render(request, 'register.html')
331
332
333 # 登出的视图函数：重定向至login视图函数
334 @login_required
335 def logout(request):
336 auth.logout(request)
337 request.session.flush()
338 return redirect("/login/")

修改 case_suite.html 模板，增加查看统计结果链接：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例集合{% endblock %}
4 {% block content %}
5
40
41

42 {% csrf_token %} 43 延迟执行的时间（单位：秒）: 44 45 运行环境： 46 50 51 52

53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 {% for case_suite in case_suites %} 69 70 71 72 73 74 75 76 77 78 79 {% endfor %} 80 81

全选	id	测试集合名称	创建者	创建时间	查看/删除测试用例	添加测试用例	用例集合执行结果
	{{ case_suite.id }}	{{ case_suite.suite_desc }}	{{ case_suite.creator }}	{{ case_suite.create_time\|date:"Y-n-d H:i" }}	查看/删除测试用例	添加测试用例	查看用例集合执行结果

84
85 {# 实现分页标签的代码 #}
86 {# 这里使用 bootstrap 渲染页面 #}
87

88 104

105 {% endblock %}

新增统计结果页面模板：templates/case_suite_statistics.html

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试集合结果统计{% endblock %}
4 {% block content %}
5
6
7 8

9 用例集合执行结果统计：成功 {{ success_num }} 次，失败 {{ fail_num }} 次 10

12 13 14 58 59

60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 {% for case_suite_record in case_suite_records %} 75 76 77 78 79 {% if case_suite_record.status %} 80 81 {% else %} 82 83 {% endif %} 84 85 86 87 88 {% endfor %} 89 90 91

id	测试集合名称	延迟执行时间	执行状态	测试结果	创建者	创建时间
{{ case_suite_record.id }}	{{ case_suite_record.case_suite.suite_desc }}	{{ case_suite_record.run_time_interval }}	执行完毕	待执行	{{ case_suite_record.test_result\|default_if_none:"" }}	{{ case_suite_record.creator }}	{{ case_suite_record.create_time\|date:"Y-n-d H:i" }}

93 94 {# 实现分页标签的代码 #} 95 {# 这里使用 bootstrap 渲染页面 #} 96

97 113

114
115
116 {% endblock %}

13. 用例集合单次执行结果统计

from django.urls import path, re_path
from . import views

1 from django.shortcuts import render, redirect, HttpResponse
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage
5 from .form import UserForm
6 import traceback
7 import json
8 from . import models
9 from .task import case_task, suite_task
10
11
12 # 封装分页处理
13 def get_paginator(request, data):
14 paginator = Paginator(data, 10) # 默认每页展示10条数据
15 # 获取 url 后面的 page 参数的值, 首页不显示 page 参数, 默认值是 1
16 page = request.GET.get('page')
17 try:
18 paginator_pages = paginator.page(page)
19 except PageNotAnInteger:
20 # 如果请求的页数不是整数, 返回第一页。
21 paginator_pages = paginator.page(1)
22 except InvalidPage:
23 # 如果请求的页数不存在, 重定向页面
24 return HttpResponse('找不到页面的内容')
25 return paginator_pages
26
27
28 # 项目菜单项
29 @login_required
30 def project(request):
31 print("request.user.is_authenticated: ", request.user.is_authenticated)
32 projects = models.Project.objects.filter().order_by('-id')
33 print("projects:", projects)
34 return render(request, 'project.html', {'projects': get_paginator(request, projects)})
35
36
37 # 模块菜单项
38 @login_required
39 def module(request):
40 if request.method == "GET": # 请求get时候，id倒序查询所有的模块数据
41 modules = models.Module.objects.filter().order_by('-id')
42 return render(request, 'module.html', {'modules': get_paginator(request, modules)})
43 else: # 否则就是Post请求，会根据输入内容，使用模糊的方式查找所有的项目
44 proj_name = request.POST['proj_name']
45 projects = models.Project.objects.filter(name__contains=proj_name.strip())
46 projs = [proj.id for proj in projects]
47 modules = models.Module.objects.filter(belong_project__in=projs) # 把项目中所有的模块都找出来
48 return render(request, 'module.html', {'modules': get_paginator(request, modules), 'proj_name': proj_name})
49
50
51 # 获取测试用例执行的接口地址
52 def get_server_address(env):
53 if env: # 环境处理
54 env_data = models.InterfaceServer.objects.filter(env=env[0])
55 print("env_data: {}".format(env_data))
56 if env_data:
57 ip = env_data[0].ip
58 port = env_data[0].port
59 print("ip: {}, port: {}".format(ip, port))
60 server_address = "http://{}:{}".format(ip, port)
61 print("server_address: {}".format(server_address))
62 return server_address
63 else:
64 return ""
65 else:
66 return ""
67
68
69 # 测试用例菜单项
70 @login_required
71 def test_case(request):
72 print("request.session['is_login']: {}".format(request.session['is_login']))
73 test_cases = ""
74 if request.method == "GET":
75 test_cases = models.TestCase.objects.filter().order_by('id')
76 print("testcases: {}".format(test_cases))
77 elif request.method == "POST":
78 print("request.POST: {}".format(request.POST))
79 test_case_id_list = request.POST.getlist('test_cases_list')
80 env = request.POST.getlist('env')
81 print("env: {}".format(env))
82 server_address = get_server_address(env)
83 if not server_address:
84 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
85 if test_case_id_list:
86 test_case_id_list.sort()
87 print("test_case_id_list: {}".format(test_case_id_list))
88 print("获取到用例，开始用例执行")
89 case_task(test_case_id_list, server_address)
90 else:
91 print("运行测试用例失败")
92 return HttpResponse("提交的运行测试用例为空，请选择用例后在提交！")
93 test_cases = models.TestCase.objects.filter().order_by('id')
94 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
95
96
97 # 用例详情页
98 @login_required
99 def test_case_detail(request, test_case_id):
100 test_case_id = int(test_case_id)
101 test_case = models.TestCase.objects.get(id=test_case_id)
102 print("test_case: {}".format(test_case))
103 print("test_case.id: {}".format(test_case.id))
104 print("test_case.belong_project: {}".format(test_case.belong_project))
105
106 return render(request, 'test_case_detail.html', {'test_case': test_case})
107
108
109 # 模块页展示测试用例
110 @login_required
111 def module_test_cases(request, module_id):
112 module = ""
113 if module_id: # 访问的时候，会从url中提取模块的id，根据模块id查询到模块数据，在模板中展现
114 module = models.Module.objects.get(id=int(module_id))
115 test_cases = models.TestCase.objects.filter(belong_module=module).order_by('-id')
116 print("test_case in module_test_cases: {}".format(test_cases))
117 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
118
119
120 # 用例集合菜单项
121 @login_required
122 def case_suite(request):
123 if request.method == "POST":
124 count_down_time = 0
125 if request.POST['delay_time']:
126 print("输入的延迟时间是: {}".format(request.POST['delay_time']))
127 try:
128 count_down_time = int(request.POST['delay_time'])
129 except:
130 print("输入的延迟时间是非数字！")
131 else:
132 print("没有输入延迟时间")
133 env = request.POST.getlist('env')
134 print("env: {}".format(env))
135 server_address = get_server_address(env)
136 if not server_address:
137 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
138 case_suite_list = request.POST.getlist('case_suite_list')
139 if case_suite_list:
140 print("所需执行的用例集合列表：", case_suite_list)
141 for suite_id in case_suite_list:
142 test_suite = models.CaseSuite.objects.get(id=int(suite_id))
143 print("所需执行的用例集合: {}".format(test_suite))
144 username = request.user.username
145 test_suite_record = models.CaseSuiteExecuteRecord.objects.create(case_suite=test_suite,
146 run_time_interval=count_down_time,
147 creator=username)
148 suite_task(test_suite_record, test_suite, server_address)
149 else:
150 print("运行测试集合用例失败")
151 return HttpResponse("运行的测试集合为空，请选择测试集合后再运行！")
152 case_suites = models.CaseSuite.objects.filter()
153 return render(request, 'case_suite.html', {'case_suites': get_paginator(request, case_suites)})
154
155
156 # 用例集合-添加测试用例页
157 @login_required
158 def add_case_in_suite(request, suite_id):
159 # 查询指定的用例集合
160 case_suite = models.CaseSuite.objects.get(id=suite_id)
161 # 根据id号查询所有的用例
162 test_cases = models.TestCase.objects.filter().order_by('id')
163 if request.method == "GET":
164 print("test cases:", test_cases)
165 elif request.method == "POST":
166 test_cases_list = request.POST.getlist('testcases_list')
167 # 如果页面勾选了用例
168 if test_cases_list:
169 print("勾选用例id：", test_cases_list)
170 # 根据页面勾选的用例与查询出的所有用例一一比较
171 for test_case in test_cases_list:
172 test_case = models.TestCase.objects.get(id=int(test_case))
173 # 匹配成功则添加用例
174 models.SuiteCase.objects.create(case_suite=case_suite, test_case=test_case)
175 # 未勾选用例
176 else:
177 print("添加测试用例失败")
178 return HttpResponse("添加的测试用例为空，请选择用例后再添加！")
179 return render(request, 'add_case_in_suite.html',
180 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
181
182
183 # 用例集合页-查看/删除用例
184 @login_required
185 def show_and_delete_case_in_suite(request, suite_id):
186 case_suite = models.CaseSuite.objects.get(id=suite_id)
187 test_cases = models.SuiteCase.objects.filter(case_suite=case_suite)
188 if request.method == "POST":
189 test_cases_list = request.POST.getlist('test_cases_list')
190 if test_cases_list:
191 print("勾选用例：", test_cases_list)
192 for test_case in test_cases_list:
193 test_case = models.TestCase.objects.get(id=int(test_case))
194 models.SuiteCase.objects.filter(case_suite=case_suite, test_case=test_case).first().delete()
195 else:
196 print("测试用例删除失败")
197 return HttpResponse("所选测试用例为空，请选择用例后再进行删除！")
198 case_suite = models.CaseSuite.objects.get(id=suite_id)
199 return render(request, 'show_and_delete_case_in_suite.html',
200 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
201
202
203 # 用例执行结果-菜单项
204 @login_required
205 def test_case_execute_record(request):
206 test_case_execute_records = models.TestCaseExecuteResult.objects.filter().order_by('-id')
207 return render(request, 'test_case_execute_records.html', {'test_case_execute_records':
208 get_paginator(request, test_case_execute_records)})
209
210
211 # 用例执行结果-对比差异
212 @login_required
213 def case_result_diff(request, test_record_id):
214 test_record_data = models.TestCaseExecuteResult.objects.get(id=test_record_id)
215 print("用例执行结果记录: {}".format(test_record_data))
216 present_response = test_record_data.response_data
217 if present_response:
218 present_response = json.dumps(json.loads(present_response), sort_keys=True, indent=4,
219 ensure_ascii=False) # 中文字符不转ascii编码
220 print("当前响应结果: {}".format(present_response))
221 last_time_execute_response = test_record_data.last_time_response_data
222 if last_time_execute_response:
223 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True, indent=4,
224 ensure_ascii=False)
225 print("上一次响应结果: {}".format(last_time_execute_response))
226 return render(request, 'case_result_diff.html', locals())
227
228
229 # 用例执行结果-异常信息展示
230 @login_required
231 def show_exception(request, execute_id):
232 test_record = models.TestCaseExecuteResult.objects.get(id=execute_id)
233 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
234
235
236 # 用例集合执行结果
237 @login_required
238 def case_suite_execute_record(request):
239 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.filter().order_by('-id')
240 return render(request, 'case_suite_execute_record.html',
241 {'case_suite_execute_records': get_paginator(request, case_suite_execute_record)})
242
243
244 # 用例集合执行结果-包含用例结果展示
245 @login_required
246 def suite_case_execute_record(request, suite_record_id):
247 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.get(id=suite_record_id)
248 suite_case_execute_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=case_suite_execute_record)
249 return render(request, 'suite_case_execute_record.html',
250 {'suite_case_execute_records': get_paginator(request, suite_case_execute_records)})
251
252
253 # 用例集合执行结果-包含用例结果展示-差异比对
254 @login_required
255 def suite_case_result_diff(request, suite_case_record_id):
256 suite_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
257 present_response = suite_record_data.response_data
258 if present_response:
259 present_response = json.dumps(json.loads(present_response),sort_keys=True, indent=4, ensure_ascii=False)
260 print("当前响应: {}".format(present_response))
261 last_time_execute_response = suite_record_data.last_time_response_data
262 if last_time_execute_response:
263 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True,
264 indent=4, ensure_ascii=False)
265 print("上一次响应: {}".format(last_time_execute_response))
266 return render(request, 'case_result_diff.html', locals())
267
268
269 # 用例集合执行结果-包含用例结果展示-异常信息展示
270 @login_required
271 def suite_case_exception(request, suite_case_record_id):
272 test_record = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
273 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
274
275
276 # 用例集合执行结果单次统计
277 def suite_case_statistics(request, suite_id):
278 success_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="成功"))
279 fail_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="失败"))
280 suite_case_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id).order_by('-id')
281 return render(request, 'suite_case_statistics.html',
282 {'suite_case_records': get_paginator(request, suite_case_records), 'success_num': success_num,
283 'fail_num': fail_num})
284
285
286 # 用例集合执行结果历史统计
287 def case_suite_statistics(request, suite_id):
288 case_suite = models.CaseSuite.objects.get(id=suite_id)
289 success_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="成功"))
290 fail_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="失败"))
291 case_suite_records = models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite).order_by('-id')
292 return render(request, 'case_suite_statistics.html',
293 {'case_suite_records': get_paginator(request, case_suite_records), 'success_num': success_num,
294 'fail_num': fail_num})
295
296
297 # 默认页的视图函数
298 @login_required
299 def index(request):
300 return render(request, 'index.html')
301
302
303 # 登录页的视图函数
304 def login(request):
305 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
306 if request.session.get('is_login', None):
307 return redirect('/')
308 # 如果是表单提交行为，则进行登录校验
309 if request.method == "POST":
310 login_form = UserForm(request.POST)
311 message = "请检查填写的内容！"
312 if login_form.is_valid():
313 username = login_form.cleaned_data['username']
314 password = login_form.cleaned_data['password']
315 try:
316 # 使用django提供的身份验证功能
317 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
318 if user is not None:
319 print("用户【%s】登录成功" % username)
320 auth.login(request, user)
321 request.session['is_login'] = True
322 # 登录成功，跳转主页
323 return redirect('/')
324 else:
325 message = "用户名不存在或者密码不正确！"
326 except:
327 traceback.print_exc()
328 message = "登录程序出现异常"
329 # 用户名或密码为空，返回登录页和错误提示信息
330 else:
331 return render(request, 'login.html', locals())
332 # 不是表单提交，代表只是访问登录页
333 else:
334 login_form = UserForm()
335 return render(request, 'login.html', locals())
336
337
338 # 注册页的视图函数
339 def register(request):
340 return render(request, 'register.html')
341
342
343 # 登出的视图函数：重定向至login视图函数
344 @login_required
345 def logout(request):
346 auth.logout(request)
347 request.session.flush()
348 return redirect("/login/")

修改 case_suite_execute_record.html：新增测试结果统计链接

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}测试集合执行结果{% endblock %}
4
5 {% block content %}
6
7

8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for case_suite_execute_record in case_suite_execute_records %} 24 25 26 27 28 {% if case_suite_execute_record.status %} 29 30 {% else %} 31 32 {% endif %} 33 34 {% ifequal case_suite_execute_record.test_result '成功' %} 35 36 {% else %} 37 38 {% endifequal %} 39 40 41 42 43 {% endfor %} 44 45 46 47

id	测试集合名称	延迟执行时间	执行状态	测试结果	测试结果统计	创建者	创建时间
{{ case_suite_execute_record.id }}	{{ case_suite_execute_record.case_suite.suite_desc }}	{{ case_suite_execute_record.run_time_interval }}	执行完毕	待执行	{{ case_suite_execute_record.test_result}}	{{ case_suite_execute_record.test_result}}	测试结果统计	{{ case_suite_execute_record.creator }}	{{ case_suite_execute_record.create_time\|date:"Y-n-d H:i" }}

49
50 {# 实现分页标签的代码 #}
51 {# 这里使用 bootstrap 渲染页面 #}
52

56 {% if case_suite_execute_records.has_previous %} 57 上一页 58 {% endif %} 59 60 61 第 {{ case_suite_execute_records.number }} 页 / 共 {{ case_suite_execute_records.paginator.num_pages }} 页 62 63 {% if case_suite_execute_records.has_next %} 64 下一页 65 {% endif %} 66

70 {% endblock %}

新增 templates/suite_case_statistics.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}用例集合单次执行结果统计{% endblock %}
4 {% block content %}
5
6
7 8

9 用例集合执行结果统计：成功 {{ success_num }} 次，失败 {{ fail_num }} 次 10

12 13 14 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 {% for suite_case in suite_case_records %} 76 77 78 79 80 81 82 83 84 85 86 87 {% endfor %} 88 89

测试集合名称	用例id	用例名称	所属项目	所属模块	编写人员	创建时间	更新时间	创建用例用户名
{{suite_case.case_suite_record.case_suite.suite_desc}}	{{ suite_case.id }}	{{ suite_case.test_case.case_name }}	{{ suite_case.test_case.belong_project.name }}	{{ suite_case.test_case.belong_module.name }}	{{ suite_case.test_case.maintainer }}	{{ suite_case.test_case.created_time\|date:"Y-n-d H:i" }}	{{ suite_case.test_case.updated_time\|date:"Y-n-d H:i" }}	{{ suite_case.test_case.user }}

91 92 93 {# 实现分页标签的代码 #} 94 {# 这里使用 bootstrap 渲染页面 #} 95

99 {% if suite_case_records.has_previous %} 100 上一页 101 {% endif %} 102 103 104 第 {{ suite_case_records.number }} 页 / 共 {{ suite_case_records.paginator.num_pages }} 页 105 106 {% if suite_case_records.has_next %} 107 下一页 108 {% endif %} 109

111

112

113

114

115
116 {% endblock %}

14. 模块测试结果统计

from django.urls import path, re_path
from . import views

1 from django.shortcuts import render, redirect, HttpResponse
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage
5 from .form import UserForm
6 import traceback
7 import json
8 from . import models
9 from .task import case_task, suite_task
10
11
12 # 封装分页处理
13 def get_paginator(request, data):
14 paginator = Paginator(data, 10) # 默认每页展示10条数据
15 # 获取 url 后面的 page 参数的值, 首页不显示 page 参数, 默认值是 1
16 page = request.GET.get('page')
17 try:
18 paginator_pages = paginator.page(page)
19 except PageNotAnInteger:
20 # 如果请求的页数不是整数, 返回第一页。
21 paginator_pages = paginator.page(1)
22 except InvalidPage:
23 # 如果请求的页数不存在, 重定向页面
24 return HttpResponse('找不到页面的内容')
25 return paginator_pages
26
27
28 # 项目菜单项
29 @login_required
30 def project(request):
31 print("request.user.is_authenticated: ", request.user.is_authenticated)
32 projects = models.Project.objects.filter().order_by('-id')
33 print("projects:", projects)
34 return render(request, 'project.html', {'projects': get_paginator(request, projects)})
35
36
37 # 模块菜单项
38 @login_required
39 def module(request):
40 if request.method == "GET": # 请求get时候，id倒序查询所有的模块数据
41 modules = models.Module.objects.filter().order_by('-id')
42 return render(request, 'module.html', {'modules': get_paginator(request, modules)})
43 else: # 否则就是Post请求，会根据输入内容，使用模糊的方式查找所有的项目
44 proj_name = request.POST['proj_name']
45 projects = models.Project.objects.filter(name__contains=proj_name.strip())
46 projs = [proj.id for proj in projects]
47 modules = models.Module.objects.filter(belong_project__in=projs) # 把项目中所有的模块都找出来
48 return render(request, 'module.html', {'modules': get_paginator(request, modules), 'proj_name': proj_name})
49
50
51 # 获取测试用例执行的接口地址
52 def get_server_address(env):
53 if env: # 环境处理
54 env_data = models.InterfaceServer.objects.filter(env=env[0])
55 print("env_data: {}".format(env_data))
56 if env_data:
57 ip = env_data[0].ip
58 port = env_data[0].port
59 print("ip: {}, port: {}".format(ip, port))
60 server_address = "http://{}:{}".format(ip, port)
61 print("server_address: {}".format(server_address))
62 return server_address
63 else:
64 return ""
65 else:
66 return ""
67
68
69 # 测试用例菜单项
70 @login_required
71 def test_case(request):
72 print("request.session['is_login']: {}".format(request.session['is_login']))
73 test_cases = ""
74 if request.method == "GET":
75 test_cases = models.TestCase.objects.filter().order_by('id')
76 print("testcases: {}".format(test_cases))
77 elif request.method == "POST":
78 print("request.POST: {}".format(request.POST))
79 test_case_id_list = request.POST.getlist('test_cases_list')
80 env = request.POST.getlist('env')
81 print("env: {}".format(env))
82 server_address = get_server_address(env)
83 if not server_address:
84 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
85 if test_case_id_list:
86 test_case_id_list.sort()
87 print("test_case_id_list: {}".format(test_case_id_list))
88 print("获取到用例，开始用例执行")
89 case_task(test_case_id_list, server_address)
90 else:
91 print("运行测试用例失败")
92 return HttpResponse("提交的运行测试用例为空，请选择用例后在提交！")
93 test_cases = models.TestCase.objects.filter().order_by('id')
94 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
95
96
97 # 用例详情页
98 @login_required
99 def test_case_detail(request, test_case_id):
100 test_case_id = int(test_case_id)
101 test_case = models.TestCase.objects.get(id=test_case_id)
102 print("test_case: {}".format(test_case))
103 print("test_case.id: {}".format(test_case.id))
104 print("test_case.belong_project: {}".format(test_case.belong_project))
105
106 return render(request, 'test_case_detail.html', {'test_case': test_case})
107
108
109 # 模块页展示测试用例
110 @login_required
111 def module_test_cases(request, module_id):
112 module = ""
113 if module_id: # 访问的时候，会从url中提取模块的id，根据模块id查询到模块数据，在模板中展现
114 module = models.Module.objects.get(id=int(module_id))
115 test_cases = models.TestCase.objects.filter(belong_module=module).order_by('-id')
116 print("test_case in module_test_cases: {}".format(test_cases))
117 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
118
119
120 # 用例集合菜单项
121 @login_required
122 def case_suite(request):
123 if request.method == "POST":
124 count_down_time = 0
125 if request.POST['delay_time']:
126 print("输入的延迟时间是: {}".format(request.POST['delay_time']))
127 try:
128 count_down_time = int(request.POST['delay_time'])
129 except:
130 print("输入的延迟时间是非数字！")
131 else:
132 print("没有输入延迟时间")
133 env = request.POST.getlist('env')
134 print("env: {}".format(env))
135 server_address = get_server_address(env)
136 if not server_address:
137 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
138 case_suite_list = request.POST.getlist('case_suite_list')
139 if case_suite_list:
140 print("所需执行的用例集合列表：", case_suite_list)
141 for suite_id in case_suite_list:
142 test_suite = models.CaseSuite.objects.get(id=int(suite_id))
143 print("所需执行的用例集合: {}".format(test_suite))
144 username = request.user.username
145 test_suite_record = models.CaseSuiteExecuteRecord.objects.create(case_suite=test_suite,
146 run_time_interval=count_down_time,
147 creator=username)
148 suite_task(test_suite_record, test_suite, server_address)
149 else:
150 print("运行测试集合用例失败")
151 return HttpResponse("运行的测试集合为空，请选择测试集合后再运行！")
152 case_suites = models.CaseSuite.objects.filter()
153 return render(request, 'case_suite.html', {'case_suites': get_paginator(request, case_suites)})
154
155
156 # 用例集合-添加测试用例页
157 @login_required
158 def add_case_in_suite(request, suite_id):
159 # 查询指定的用例集合
160 case_suite = models.CaseSuite.objects.get(id=suite_id)
161 # 根据id号查询所有的用例
162 test_cases = models.TestCase.objects.filter().order_by('id')
163 if request.method == "GET":
164 print("test cases:", test_cases)
165 elif request.method == "POST":
166 test_cases_list = request.POST.getlist('testcases_list')
167 # 如果页面勾选了用例
168 if test_cases_list:
169 print("勾选用例id：", test_cases_list)
170 # 根据页面勾选的用例与查询出的所有用例一一比较
171 for test_case in test_cases_list:
172 test_case = models.TestCase.objects.get(id=int(test_case))
173 # 匹配成功则添加用例
174 models.SuiteCase.objects.create(case_suite=case_suite, test_case=test_case)
175 # 未勾选用例
176 else:
177 print("添加测试用例失败")
178 return HttpResponse("添加的测试用例为空，请选择用例后再添加！")
179 return render(request, 'add_case_in_suite.html',
180 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
181
182
183 # 用例集合页-查看/删除用例
184 @login_required
185 def show_and_delete_case_in_suite(request, suite_id):
186 case_suite = models.CaseSuite.objects.get(id=suite_id)
187 test_cases = models.SuiteCase.objects.filter(case_suite=case_suite)
188 if request.method == "POST":
189 test_cases_list = request.POST.getlist('test_cases_list')
190 if test_cases_list:
191 print("勾选用例：", test_cases_list)
192 for test_case in test_cases_list:
193 test_case = models.TestCase.objects.get(id=int(test_case))
194 models.SuiteCase.objects.filter(case_suite=case_suite, test_case=test_case).first().delete()
195 else:
196 print("测试用例删除失败")
197 return HttpResponse("所选测试用例为空，请选择用例后再进行删除！")
198 case_suite = models.CaseSuite.objects.get(id=suite_id)
199 return render(request, 'show_and_delete_case_in_suite.html',
200 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
201
202
203 # 用例执行结果-菜单项
204 @login_required
205 def test_case_execute_record(request):
206 test_case_execute_records = models.TestCaseExecuteResult.objects.filter().order_by('-id')
207 return render(request, 'test_case_execute_records.html', {'test_case_execute_records':
208 get_paginator(request, test_case_execute_records)})
209
210
211 # 用例执行结果-对比差异
212 @login_required
213 def case_result_diff(request, test_record_id):
214 test_record_data = models.TestCaseExecuteResult.objects.get(id=test_record_id)
215 print("用例执行结果记录: {}".format(test_record_data))
216 present_response = test_record_data.response_data
217 if present_response:
218 present_response = json.dumps(json.loads(present_response), sort_keys=True, indent=4,
219 ensure_ascii=False) # 中文字符不转ascii编码
220 print("当前响应结果: {}".format(present_response))
221 last_time_execute_response = test_record_data.last_time_response_data
222 if last_time_execute_response:
223 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True, indent=4,
224 ensure_ascii=False)
225 print("上一次响应结果: {}".format(last_time_execute_response))
226 return render(request, 'case_result_diff.html', locals())
227
228
229 # 用例执行结果-异常信息展示
230 @login_required
231 def show_exception(request, execute_id):
232 test_record = models.TestCaseExecuteResult.objects.get(id=execute_id)
233 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
234
235
236 # 用例集合执行结果
237 @login_required
238 def case_suite_execute_record(request):
239 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.filter().order_by('-id')
240 return render(request, 'case_suite_execute_record.html',
241 {'case_suite_execute_records': get_paginator(request, case_suite_execute_record)})
242
243
244 # 用例集合执行结果-包含用例结果展示
245 @login_required
246 def suite_case_execute_record(request, suite_record_id):
247 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.get(id=suite_record_id)
248 suite_case_execute_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=case_suite_execute_record)
249 return render(request, 'suite_case_execute_record.html',
250 {'suite_case_execute_records': get_paginator(request, suite_case_execute_records)})
251
252
253 # 用例集合执行结果-包含用例结果展示-差异比对
254 @login_required
255 def suite_case_result_diff(request, suite_case_record_id):
256 suite_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
257 present_response = suite_record_data.response_data
258 if present_response:
259 present_response = json.dumps(json.loads(present_response),sort_keys=True, indent=4, ensure_ascii=False)
260 print("当前响应: {}".format(present_response))
261 last_time_execute_response = suite_record_data.last_time_response_data
262 if last_time_execute_response:
263 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True,
264 indent=4, ensure_ascii=False)
265 print("上一次响应: {}".format(last_time_execute_response))
266 return render(request, 'case_result_diff.html', locals())
267
268
269 # 用例集合执行结果-包含用例结果展示-异常信息展示
270 @login_required
271 def suite_case_exception(request, suite_case_record_id):
272 test_record = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
273 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
274
275
276 # 用例集合执行结果单次统计
277 def suite_case_statistics(request, suite_id):
278 success_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="成功"))
279 fail_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="失败"))
280 suite_case_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id).order_by('-id')
281 return render(request, 'suite_case_statistics.html',
282 {'suite_case_records': get_paginator(request, suite_case_records), 'success_num': success_num,
283 'fail_num': fail_num})
284
285
286 # 用例集合执行结果历史统计
287 def case_suite_statistics(request, suite_id):
288 case_suite = models.CaseSuite.objects.get(id=suite_id)
289 success_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="成功"))
290 fail_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="失败"))
291 case_suite_records = models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite).order_by('-id')
292 return render(request, 'case_suite_statistics.html',
293 {'case_suite_records': get_paginator(request, case_suite_records), 'success_num': success_num,
294 'fail_num': fail_num})
295
296 # 模块测试结果统计
297 @login_required
298 def module_statistics(request, module_id):
299 test_module = models.Module.objects.get(id=int(module_id))
300 test_cases = models.TestCase.objects.filter(belong_module=test_module)
301 test_suit_success_num = len(
302 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="成功"))
303 test_suit_fail_num = len(
304 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="失败"))
305 test_case_success_num = len(
306 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="成功"))
307 test_case_fail_num = len(
308 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="失败"))
309 success_num = test_suit_success_num + test_case_success_num
310 fail_num = test_suit_fail_num + test_case_fail_num
311 return render(request, 'module_statistics.html',
312 {'test_module': test_module, 'success_num': success_num, 'fail_num': fail_num})
313
314
315 # 默认页的视图函数
316 @login_required
317 def index(request):
318 return render(request, 'index.html')
319
320
321 # 登录页的视图函数
322 def login(request):
323 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
324 if request.session.get('is_login', None):
325 return redirect('/')
326 # 如果是表单提交行为，则进行登录校验
327 if request.method == "POST":
328 login_form = UserForm(request.POST)
329 message = "请检查填写的内容！"
330 if login_form.is_valid():
331 username = login_form.cleaned_data['username']
332 password = login_form.cleaned_data['password']
333 try:
334 # 使用django提供的身份验证功能
335 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
336 if user is not None:
337 print("用户【%s】登录成功" % username)
338 auth.login(request, user)
339 request.session['is_login'] = True
340 # 登录成功，跳转主页
341 return redirect('/')
342 else:
343 message = "用户名不存在或者密码不正确！"
344 except:
345 traceback.print_exc()
346 message = "登录程序出现异常"
347 # 用户名或密码为空，返回登录页和错误提示信息
348 else:
349 return render(request, 'login.html', locals())
350 # 不是表单提交，代表只是访问登录页
351 else:
352 login_form = UserForm()
353 return render(request, 'login.html', locals())
354
355
356 # 注册页的视图函数
357 def register(request):
358 return render(request, 'register.html')
359
360
361 # 登出的视图函数：重定向至login视图函数
362 @login_required
363 def logout(request):
364 auth.logout(request)
365 request.session.flush()
366 return redirect("/login/")

模块页面 module.html 新增测试结果统计的链接：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}模块{% endblock %}
4
5 {% block content %}
6

11
12

13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 {% for module in modules %} 30 31 32 33 34 35 36 37 38 39 40 {% endfor %} 41 42 43

id	模块名称	所属项目	测试负责人	模块描述	创建时间	更新时间	测试结果统计
{{ module.id }}	{{ module.name }}	{{ module.belong_project.name }}	{{ module.test_owner }}	{{ module.desc }}	{{ module.create_time\|date:"Y-n-d H:i" }}	{{ module.update_time\|date:"Y-n-d H:i" }}	查看

45
46 {# 实现分页标签的代码 #}
47 {# 这里使用 bootstrap 渲染页面 #}
48

49 65

66 {% endblock %}

新增模块测试结果统计模板 module_statistics.html：

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}模块测试结果统计{% endblock %}
4 {% block content %}
5
6
15 16

17 【{{ test_module.name }}】执行统计结果：成功 {{ success_num }} 次，失败 {{ fail_num }} 次 18

20 21

23 24 25 60
61
62 {% endblock %}

15. 项目测试结果统计

from django.urls import path, re_path
from . import views

1 from django.shortcuts import render, redirect, HttpResponse
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage
5 from .form import UserForm
6 import traceback
7 import json
8 from . import models
9 from .task import case_task, suite_task
10
11
12 # 封装分页处理
13 def get_paginator(request, data):
14 paginator = Paginator(data, 10) # 默认每页展示10条数据
15 # 获取 url 后面的 page 参数的值, 首页不显示 page 参数, 默认值是 1
16 page = request.GET.get('page')
17 try:
18 paginator_pages = paginator.page(page)
19 except PageNotAnInteger:
20 # 如果请求的页数不是整数, 返回第一页。
21 paginator_pages = paginator.page(1)
22 except InvalidPage:
23 # 如果请求的页数不存在, 重定向页面
24 return HttpResponse('找不到页面的内容')
25 return paginator_pages
26
27
28 # 项目菜单项
29 @login_required
30 def project(request):
31 print("request.user.is_authenticated: ", request.user.is_authenticated)
32 projects = models.Project.objects.filter().order_by('-id')
33 print("projects:", projects)
34 return render(request, 'project.html', {'projects': get_paginator(request, projects)})
35
36
37 # 模块菜单项
38 @login_required
39 def module(request):
40 if request.method == "GET": # 请求get时候，id倒序查询所有的模块数据
41 modules = models.Module.objects.filter().order_by('-id')
42 return render(request, 'module.html', {'modules': get_paginator(request, modules)})
43 else: # 否则就是Post请求，会根据输入内容，使用模糊的方式查找所有的项目
44 proj_name = request.POST['proj_name']
45 projects = models.Project.objects.filter(name__contains=proj_name.strip())
46 projs = [proj.id for proj in projects]
47 modules = models.Module.objects.filter(belong_project__in=projs) # 把项目中所有的模块都找出来
48 return render(request, 'module.html', {'modules': get_paginator(request, modules), 'proj_name': proj_name})
49
50
51 # 获取测试用例执行的接口地址
52 def get_server_address(env):
53 if env: # 环境处理
54 env_data = models.InterfaceServer.objects.filter(env=env[0])
55 print("env_data: {}".format(env_data))
56 if env_data:
57 ip = env_data[0].ip
58 port = env_data[0].port
59 print("ip: {}, port: {}".format(ip, port))
60 server_address = "http://{}:{}".format(ip, port)
61 print("server_address: {}".format(server_address))
62 return server_address
63 else:
64 return ""
65 else:
66 return ""
67
68
69 # 测试用例菜单项
70 @login_required
71 def test_case(request):
72 print("request.session['is_login']: {}".format(request.session['is_login']))
73 test_cases = ""
74 if request.method == "GET":
75 test_cases = models.TestCase.objects.filter().order_by('id')
76 print("testcases: {}".format(test_cases))
77 elif request.method == "POST":
78 print("request.POST: {}".format(request.POST))
79 test_case_id_list = request.POST.getlist('test_cases_list')
80 env = request.POST.getlist('env')
81 print("env: {}".format(env))
82 server_address = get_server_address(env)
83 if not server_address:
84 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
85 if test_case_id_list:
86 test_case_id_list.sort()
87 print("test_case_id_list: {}".format(test_case_id_list))
88 print("获取到用例，开始用例执行")
89 case_task(test_case_id_list, server_address)
90 else:
91 print("运行测试用例失败")
92 return HttpResponse("提交的运行测试用例为空，请选择用例后在提交！")
93 test_cases = models.TestCase.objects.filter().order_by('id')
94 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
95
96
97 # 用例详情页
98 @login_required
99 def test_case_detail(request, test_case_id):
100 test_case_id = int(test_case_id)
101 test_case = models.TestCase.objects.get(id=test_case_id)
102 print("test_case: {}".format(test_case))
103 print("test_case.id: {}".format(test_case.id))
104 print("test_case.belong_project: {}".format(test_case.belong_project))
105
106 return render(request, 'test_case_detail.html', {'test_case': test_case})
107
108
109 # 模块页展示测试用例
110 @login_required
111 def module_test_cases(request, module_id):
112 module = ""
113 if module_id: # 访问的时候，会从url中提取模块的id，根据模块id查询到模块数据，在模板中展现
114 module = models.Module.objects.get(id=int(module_id))
115 test_cases = models.TestCase.objects.filter(belong_module=module).order_by('-id')
116 print("test_case in module_test_cases: {}".format(test_cases))
117 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
118
119
120 # 用例集合菜单项
121 @login_required
122 def case_suite(request):
123 if request.method == "POST":
124 count_down_time = 0
125 if request.POST['delay_time']:
126 print("输入的延迟时间是: {}".format(request.POST['delay_time']))
127 try:
128 count_down_time = int(request.POST['delay_time'])
129 except:
130 print("输入的延迟时间是非数字！")
131 else:
132 print("没有输入延迟时间")
133 env = request.POST.getlist('env')
134 print("env: {}".format(env))
135 server_address = get_server_address(env)
136 if not server_address:
137 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
138 case_suite_list = request.POST.getlist('case_suite_list')
139 if case_suite_list:
140 print("所需执行的用例集合列表：", case_suite_list)
141 for suite_id in case_suite_list:
142 test_suite = models.CaseSuite.objects.get(id=int(suite_id))
143 print("所需执行的用例集合: {}".format(test_suite))
144 username = request.user.username
145 test_suite_record = models.CaseSuiteExecuteRecord.objects.create(case_suite=test_suite,
146 run_time_interval=count_down_time,
147 creator=username)
148 suite_task(test_suite_record, test_suite, server_address)
149 else:
150 print("运行测试集合用例失败")
151 return HttpResponse("运行的测试集合为空，请选择测试集合后再运行！")
152 case_suites = models.CaseSuite.objects.filter()
153 return render(request, 'case_suite.html', {'case_suites': get_paginator(request, case_suites)})
154
155
156 # 用例集合-添加测试用例页
157 @login_required
158 def add_case_in_suite(request, suite_id):
159 # 查询指定的用例集合
160 case_suite = models.CaseSuite.objects.get(id=suite_id)
161 # 根据id号查询所有的用例
162 test_cases = models.TestCase.objects.filter().order_by('id')
163 if request.method == "GET":
164 print("test cases:", test_cases)
165 elif request.method == "POST":
166 test_cases_list = request.POST.getlist('testcases_list')
167 # 如果页面勾选了用例
168 if test_cases_list:
169 print("勾选用例id：", test_cases_list)
170 # 根据页面勾选的用例与查询出的所有用例一一比较
171 for test_case in test_cases_list:
172 test_case = models.TestCase.objects.get(id=int(test_case))
173 # 匹配成功则添加用例
174 models.SuiteCase.objects.create(case_suite=case_suite, test_case=test_case)
175 # 未勾选用例
176 else:
177 print("添加测试用例失败")
178 return HttpResponse("添加的测试用例为空，请选择用例后再添加！")
179 return render(request, 'add_case_in_suite.html',
180 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
181
182
183 # 用例集合页-查看/删除用例
184 @login_required
185 def show_and_delete_case_in_suite(request, suite_id):
186 case_suite = models.CaseSuite.objects.get(id=suite_id)
187 test_cases = models.SuiteCase.objects.filter(case_suite=case_suite)
188 if request.method == "POST":
189 test_cases_list = request.POST.getlist('test_cases_list')
190 if test_cases_list:
191 print("勾选用例：", test_cases_list)
192 for test_case in test_cases_list:
193 test_case = models.TestCase.objects.get(id=int(test_case))
194 models.SuiteCase.objects.filter(case_suite=case_suite, test_case=test_case).first().delete()
195 else:
196 print("测试用例删除失败")
197 return HttpResponse("所选测试用例为空，请选择用例后再进行删除！")
198 case_suite = models.CaseSuite.objects.get(id=suite_id)
199 return render(request, 'show_and_delete_case_in_suite.html',
200 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
201
202
203 # 用例执行结果-菜单项
204 @login_required
205 def test_case_execute_record(request):
206 test_case_execute_records = models.TestCaseExecuteResult.objects.filter().order_by('-id')
207 return render(request, 'test_case_execute_records.html', {'test_case_execute_records':
208 get_paginator(request, test_case_execute_records)})
209
210
211 # 用例执行结果-对比差异
212 @login_required
213 def case_result_diff(request, test_record_id):
214 test_record_data = models.TestCaseExecuteResult.objects.get(id=test_record_id)
215 print("用例执行结果记录: {}".format(test_record_data))
216 present_response = test_record_data.response_data
217 if present_response:
218 present_response = json.dumps(json.loads(present_response), sort_keys=True, indent=4,
219 ensure_ascii=False) # 中文字符不转ascii编码
220 print("当前响应结果: {}".format(present_response))
221 last_time_execute_response = test_record_data.last_time_response_data
222 if last_time_execute_response:
223 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True, indent=4,
224 ensure_ascii=False)
225 print("上一次响应结果: {}".format(last_time_execute_response))
226 return render(request, 'case_result_diff.html', locals())
227
228
229 # 用例执行结果-异常信息展示
230 @login_required
231 def show_exception(request, execute_id):
232 test_record = models.TestCaseExecuteResult.objects.get(id=execute_id)
233 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
234
235
236 # 用例集合执行结果
237 @login_required
238 def case_suite_execute_record(request):
239 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.filter().order_by('-id')
240 return render(request, 'case_suite_execute_record.html',
241 {'case_suite_execute_records': get_paginator(request, case_suite_execute_record)})
242
243
244 # 用例集合执行结果-包含用例结果展示
245 @login_required
246 def suite_case_execute_record(request, suite_record_id):
247 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.get(id=suite_record_id)
248 suite_case_execute_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=case_suite_execute_record)
249 return render(request, 'suite_case_execute_record.html',
250 {'suite_case_execute_records': get_paginator(request, suite_case_execute_records)})
251
252
253 # 用例集合执行结果-包含用例结果展示-差异比对
254 @login_required
255 def suite_case_result_diff(request, suite_case_record_id):
256 suite_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
257 present_response = suite_record_data.response_data
258 if present_response:
259 present_response = json.dumps(json.loads(present_response),sort_keys=True, indent=4, ensure_ascii=False)
260 print("当前响应: {}".format(present_response))
261 last_time_execute_response = suite_record_data.last_time_response_data
262 if last_time_execute_response:
263 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True,
264 indent=4, ensure_ascii=False)
265 print("上一次响应: {}".format(last_time_execute_response))
266 return render(request, 'case_result_diff.html', locals())
267
268
269 # 用例集合执行结果-包含用例结果展示-异常信息展示
270 @login_required
271 def suite_case_exception(request, suite_case_record_id):
272 test_record = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
273 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
274
275
276 # 用例集合执行结果单次统计
277 def suite_case_statistics(request, suite_id):
278 success_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="成功"))
279 fail_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="失败"))
280 suite_case_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id).order_by('-id')
281 return render(request, 'suite_case_statistics.html',
282 {'suite_case_records': get_paginator(request, suite_case_records), 'success_num': success_num,
283 'fail_num': fail_num})
284
285
286 # 用例集合执行结果历史统计
287 def case_suite_statistics(request, suite_id):
288 case_suite = models.CaseSuite.objects.get(id=suite_id)
289 success_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="成功"))
290 fail_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="失败"))
291 case_suite_records = models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite).order_by('-id')
292 return render(request, 'case_suite_statistics.html',
293 {'case_suite_records': get_paginator(request, case_suite_records), 'success_num': success_num,
294 'fail_num': fail_num})
295
296
297 # 模块测试结果统计
298 @login_required
299 def module_statistics(request, module_id):
300 test_module = models.Module.objects.get(id=int(module_id))
301 test_cases = models.TestCase.objects.filter(belong_module=test_module)
302 test_suit_success_num = len(
303 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="成功"))
304 test_suit_fail_num = len(
305 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="失败"))
306 test_case_success_num = len(
307 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="成功"))
308 test_case_fail_num = len(
309 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="失败"))
310 success_num = test_suit_success_num + test_case_success_num
311 fail_num = test_suit_fail_num + test_case_fail_num
312 return render(request, 'module_statistics.html',
313 {'test_module': test_module, 'success_num': success_num, 'fail_num': fail_num})
314
315
316 # 项目测试结果统计
317 @login_required
318 def project_statistics(request, project_id):
319 test_project = models.Project.objects.get(id=int(project_id))
320 test_cases = models.TestCase.objects.filter(belong_project=test_project)
321 test_suit_success_num = len(
322 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="成功"))
323 test_suit_fail_num = len(
324 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="失败"))
325 test_case_success_num = len(
326 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="成功"))
327 test_case_fail_num = len(
328 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="失败"))
329 success_num = test_suit_success_num + test_case_success_num
330 fail_num = test_suit_fail_num + test_case_fail_num
331 return render(request, 'project_statistics.html',
332 {'test_project': test_project, 'success_num': success_num, 'fail_num': fail_num})
333
334
335 # 默认页的视图函数
336 @login_required
337 def index(request):
338 return render(request, 'index.html')
339
340
341 # 登录页的视图函数
342 def login(request):
343 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
344 if request.session.get('is_login', None):
345 return redirect('/')
346 # 如果是表单提交行为，则进行登录校验
347 if request.method == "POST":
348 login_form = UserForm(request.POST)
349 message = "请检查填写的内容！"
350 if login_form.is_valid():
351 username = login_form.cleaned_data['username']
352 password = login_form.cleaned_data['password']
353 try:
354 # 使用django提供的身份验证功能
355 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
356 if user is not None:
357 print("用户【%s】登录成功" % username)
358 auth.login(request, user)
359 request.session['is_login'] = True
360 # 登录成功，跳转主页
361 return redirect('/')
362 else:
363 message = "用户名不存在或者密码不正确！"
364 except:
365 traceback.print_exc()
366 message = "登录程序出现异常"
367 # 用户名或密码为空，返回登录页和错误提示信息
368 else:
369 return render(request, 'login.html', locals())
370 # 不是表单提交，代表只是访问登录页
371 else:
372 login_form = UserForm()
373 return render(request, 'login.html', locals())
374
375
376 # 注册页的视图函数
377 def register(request):
378 return render(request, 'register.html')
379
380
381 # 登出的视图函数：重定向至login视图函数
382 @login_required
383 def logout(request):
384 auth.logout(request)
385 request.session.flush()
386 return redirect("/login/")

修改项目页模板 project.html：新增测试结果统计的链接

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}主页{% endblock %}
4
5 {% block content %}
6

7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 {% for project in projects %} 24 25 26 27 28 29 30 31 32 33 34 35 {% endfor %} 36 37

id	项目名称	项目负责人	测试负责人	开发负责人	简要描述	创建时间	更新时间	测试结果统计
{{ project.id }}	{{ project.name }}	{{ project.proj_owner }}	{{ project.test_owner }}	{{ project.dev_owner }}	{{ project.desc }}	{{ project.create_time\|date:"Y-n-d H:i" }}	{{ project.update_time\|date:"Y-n-d H:i" }}	查看

39
40 {# 实现分页标签的代码 #}
41 {# 这里使用 bootstrap 渲染页面 #}
42

43 59

60 {% endblock %}

新增项目测试结果统计模板：project_statistics.html

1 {% extends 'base.html' %}
2 {% load static %}
3 {% block title %}模块测试结果统计{% endblock %}
4 {% block content %}
5
6
14 15

16 【{{ test_project.name }}】执行统计结果：成功 {{ success_num }} 次，失败 {{ fail_num }} 次 17

19 20

22 23 24 59
60
61 {% endblock %}

16. Celery 异步执行用例

本项目使用 Redis 存储 Celery 的任务执行结果，因此 Redis 需同时启动着。

1）定义 celery app：在应用目录下新建 celery.py

1 from __future__ import absolute_import, unicode_literals
2 import os
3 from celery import Celery
4 from django.conf import settings
5
6
7 # 参数为项目名称
8 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'InterfaceAutoTest.settings') # 设置django环境
9 # 参数为项目名称
10 app = Celery('InterfaceAutoTest', backend='redis://127.0.0.1:6379/1', broker='redis://127.0.0.1:6379/0')
11
12 app.config_from_object('django.conf:settings') # 使用CELERY_作为前缀，在settings中写配置
13
14 app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) # 发现任务文件每个app下的task.py
15
16 # 时区
17 app.conf.timezone = 'Asia/Shanghai'
18 # 是否使用UTC
19 app.conf.enable_utc = False

2）引入 celery app：应用目录下（新建） __init__.py

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ['celery_app']

在应用下的 task.py 中修改如下代码：把用例执行代码通过装饰器装饰成 celery 任务

1 from __future__ import absolute_import, unicode_literals
2 from celery import shared_task
3 import time
4 import os
5 import traceback
6 import json
7 from . import models
8 from .utils.data_process import data_preprocess, assert_result, data_postprocess
9 from .utils.request_process import request_process
10
11
12 # 测试用例执行
13 @shared_task
14 def case_task(test_case_id_list, server_address):
15 global_key = 'case'+ str(int(time.time() * 100000))
16 os.environ[global_key] = '{}'
17 print()
18 print("全局变量标识符【global_key】: {}".format(global_key))
19 print("全局变量内容【os.environ[global_key]】: {}".format(os.environ[global_key]))
20 for test_case_id in test_case_id_list:
21
22 test_case = models.TestCase.objects.filter(id=int(test_case_id))[0]
23 last_execute_record_data = models.TestCaseExecuteResult.objects.filter(
24 belong_test_case_id=test_case_id).order_by('-id')
25 if last_execute_record_data:
26 last_time_execute_response_data = last_execute_record_data[0].response_data
27 else:
28 last_time_execute_response_data = ''
29 print("上一次响应结果: {}".format(last_execute_record_data))
30 print("上一次响应时间: {}".format(last_time_execute_response_data))
31 execute_record = models.TestCaseExecuteResult.objects.create(belong_test_case=test_case)
32 execute_record.last_time_response_data = last_time_execute_response_data
33 # 获取当前用例上一次执行结果
34 execute_record.save()
35
36 test_case = models.TestCase.objects.filter(id=int(test_case_id))[0]
37 print("\n######### 开始执行用例【{}】 #########".format(test_case))
38 execute_start_time = time.time() # 记录时间戳，便于计算总耗时（毫秒）
39 execute_record.execute_start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_start_time))
40
41 request_data = test_case.request_data
42 extract_var = test_case.extract_var
43 assert_key = test_case.assert_key
44 interface_name = test_case.uri
45 belong_project = test_case.belong_project
46 belong_module = test_case.belong_module
47 maintainer = test_case.maintainer
48 request_method = test_case.request_method
49 print("初始请求数据: {}".format(request_data))
50 print("关联参数: {}".format(extract_var))
51 print("断言关键字: {}".format(assert_key))
52 print("接口名称: {}".format(interface_name))
53 print("所属项目: {}".format(belong_project))
54 print("所属模块: {}".format(belong_module))
55 print("用例维护人: {}".format(maintainer))
56 print("请求方法: {}".format(request_method))
57 url = "{}{}".format(server_address, interface_name)
58 print("接口地址: {}".format(url))
59 code, request_data, error_msg = data_preprocess(global_key, str(request_data))
60 # 请求数据预处理异常，结束用例执行
61 if code != 0:
62 print("数据处理异常，error: {}".format(error_msg))
63 execute_record.execute_result = "失败"
64 execute_record.status = 1
65 execute_record.exception_info = error_msg
66 execute_end_time = time.time()
67 execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_end_time))
68 execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
69 execute_record.save()
70 return
71 # 记录请求预处理结果
72 else:
73 execute_record.request_data = request_data
74 # 调用接口
75 try:
76 res_data = request_process(url, request_method, json.loads(request_data))
77 print("响应数据: {}".format(json.dumps(res_data.json(), ensure_ascii=False))) # ensure_ascii：兼容中文
78 result_flag, exception_info = assert_result(res_data, assert_key)
79 # 结果记录保存
80 if result_flag:
81 print("用例【%s】执行成功！" % test_case)
82 execute_record.execute_result = "成功"
83 if extract_var.strip() != "None":
84 var_value = data_postprocess(global_key, json.dumps(res_data.json(), ensure_ascii=False), extract_var)
85 execute_record.extract_var = var_value
86 else:
87 print("用例【%s】执行失败！" % test_case)
88 execute_record.execute_result = "失败"
89 execute_record.exception_info = exception_info
90 execute_record.response_data = json.dumps(res_data.json(), ensure_ascii=False)
91 execute_record.status = 1
92 execute_end_time = time.time()
93 execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_end_time))
94 print("执行结果结束时间: {}".format(execute_record.execute_end_time))
95 execute_record.execute_total_time = int((execute_end_time - execute_start_time) * 1000)
96 print("用例执行耗时: {}".format(execute_record.execute_total_time))
97 execute_record.save()
98 except Exception as e:
99 print("接口请求异常，error: {}".format(traceback.format_exc()))
100 execute_record.execute_result = "失败"
101 execute_record.exception_info = traceback.format_exc()
102 execute_record.status = 1
103 execute_end_time = time.time()
104 execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(execute_end_time))
105 print("执行结果结束时间: {}".format(execute_record.execute_end_time))
106 execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
107 print("用例执行耗时: {} 毫秒".format(execute_record.execute_total_time))
108 execute_record.save()
109
110
111 # 用例集合执行
112 @shared_task
113 def suite_task(case_suite_record, case_suite, server_address):
114 global_key = case_suite.suite_desc + str(int(time.time() * 100000))
115 # global_vars = {"{}".format(global_key): {}}
116 os.environ[global_key] = '{}'
117 print("global_key: {}".format(global_key))
118 print("os.environ[global_key]: {}".format(os.environ[global_key]))
119 case_suite_test_cases = models.SuiteCase.objects.filter(case_suite=case_suite).order_by('id')
120 print("用例集合的测试用例列表: {}".format(case_suite_test_cases))
121 case_suite_record.test_result = "成功"
122 case_suite_record.execute_start_time = time.strftime("%Y-%m-%d %H:%M:%S")
123
124 for case_suite_test_case in case_suite_test_cases:
125 test_case = case_suite_test_case.test_case
126 print("\n######### 开始执行用例【{}】 #########".format(test_case))
127 last_execute_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.filter(
128 test_case_id=test_case.id).order_by('-id')
129 if last_execute_record_data:
130 last_time_execute_response_data = last_execute_record_data[0].response_data
131 else:
132 last_time_execute_response_data = ''
133 print("上一次响应结果: {}".format(last_execute_record_data))
134 print("上一次响应时间: {}".format(last_time_execute_response_data))
135 suite_case_execute_record = models.CaseSuiteTestCaseExecuteRecord.objects.create(case_suite_record=case_suite_record,
136 test_case=test_case)
137 execute_start_time = time.time() # 记录时间戳，便于计算总耗时（毫秒）
138 suite_case_execute_record.execute_start_time = time.strftime("%Y-%m-%d %H:%M:%S",
139 time.localtime(execute_start_time))
140 print("用例集合开始执行时间: {}".format(suite_case_execute_record.execute_start_time))
141 suite_case_execute_record.last_time_response_data = last_time_execute_response_data
142 suite_case_execute_record.save()
143 request_data = test_case.request_data
144 extract_var = test_case.extract_var
145 assert_key = test_case.assert_key
146 interface_name = test_case.uri
147 belong_project = test_case.belong_project
148 belong_module = test_case.belong_module
149 maintainer = test_case.maintainer
150 request_method = test_case.request_method
151 print("初始请求数据: {}".format(request_data))
152 print("关联参数: {}".format(extract_var))
153 print("断言关键字: {}".format(assert_key))
154 print("接口名称: {}".format(interface_name))
155 print("所属项目: {}".format(belong_project))
156 print("所属模块: {}".format(belong_module))
157 print("用例维护人: {}".format(maintainer))
158 print("请求方法: {}".format(request_method))
159 url = "{}{}".format(server_address, interface_name)
160 print("接口地址: {}".format(url))
161 # 请求数据预处理
162 code, request_data, error_msg = data_preprocess(global_key, str(request_data))
163 # 请求数据预处理异常，结束用例执行
164 if code != 0:
165 print("数据处理异常，error: {}".format(error_msg))
166 suite_case_execute_record.execute_result = "失败"
167 suite_case_execute_record.status = 1
168 suite_case_execute_record.exception_info = error_msg
169 execute_end_time = time.time()
170 suite_case_execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S",
171 time.localtime(execute_end_time))
172 suite_case_execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
173 suite_case_execute_record.save()
174 case_suite_record.test_result = "失败"
175 # 记录请求预处理的结果
176 suite_case_execute_record.request_data = request_data
177 try:
178 # 调用接口
179 res_data = request_process(url, request_method, json.loads(request_data))
180 print("响应数据: {}".format(json.dumps(res_data.json(), ensure_ascii=False)))
181
182 result_flag, exception_info = assert_result(res_data, assert_key)
183 # 结果记录保存
184 if result_flag:
185 print("用例【%s】执行成功！" % test_case)
186 suite_case_execute_record.execute_result = "成功"
187 if extract_var.strip() != "None":
188 var_value = data_postprocess(global_key, json.dumps(res_data.json(), ensure_ascii=False),
189 extract_var)
190 suite_case_execute_record.extract_var = var_value
191 else:
192 print("用例【%s】执行失败！" % test_case)
193 suite_case_execute_record.execute_result = "失败"
194 suite_case_execute_record.exception_info = exception_info
195 case_suite_record.test_result = "失败"
196 suite_case_execute_record.response_data = json.dumps(res_data.json(), ensure_ascii=False)
197 suite_case_execute_record.status = 1
198 execute_end_time = time.time()
199 suite_case_execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S",
200 time.localtime(execute_end_time))
201 suite_case_execute_record.execute_total_time = int((execute_end_time - execute_start_time) * 1000)
202 print("用例执行耗时: {} 毫秒".format(
203 suite_case_execute_record.execute_total_time))
204 suite_case_execute_record.save()
205 except Exception as e:
206 print("接口请求异常，error: {}".format(e))
207 suite_case_execute_record.execute_result = "失败"
208 suite_case_execute_record.exception_info = traceback.format_exc()
209 suite_case_execute_record.status = 1
210 execute_end_time = time.time()
211 suite_case_execute_record.execute_end_time = time.strftime("%Y-%m-%d %H:%M:%S",
212 time.localtime(execute_end_time))
213 suite_case_execute_record.execute_total_time = int(execute_end_time - execute_start_time) * 1000
214 print("用例集合执行总耗时: {} 毫秒".format(suite_case_execute_record.execute_total_time))
215 suite_case_execute_record.save()
216 case_suite_record.test_result = "失败"
217
218 case_suite_record.status = 1 # 执行完毕
219 case_suite_record.save()

在项目目录下执行如下命令（-A 后接应用名称）：

celery worker -A interfacetestplatform -l info -P eventlet

启动成功日志如下：

需要注意，在 celery 的 task.py 中调用的函数如果有修改，则需要重启 celery。

修改用例执行和用例集合执行的视图函数，改为使用 celery 执行的方式。

1 from django.shortcuts import render, redirect, HttpResponse
2 from django.contrib import auth # Django用户认证（Auth）组件一般用在用户的登录注册上，用于判断当前的用户是否合法
3 from django.contrib.auth.decorators import login_required
4 from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage
5 from .form import UserForm
6 import traceback
7 import json
8 from . import models
9 from .task import case_task, suite_task
10
11
12 # 封装分页处理
13 def get_paginator(request, data):
14 paginator = Paginator(data, 10) # 默认每页展示10条数据
15 # 获取 url 后面的 page 参数的值, 首页不显示 page 参数, 默认值是 1
16 page = request.GET.get('page')
17 try:
18 paginator_pages = paginator.page(page)
19 except PageNotAnInteger:
20 # 如果请求的页数不是整数, 返回第一页。
21 paginator_pages = paginator.page(1)
22 except InvalidPage:
23 # 如果请求的页数不存在, 重定向页面
24 return HttpResponse('找不到页面的内容')
25 return paginator_pages
26
27
28 # 项目菜单项
29 @login_required
30 def project(request):
31 print("request.user.is_authenticated: ", request.user.is_authenticated)
32 projects = models.Project.objects.filter().order_by('-id')
33 print("projects:", projects)
34 return render(request, 'project.html', {'projects': get_paginator(request, projects)})
35
36
37 # 模块菜单项
38 @login_required
39 def module(request):
40 if request.method == "GET": # 请求get时候，id倒序查询所有的模块数据
41 modules = models.Module.objects.filter().order_by('-id')
42 return render(request, 'module.html', {'modules': get_paginator(request, modules)})
43 else: # 否则就是Post请求，会根据输入内容，使用模糊的方式查找所有的项目
44 proj_name = request.POST['proj_name']
45 projects = models.Project.objects.filter(name__contains=proj_name.strip())
46 projs = [proj.id for proj in projects]
47 modules = models.Module.objects.filter(belong_project__in=projs) # 把项目中所有的模块都找出来
48 return render(request, 'module.html', {'modules': get_paginator(request, modules), 'proj_name': proj_name})
49
50
51 # 获取测试用例执行的接口地址
52 def get_server_address(env):
53 if env: # 环境处理
54 env_data = models.InterfaceServer.objects.filter(env=env[0])
55 print("env_data: {}".format(env_data))
56 if env_data:
57 ip = env_data[0].ip
58 port = env_data[0].port
59 print("ip: {}, port: {}".format(ip, port))
60 server_address = "http://{}:{}".format(ip, port)
61 print("server_address: {}".format(server_address))
62 return server_address
63 else:
64 return ""
65 else:
66 return ""
67
68
69 # 测试用例菜单项
70 @login_required
71 def test_case(request):
72 print("request.session['is_login']: {}".format(request.session['is_login']))
73 test_cases = ""
74 if request.method == "GET":
75 test_cases = models.TestCase.objects.filter().order_by('id')
76 print("testcases: {}".format(test_cases))
77 elif request.method == "POST":
78 print("request.POST: {}".format(request.POST))
79 test_case_id_list = request.POST.getlist('test_cases_list')
80 env = request.POST.getlist('env')
81 print("env: {}".format(env))
82 server_address = get_server_address(env)
83 if not server_address:
84 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
85 if test_case_id_list:
86 test_case_id_list.sort()
87 print("test_case_id_list: {}".format(test_case_id_list))
88 print("获取到用例，开始用例执行")
89 # 普通执行
90 # case_task(test_case_id_list, server_address)
91 # celery 执行
92 case_task.apply_async((test_case_id_list, server_address))
93 else:
94 print("运行测试用例失败")
95 return HttpResponse("提交的运行测试用例为空，请选择用例后在提交！")
96 test_cases = models.TestCase.objects.filter().order_by('id')
97 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
98
99
100 # 用例详情页
101 @login_required
102 def test_case_detail(request, test_case_id):
103 test_case_id = int(test_case_id)
104 test_case = models.TestCase.objects.get(id=test_case_id)
105 print("test_case: {}".format(test_case))
106 print("test_case.id: {}".format(test_case.id))
107 print("test_case.belong_project: {}".format(test_case.belong_project))
108
109 return render(request, 'test_case_detail.html', {'test_case': test_case})
110
111
112 # 模块页展示测试用例
113 @login_required
114 def module_test_cases(request, module_id):
115 module = ""
116 if module_id: # 访问的时候，会从url中提取模块的id，根据模块id查询到模块数据，在模板中展现
117 module = models.Module.objects.get(id=int(module_id))
118 test_cases = models.TestCase.objects.filter(belong_module=module).order_by('-id')
119 print("test_case in module_test_cases: {}".format(test_cases))
120 return render(request, 'test_case.html', {'test_cases': get_paginator(request, test_cases)})
121
122
123 # 用例集合菜单项
124 @login_required
125 def case_suite(request):
126 if request.method == "POST":
127 count_down_time = 0
128 if request.POST['delay_time']:
129 print("输入的延迟时间是: {}".format(request.POST['delay_time']))
130 try:
131 count_down_time = int(request.POST['delay_time'])
132 except:
133 print("输入的延迟时间是非数字！")
134 else:
135 print("没有输入延迟时间")
136 env = request.POST.getlist('env')
137 print("env: {}".format(env))
138 server_address = get_server_address(env)
139 if not server_address:
140 return HttpResponse("提交的运行环境为空，请选择环境后再提交！")
141 case_suite_list = request.POST.getlist('case_suite_list')
142 if case_suite_list:
143 print("所需执行的用例集合列表：", case_suite_list)
144 for suite_id in case_suite_list:
145 test_suite = models.CaseSuite.objects.get(id=int(suite_id))
146 print("所需执行的用例集合: {}".format(test_suite))
147 username = request.user.username
148 test_suite_record = models.CaseSuiteExecuteRecord.objects.create(case_suite=test_suite,
149 run_time_interval=count_down_time,
150 creator=username)
151 # 普通执行
152 # suite_task(test_suite_record, test_suite, server_address)
153 # celery 执行：countdown表示任务延迟的时间，该参数值可以从前端表单中传递过来
154 suite_task.apply_async((test_suite_record, test_suite, server_address), countdown=count_down_time)
155 else:
156 print("运行测试集合用例失败")
157 return HttpResponse("运行的测试集合为空，请选择测试集合后再运行！")
158 case_suites = models.CaseSuite.objects.filter()
159 return render(request, 'case_suite.html', {'case_suites': get_paginator(request, case_suites)})
160
161
162 # 用例集合-添加测试用例页
163 @login_required
164 def add_case_in_suite(request, suite_id):
165 # 查询指定的用例集合
166 case_suite = models.CaseSuite.objects.get(id=suite_id)
167 # 根据id号查询所有的用例
168 test_cases = models.TestCase.objects.filter().order_by('id')
169 if request.method == "GET":
170 print("test cases:", test_cases)
171 elif request.method == "POST":
172 test_cases_list = request.POST.getlist('testcases_list')
173 # 如果页面勾选了用例
174 if test_cases_list:
175 print("勾选用例id：", test_cases_list)
176 # 根据页面勾选的用例与查询出的所有用例一一比较
177 for test_case in test_cases_list:
178 test_case = models.TestCase.objects.get(id=int(test_case))
179 # 匹配成功则添加用例
180 models.SuiteCase.objects.create(case_suite=case_suite, test_case=test_case)
181 # 未勾选用例
182 else:
183 print("添加测试用例失败")
184 return HttpResponse("添加的测试用例为空，请选择用例后再添加！")
185 return render(request, 'add_case_in_suite.html',
186 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
187
188
189 # 用例集合页-查看/删除用例
190 @login_required
191 def show_and_delete_case_in_suite(request, suite_id):
192 case_suite = models.CaseSuite.objects.get(id=suite_id)
193 test_cases = models.SuiteCase.objects.filter(case_suite=case_suite)
194 if request.method == "POST":
195 test_cases_list = request.POST.getlist('test_cases_list')
196 if test_cases_list:
197 print("勾选用例：", test_cases_list)
198 for test_case in test_cases_list:
199 test_case = models.TestCase.objects.get(id=int(test_case))
200 models.SuiteCase.objects.filter(case_suite=case_suite, test_case=test_case).first().delete()
201 else:
202 print("测试用例删除失败")
203 return HttpResponse("所选测试用例为空，请选择用例后再进行删除！")
204 case_suite = models.CaseSuite.objects.get(id=suite_id)
205 return render(request, 'show_and_delete_case_in_suite.html',
206 {'test_cases': get_paginator(request, test_cases), 'case_suite': case_suite})
207
208
209 # 用例执行结果-菜单项
210 @login_required
211 def test_case_execute_record(request):
212 test_case_execute_records = models.TestCaseExecuteResult.objects.filter().order_by('-id')
213 return render(request, 'test_case_execute_records.html', {'test_case_execute_records':
214 get_paginator(request, test_case_execute_records)})
215
216
217 # 用例执行结果-对比差异
218 @login_required
219 def case_result_diff(request, test_record_id):
220 test_record_data = models.TestCaseExecuteResult.objects.get(id=test_record_id)
221 print("用例执行结果记录: {}".format(test_record_data))
222 present_response = test_record_data.response_data
223 if present_response:
224 present_response = json.dumps(json.loads(present_response), sort_keys=True, indent=4,
225 ensure_ascii=False) # 中文字符不转ascii编码
226 print("当前响应结果: {}".format(present_response))
227 last_time_execute_response = test_record_data.last_time_response_data
228 if last_time_execute_response:
229 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True, indent=4,
230 ensure_ascii=False)
231 print("上一次响应结果: {}".format(last_time_execute_response))
232 return render(request, 'case_result_diff.html', locals())
233
234
235 # 用例执行结果-异常信息展示
236 @login_required
237 def show_exception(request, execute_id):
238 test_record = models.TestCaseExecuteResult.objects.get(id=execute_id)
239 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
240
241
242 # 用例集合执行结果
243 @login_required
244 def case_suite_execute_record(request):
245 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.filter().order_by('-id')
246 return render(request, 'case_suite_execute_record.html',
247 {'case_suite_execute_records': get_paginator(request, case_suite_execute_record)})
248
249
250 # 用例集合执行结果-包含用例结果展示
251 @login_required
252 def suite_case_execute_record(request, suite_record_id):
253 case_suite_execute_record = models.CaseSuiteExecuteRecord.objects.get(id=suite_record_id)
254 suite_case_execute_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=case_suite_execute_record)
255 return render(request, 'suite_case_execute_record.html',
256 {'suite_case_execute_records': get_paginator(request, suite_case_execute_records)})
257
258
259 # 用例集合执行结果-包含用例结果展示-差异比对
260 @login_required
261 def suite_case_result_diff(request, suite_case_record_id):
262 suite_record_data = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
263 present_response = suite_record_data.response_data
264 if present_response:
265 present_response = json.dumps(json.loads(present_response),sort_keys=True, indent=4, ensure_ascii=False)
266 print("当前响应: {}".format(present_response))
267 last_time_execute_response = suite_record_data.last_time_response_data
268 if last_time_execute_response:
269 last_time_execute_response = json.dumps(json.loads(last_time_execute_response), sort_keys=True,
270 indent=4, ensure_ascii=False)
271 print("上一次响应: {}".format(last_time_execute_response))
272 return render(request, 'case_result_diff.html', locals())
273
274
275 # 用例集合执行结果-包含用例结果展示-异常信息展示
276 @login_required
277 def suite_case_exception(request, suite_case_record_id):
278 test_record = models.CaseSuiteTestCaseExecuteRecord.objects.get(id=suite_case_record_id)
279 return render(request, 'show_exception.html', {'exception_info': test_record.exception_info})
280
281
282 # 用例集合执行结果单次统计
283 def suite_case_statistics(request, suite_id):
284 success_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="成功"))
285 fail_num = len(models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id, execute_result="失败"))
286 suite_case_records = models.CaseSuiteTestCaseExecuteRecord.objects.filter(case_suite_record=suite_id).order_by('-id')
287 return render(request, 'suite_case_statistics.html',
288 {'suite_case_records': get_paginator(request, suite_case_records), 'success_num': success_num,
289 'fail_num': fail_num})
290
291
292 # 用例集合执行结果历史统计
293 def case_suite_statistics(request, suite_id):
294 case_suite = models.CaseSuite.objects.get(id=suite_id)
295 success_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="成功"))
296 fail_num = len(models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite, test_result="失败"))
297 case_suite_records = models.CaseSuiteExecuteRecord.objects.filter(case_suite=case_suite).order_by('-id')
298 return render(request, 'case_suite_statistics.html',
299 {'case_suite_records': get_paginator(request, case_suite_records), 'success_num': success_num,
300 'fail_num': fail_num})
301
302
303 # 模块测试结果统计
304 @login_required
305 def module_statistics(request, module_id):
306 test_module = models.Module.objects.get(id=int(module_id))
307 test_cases = models.TestCase.objects.filter(belong_module=test_module)
308 test_suit_success_num = len(
309 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="成功"))
310 test_suit_fail_num = len(
311 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="失败"))
312 test_case_success_num = len(
313 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="成功"))
314 test_case_fail_num = len(
315 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="失败"))
316 success_num = test_suit_success_num + test_case_success_num
317 fail_num = test_suit_fail_num + test_case_fail_num
318 return render(request, 'module_statistics.html',
319 {'test_module': test_module, 'success_num': success_num, 'fail_num': fail_num})
320
321
322 # 项目测试结果统计
323 @login_required
324 def project_statistics(request, project_id):
325 test_project = models.Project.objects.get(id=int(project_id))
326 test_cases = models.TestCase.objects.filter(belong_project=test_project)
327 test_suit_success_num = len(
328 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="成功"))
329 test_suit_fail_num = len(
330 models.CaseSuiteTestCaseExecuteRecord.objects.filter(test_case__in=test_cases, execute_result="失败"))
331 test_case_success_num = len(
332 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="成功"))
333 test_case_fail_num = len(
334 models.TestCaseExecuteResult.objects.filter(belong_test_case__in=test_cases, execute_result="失败"))
335 success_num = test_suit_success_num + test_case_success_num
336 fail_num = test_suit_fail_num + test_case_fail_num
337 return render(request, 'project_statistics.html',
338 {'test_project': test_project, 'success_num': success_num, 'fail_num': fail_num})
339
340
341 # 默认页的视图函数
342 @login_required
343 def index(request):
344 return render(request, 'index.html')
345
346
347 # 登录页的视图函数
348 def login(request):
349 print("request.session.items(): {}".format(request.session.items())) # 打印session信息
350 if request.session.get('is_login', None):
351 return redirect('/')
352 # 如果是表单提交行为，则进行登录校验
353 if request.method == "POST":
354 login_form = UserForm(request.POST)
355 message = "请检查填写的内容！"
356 if login_form.is_valid():
357 username = login_form.cleaned_data['username']
358 password = login_form.cleaned_data['password']
359 try:
360 # 使用django提供的身份验证功能
361 user = auth.authenticate(username=username, password=password) # 从auth_user表中匹配信息，匹配到则返回用户对象
362 if user is not None:
363 print("用户【%s】登录成功" % username)
364 auth.login(request, user)
365 request.session['is_login'] = True
366 # 登录成功，跳转主页
367 return redirect('/')
368 else:
369 message = "用户名不存在或者密码不正确！"
370 except:
371 traceback.print_exc()
372 message = "登录程序出现异常"
373 # 用户名或密码为空，返回登录页和错误提示信息
374 else:
375 return render(request, 'login.html', locals())
376 # 不是表单提交，代表只是访问登录页
377 else:
378 login_form = UserForm()
379 return render(request, 'login.html', locals())
380
381
382 # 注册页的视图函数
383 def register(request):
384 return render(request, 'register.html')
385
386
387 # 登出的视图函数：重定向至login视图函数
388 @login_required
389 def logout(request):
390 auth.logout(request)
391 request.session.flush()
392 return redirect("/login/")