程序化交易券商接口视频-2019 Python接口自动化测试框架实战开发一

说明：该篇博客是博主一字一码编写的，实属不易，请尊重原创，谢谢大家！

项目源码下载

一丶叙述

项目介绍

整个项目分为四个部分：接口基础丶接口开发丶Unittest与接口测试结合以及接口自动化框架从设计到开发

接口基础包括：HTTP接口/常见接口/接口工具/接口基础知识

接口开发：通过Django来开发get/post接口

Unittest与接口测试结合：unittest应用/断言/requests引入/HTMLTestRunner/case的管理

接口自动化框架从设计到开发：如何设计框架/封装工具类/重构基类/错误调试/结果收集以及处理/解决数据依赖/结果统计及报告发送

项目整体思路：通过对接口数据文档的读写操作，来获取文档中case的所有数据，然后通过requests模块来发送请求获取的响应数据，通过返回的响应数据中的某个标志性字段的值来判断是否测试成功或者失败，最后将测试的结果数据写入到测试文档或者是html页面又或者是将结果以邮件的形式发送到指定邮箱，这是整个大框架思路，要完成这一系列自动化的测试框架，则需要有一定的python代码基础，博主这里只是粗略的叙述了思路，有很多地方就不细说了比如数据依赖等就请大家慢慢的阅读吧

测试报告效果预览

unittest和HTMLTestRunner结合生成报告

unittest和HTMLTestRunner结合生成报告

测试报告邮件通知

二丶接口基础知识

什么是接口

连接前后端以及移动端，通俗来说就是前端和后端之间的桥梁，比如网站需要去调用银行丶微信及支付宝的接口来完成业务需求

接口的种类

外部接口和内部接口；内部接口又分为上层服务与下层服务以及同级服务

接口的分类

请求方式：post丶get丶delete丶put

为什么要做接口测试

原因：不同端的工作进度肯定是不一致的，那么就需要对最开始开发出来的接口进行测试；对于项目来说缩短项目周期，提高开发效率以及提高系统的健壮性

接口测试流程

需求讨论——需求评审——场景设计——用例设计——数据准备——执行

为什么要设计测试用例

理清思路，避免侧漏提高测试效率跟进测试进度告诉领导做过跟进重复重复性工作

用例设计分类

功能用例测试：测试功能是否正常丶测试功能是否按照接口文档实现

逻辑用例设计：是否存在依赖业务，例如有些操作是需要用户登录成功的状态下才能进行的操作

异常测试用例设计：参数异常以及数据异常；参数异常包括关键字参数丶参数为空丶多参数丶少参数丶错误参数，数据异常包括关键字数据丶数据为空丶长度不一致丶错误数据

安全测试用例设计：cookie丶header丶唯一识别码

三丶接口测试工具

接口测试工具分类

抓取接口工具

httpwatch：集成于IE和Firefox浏览器中，在其他浏览器中无法使用，查看数据也比较麻烦

wireshark：只要是经过电脑的所有请求都会去抓取，导致数据量比较庞大，看数据也比较麻烦

fiddler：轻量级抓包工具，功能比较全，只会记录http请求不会像wireshark工具记录tcp和udp等请求

测试接口工具：

loadrunner：不仅仅是性能测试工具，由于该工具几乎都是基于http请求，所以也可以用来测试接口

fiddler：它除了可以抓包还可以向接口发送各种请求

soapui：接口和自动化测试工具，功能也比较强大

jmeter：跟loadrunner一样不仅仅是做性能测试，也可以对接口进行测试

postman：谷歌旗下的接口测试工具

四丶Fiddler的使用

抓取不同类型接口数据

查看windows本机的IP

配置fiddler

需要保证要抓取的手机与电脑保持同一网段，博主这里使用逍遥模拟器模拟安卓手机，修改手机网络

在高级选项中设置手动代理IP为windows本机IP地址，端口设置与fiddler抓取端口保持一致

再安卓手机中打开知乎app，抓取知乎app的http服务的数据

现在的移动app都是基于https请求的，所以需要在fiddler中设置https请求

然后在手机端浏览器中访问windows电脑IP+port，进行网络安全证书的下载安装

点击下面一个下载证书

然后设置密码即可

证书安装成功后，重新打开知乎app，则成功抓取https请求的数据

数据模拟以及过滤规则

如下进行选择要过滤的hosts类型，并在输入框添加要过滤的hosts即可

对知乎上的一篇文章进行回答后，获取https://apzhihu.com/answers接口，查看发送的post请求数据中的content字段内容也就是博主回答的内容

然后进行数据模拟，也就是点击fiddler软件上的replay对https://apzhihu.com/answers接口进行post请求数据的而二次发送，由于知乎这边设定对一个问题只能进行一次回答，所以知乎服务器返回的json数据提示我们失败，同时也说明对接口进行二次数据发送成功，只是规则逻辑失败

如何模拟接口响应数据

首先第一步，访问知乎app热榜，在fiddler软件中获取接口查看服务器响应的json格式数据，从服务器返回的json数据看出热榜标题字段名为title_area

然后选择服务器返回的数据类型为TextView，点击.ViewinNotepad即打开数据记事本，如下在记事本中找到title_area字段的内容，该字段内容进行了将中文转换为一串字符串

将记事本中的title_area字段的数据修改为thisisatestforcdtaogang

点击文件——另存为保存到桌面

回到fiddler中，左侧选中热榜接口，右侧选中AutoResponder，在此窗口下点击AddRule将左侧的接口添加进去，在右侧下方导入保存在桌面的zhihu_hot.htm文件，最后点击sava保存

回到知乎app中刷新当前热榜页面，则成功返回修改的热榜标题

请求头数据

请求体数据

将上面复制的请求头和请求体数据分别粘贴到如下输入框中，点击Execute执行发送，然后在左侧则出现了另一个comments接口数据

查看该comments接口，服务器返回的响应数据中与第一个comments接口一致，说明接口测试成功

五丶unittest使用

unittest简单使用

在IDE中使用python的环境随便创建个py文件，需要注意的是该py文件的名字不能是test.py，否在运行时会出错，unittest包是python自带的不需要下载安装，代码如下

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/17 13:10'

import unittest


class TestMethod(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        print('Method before class execution')

    @classmethod
    def tearDownClass(cls):
        print('Method after class execution')

    def setUp(self):
        print('------setUp------')

    def tearDown(self):
        print('------tearDown------')

    def test_01(self):
        print('First test method')

    def test_02(self):
        print('The second test method')


if __name__ == '__main__':
    unittest.main()

直接run运行以上代码

unittest和request重构封装

说明：使用requests模块对接口url地址发送请求，通过unittest测试框架进行case测试

首先博主在逍遥安卓模拟器中下载了一个看书app，通过fiddler对app上的某一接口进行获取，之所以选择对此app进行接口测试，是因为该app的所有接口全是POST请求

在demo.py文件中，使用requestsget以及post方法进行了封装，主要是根据传递的参数method来对get以及post方法进行分别调用而已，具体实现如下

import requests


class RunMain:
	def send_get(self,url,data):
		res = requests.get(url=url,data=data).json()
		return res
		
	def send_post(self,url,data):
		res = requests.post(url=url,data=data).json()
		return res

	def run_main(self,url,method,data=None):
		res = None
		if method == 'GET':
			res = self.send_get(url,data)
		else:
			res = self.send_post(url,data)
		return res

在test_methopy文件中则创建测试类以及test方法，在test方法中调用demo.py中的run_main方法，即使用requests模块向传递的接口url地址和请求方式以及请求体发送对应的请求，这里使用setUp方法则是利用其优先调用而对RunMain类进行实例化

import unittest
import json
import HtmlTestRunner
from .demo import RunMain


class TestMethod(unittest.TestCase):
    def setUp(self):
        self.run = RunMain()

    def test_01(self):
        url = 'http://api.ishugui.com/asg/portal/call/265.do'
        data = {
            'sstoken':'eyJleHAiOjE1Njg1MzgyNTczMzUsImlhdCI6MTU2MDc2MjI1NzMzNSwicHAiOiIxMTQwNTQ1Njg5MDYwMDQ0ODAwQHNvaHUuY29tIiwidGsiOiIwZkNYSHpjTUZzR0dFMEswbVdvUVFCNWVCanpXa0hmWiIsInYiOjB9.SDYkT9FpWrBbko6xRrESN74IXJhzkqQLtijKjGiVrqA',
            'gidinf':'x011060802ff0fd40695d68140002799751474c540b3',
            'ppinf':'2|1560762257|1561971857|bG9naW5pZDowOnx1c2VyaWQ6Mjg6MTE0MDU0NTY4OTA2MDA0NDgwMEBzb2h1LmNvbXxzZXJ2aWNldXNlOjMwOjAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMHxjcnQ6MTA6MjAxOS0wNi0xN3xlbXQ6MTowfGFwcGlkOjY6MTEwNjA4fHRydXN0OjE6MXxwYXJ0bmVyaWQ6MTowfHJlbGF0aW9uOjA6fHV1aWQ6MTY6czk1YWIwNDk5NjE3YmJhNnx1aWQ6MTY6czk1YWIwNDk5NjE3YmJhNnx1bmlxbmFtZTowOnw',
            'pprdig':'kaKPdU0WwIdzL58CqxNz5pgMyv23P0-Y5GRnd5ufPlXIGzrk7_7TlIK5XFQiuoqAHNqGVXHCVd4cB1DIkR5yFZ_nExnSjIZbBJWYlMkrsiIjDYqWCvedZRLm8sZqS0WqA0FcKXuSn3Z0gVRus9YpEonNz5wyuWdUqxaSmzlzygY',
            'ppsmu':'1|1560762257|1561971857|dXNlcmlkOjI4OjExNDA1NDU2ODkwNjAwNDQ4MDBAc29odS5jb218dWlkOjA6fHV1aWQ6MDo|byWcaoPqy02s2_9GHLhZFAQ6Ov_GazMPFLrq115HiSTBS9Ijr33a55quRq2Mr1_6ZMruKEk-BYFpShUaMtwRYA'
        }
        res1 = self.run.run_main(url, 'POST', json.dumps(data))
        print(res1)


    def test_02(self):
        url = 'http://api.ishugui.com/asg/portal/call/265.do'
        data = {

        }
        res2 = self.run.run_main(url, 'POST', data)

        print(res2)


if __name__ == '__main__':
    unittest.main()

运行test_method模块，查看测试接口，test_02则是错误测试

unittest中assert的使用

首先根据返回的结果字典dict数据中的status状态值来判断测试是否通过或者失败，逻辑很基础就不细说了

class TestMethod(unittest.TestCase):
    def setUp(self):
        self.run = RunMain()

    def test_01(self):
        url = 'http://api.ishugui.com/asg/portal/call/265.do'
        data = {
            'sstoken':'eyJleHAiOjE1Njg1MzgyNTczMzUsImlhdCI6MTU2MDc2MjI1NzMzNSwicHAiOiIxMTQwNTQ1Njg5MDYwMDQ0ODAwQHNvaHUuY29tIiwidGsiOiIwZkNYSHpjTUZzR0dFMEswbVdvUVFCNWVCanpXa0hmWiIsInYiOjB9.SDYkT9FpWrBbko6xRrESN74IXJhzkqQLtijKjGiVrqA',
            'gidinf':'x011060802ff0fd40695d68140002799751474c540b3',
            'ppinf':'2|1560762257|1561971857|bG9naW5pZDowOnx1c2VyaWQ6Mjg6MTE0MDU0NTY4OTA2MDA0NDgwMEBzb2h1LmNvbXxzZXJ2aWNldXNlOjMwOjAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMHxjcnQ6MTA6MjAxOS0wNi0xN3xlbXQ6MTowfGFwcGlkOjY6MTEwNjA4fHRydXN0OjE6MXxwYXJ0bmVyaWQ6MTowfHJlbGF0aW9uOjA6fHV1aWQ6MTY6czk1YWIwNDk5NjE3YmJhNnx1aWQ6MTY6czk1YWIwNDk5NjE3YmJhNnx1bmlxbmFtZTowOnw',
            'pprdig':'kaKPdU0WwIdzL58CqxNz5pgMyv23P0-Y5GRnd5ufPlXIGzrk7_7TlIK5XFQiuoqAHNqGVXHCVd4cB1DIkR5yFZ_nExnSjIZbBJWYlMkrsiIjDYqWCvedZRLm8sZqS0WqA0FcKXuSn3Z0gVRus9YpEonNz5wyuWdUqxaSmzlzygY',
            'ppsmu':'1|1560762257|1561971857|dXNlcmlkOjI4OjExNDA1NDU2ODkwNjAwNDQ4MDBAc29odS5jb218dWlkOjA6fHV1aWQ6MDo|byWcaoPqy02s2_9GHLhZFAQ6Ov_GazMPFLrq115HiSTBS9Ijr33a55quRq2Mr1_6ZMruKEk-BYFpShUaMtwRYA'
        }
        res1 = self.run.run_main(url, 'POST', json.dumps(data))
        # print(type(res1))
        # print(res1['pub'])
        # print(type(res1['pub']))
        if res1['pub']['status'] == 0:
            print('测试通过')
        else:
            print('测试失败')
        print(res1)


    def test_02(self):
        url = 'http://api.ishugui.com/asg/portal/call/265.do'
        data = {

        }
        res2 = self.run.run_main(url, 'POST', data)
        if res2['pub']['status'] == 0:
            print('测试通过')
        else:
            print('测试失败')
        print(res2)


if __name__ == '__main__':
    unittest.main()

运行以上代码，查看结果与预期一样

将if判断代码更换成unittest模块中的assert断言进行判断，这里使用assertEqual方法来判断两个值是否相等，当两个值相等则返回OK，当不相同时返回assertEqual方法msg变量自定义的值

class TestMethod(unittest.TestCase):
    def setUp(self):
        self.run = RunMain()

    def test_01(self):
        url = 'http://api.ishugui.com/asg/portal/call/265.do'
        data = {
            'sstoken':'eyJleHAiOjE1Njg1MzgyNTczMzUsImlhdCI6MTU2MDc2MjI1NzMzNSwicHAiOiIxMTQwNTQ1Njg5MDYwMDQ0ODAwQHNvaHUuY29tIiwidGsiOiIwZkNYSHpjTUZzR0dFMEswbVdvUVFCNWVCanpXa0hmWiIsInYiOjB9.SDYkT9FpWrBbko6xRrESN74IXJhzkqQLtijKjGiVrqA',
            'gidinf':'x011060802ff0fd40695d68140002799751474c540b3',
            'ppinf':'2|1560762257|1561971857|bG9naW5pZDowOnx1c2VyaWQ6Mjg6MTE0MDU0NTY4OTA2MDA0NDgwMEBzb2h1LmNvbXxzZXJ2aWNldXNlOjMwOjAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMHxjcnQ6MTA6MjAxOS0wNi0xN3xlbXQ6MTowfGFwcGlkOjY6MTEwNjA4fHRydXN0OjE6MXxwYXJ0bmVyaWQ6MTowfHJlbGF0aW9uOjA6fHV1aWQ6MTY6czk1YWIwNDk5NjE3YmJhNnx1aWQ6MTY6czk1YWIwNDk5NjE3YmJhNnx1bmlxbmFtZTowOnw',
            'pprdig':'kaKPdU0WwIdzL58CqxNz5pgMyv23P0-Y5GRnd5ufPlXIGzrk7_7TlIK5XFQiuoqAHNqGVXHCVd4cB1DIkR5yFZ_nExnSjIZbBJWYlMkrsiIjDYqWCvedZRLm8sZqS0WqA0FcKXuSn3Z0gVRus9YpEonNz5wyuWdUqxaSmzlzygY',
            'ppsmu':'1|1560762257|1561971857|dXNlcmlkOjI4OjExNDA1NDU2ODkwNjAwNDQ4MDBAc29odS5jb218dWlkOjA6fHV1aWQ6MDo|byWcaoPqy02s2_9GHLhZFAQ6Ov_GazMPFLrq115HiSTBS9Ijr33a55quRq2Mr1_6ZMruKEk-BYFpShUaMtwRYA'
        }
        res1 = self.run.run_main(url, 'POST', json.dumps(data))
        # print(type(res1))
        # print(res1['pub'])
        # print(type(res1['pub']))
        # if res1['pub']['status'] == 0:
        #     print('测试通过')
        # else:
        #     print('测试失败')
        self.assertEqual(res1['pub']['status'], 0, '测试失败')
        print(res1)


    def test_02(self):
        url = 'http://api.ishugui.com/asg/portal/call/265.do'
        data = {

        }
        res2 = self.run.run_main(url, 'POST', data)
        # if res2['pub']['status'] == 0:
        #     print('测试通过')
        # else:
        #     print('测试失败')
        self.assertEqual(res2['pub']['status'], 0, '测试失败')
        print(res2)


if __name__ == '__main__':
    unittest.main()

测试查看结果，断言失败，测试结果如下很清晰

unittest中case的管理及运用

在测试一些接口时，有些接口的返回数据需要在下一个接口进行使用，所以需要定义全局变量，方便每个case都能够得着，当在test_01中定义全局变量userid，然后在test_02中进行打印

在unittest中，是按照字母数字来进行case先后执行顺序的，将test_01改为test_03后，运行代码后，会提示test_02中的userid未定义，原因是程序先去执行了test_02这个case，所以出现该提示是正常的

当在测试代码中有很多case时，我想跳过某个case，则在该case方法上定义unittest的skip方法装饰器，并需要传递此方法名作为实参进行传递

除了在if__name__=='__main__'中使用unittest.main方法执行所有的case以外，还可以将要测试的case添加到unittest.TestSuite集合中执行想要执行的case，若想要全部都执行则需要一个一个的添加

unittest和HTMLTestRunner结合生成报告

第一种：比较新版本的htmltestrunner报告

安装HTMLTestRunner，在html-testRunner·PyPI网站上进行下载whl文件

安装成功后，即在以下路径中可以找到安装的HTMLTestRunner的包了

在if__name__=='__main__'中只需要调用HtmlTestRunner模块中的HtmlTestRunner类，向该类传递报告标题参数值即可，其他均默认，需要注意的时启动文件run为当前的py文件，如果是Unittests开头的启动文件，则不会运行if__name__=='__main__'下的代码，只会执行unittest框架的setUp以及test开头的case代码

第二种：比较经典版本的htmltestrunner报告

紧接着到http://tungwaiyip.info/software/HTMLTestRunner_0_8_2/HTMLTestRunner.py地址中将HTMLTestRunner.py的代码全选拷贝，然后粘贴到base2/HTMLTestRunner.py文件中，因为该版本的HTMLTestRunner代码是以py2进行编写的，博主的环境使用的是py所以需要进行修改，修改的内容如下

第94行， 将import StringIO修改成import io
第539行，将self.outputBuffer = StringIO.StringIO()修改成self.outputBuffer = io.StringIO()
第642行，将if not rmap.has_key(cls):修改成if not cls in rmap:
第631行，将print >> sys.stderr, '
Time Elapsed: %s' % (self.stopTime-self.startTime)修改成print(sys.stderr, '
Time Elapsed: %s' % (self.stopTime-self.startTime))
第766行，将uo = o.decode('latin-1')修改成uo = e
第775行，将ue = e.decode('latin-1')修改成ue = e

在test_method2模块中首先需要从base2模块中去导入HTMLTestRunner文件，然后if__name__=='__main__'中，需要创建一个文件源，同样是调用HTMLTestRunner模块中的HTMLTestRunner类，不同的是需要将创建的文件源传递给实例属性stream变量

打开html_report.html测试报告，测试结果与代码设定一致

六丶mock服务入门到实战

mock简介

mock测试就是在测试过程中，对于某些不容易构造或者不容易获取的对象，用一个虚拟的对象来创建以便测试的测试方法，mock是在测试过程中，对于一些不容易构造/获取的对象，创建一个mock对象来模拟对象的行为即就是模拟fiddler返回接口响应数据的一个过程。

mock安装

在终端使用pip进行安装即可

在case中通过底层函数实现mock

在test_method模块中导入mock，然后在test_03函数中通过以下代码设置返回的return_value的值为请求的data数据

将调用run_main方法的值设定为mock_data，即print(res则表示打印请求的data数据的值，因为res1的数据不再是接口返回的响应数据，则arrest断言是会提示报错的，这是正常的

重构封装mock服务

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/20 16:26'
from mock import mock
import json

def mock_test(mock_method, request_data, url, method, response_data):
    '''
    :param mock_method:
    :param request_data:
    :param url:
    :param method:
    :param response_data:
    :return: res
    '''
    mock_method = mock.Mock(return_value=response_data)
    print('mock_method:', mock_method)
    res = mock_method(url, method, json.dumps(request_data))
    return res

那么在test_03方法中，如下进行调用即可

res1 = mock_test(self.run.run_main, data, url, 'POST', 'ssssssss')
print('res1:', res1)

运行Unittestsintest_methopy，查看运行结果和博主设定一样成功返回自定义的response_data数据

七丶接口自动化框架设计到开发

如何设计一个接口自动化测试框架

根据接口地址丶接口类型丶请求数据丶预期结果来进行设计，对于需要登录后才能进行操作的接口那么则需要进行headercookie等数据的传递，自动化测试的难点就是数据依赖。

python操作excel获得内容

首先python操作excel，需要安装两个包，分别是xlrd和xlwt这两个库，xlrd这个库是负责读取excel数据的，而xlwt库是负责向excel写入数据的

重构操作excel函数

根据上一步骤读取excel表的内容代码后，进行了一个简单的封装，提高代码的通用性，过程相当的简单，实现代码如下

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/20 17:33'
import xlrd

data = xlrd.open_workbook('../test_data/rs.xls')
tables = data.sheets()[0]  # 获取表格数据对象
print(tables.nrows) # 打印表格行数
print(tables.cell_value(0,0))  # 打印excel表格数据，需要传递数据所在的坐标(x,y)
print(tables.cell_value(0,1))
print('*'*50+'封装前后数据对比'+'*'*50)


class operationExcel(object):
    def __init__(self, file_path='../test_data/rs.xls', sheet_id=0):
        self.file_path = file_path
        self.sheet_id = sheet_id
        self.data = self.get_data()

    def get_data(self):
        data = xlrd.open_workbook(self.file_path)
        tables = data.sheets()[self.sheet_id]
        return tables

    def get_rows(self):
        '''获取单元格的排数'''
        return self.data.nrows

    def get_cell_value(self, x=0, y=0):
        '''获取某个单元格的数据'''
        return self.data.cell_value(x, y)


if __name__ == '__main__':
    print(operationExcel().get_rows())
    print(operationExcel().get_cell_value())
    print(operationExcel().get_cell_value(0,1))

运行op_excepy文件后，结果与封装之前代码结果一致，表示重构封装代码成功

学习操作json文件

在utils工具包下创建op_jsopy文件，在文件中对logijson文件内容进行读取操作，代码如下

重构json工具类

将上一步操作json的代码进行封装

class operationJson(object):
    def __init__(self, file_path='../test_data/login.json'):
        self.file_path = file_path
        self.data = self.get_data()

    def get_data(self):
        with open(self.file_path) as f:
            data = json.load(f)
            return data

    def get_key_words(self, key=None):
        if key:
            return self.data[key]
        else:
            return self.data


if __name__ == '__main__':
    print(operationJson().get_key_words())
    print(operationJson().get_key_words('login'))
    print(operationJson().get_key_words('login')['username'])

运行op_jsopy文件，结果与封装之前代码结果一致，表示重构封装代码成功

封装获取常量方法

首先打开excel表格，查看需要获取的字段有哪些

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/21 9:29'


class global_var:
    id = '0'  # id
    module = '1'  # 模块
    url = '2'  # url
    run = '3'  # 是否运行
    request_type = '4'  # 请求类型
    request_header = '5'  # 是否携带header
    case_depend = '6'  # case依赖
    response_data_depend = '7'  # 依赖的返回数据
    data_depend = '8'  #  数据依赖
    request_data = '9'  # 请求数据
    expect_result = '10'  # 预期结果
    reality_result = '11'  # 实际结果


def get_id():
    return global_var.id

def get_module():
    return global_var.module

def get_url():
    return global_var.url

def get_run():
    return global_var.run

def get_request_type():
    return global_var.request_type

def get_request_header():
    return global_var.request_header

def get_case_depend():
    return global_var.case_depend

def get_response_data_depend():
    return global_var.response_data_depend

def get_data_depend():
    return global_var.data_depend

def get_request_data():
    return global_var.request_data

def get_expect_result():
    return global_var.expect_result

def get_reality_result():
    return global_var.reality_result

封装获取接口数据

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/21 10:01'
from utils.op_excel import operationExcel
from utils.op_json import operationJson
from data import data_conf

class getData(object):
    def __init__(self):
        self.op_excel = operationExcel()

    def get_case_lines(self):
        '''获取表格行数'''
        return self.op_excel.get_rows()

    def get_is_run(self, x):
        '''获取case是否运行'''
        flag = None
        y = data_conf.get_run()
        run_value = self.op_excel.get_cell_value(x, y)
        if run_value == 'yes':
            flag = True
        else:
            flag = False
        return flag

    def get_is_header(self, x):
        '''是否携带header'''
        y = data_conf.get_request_header()
        header = self.op_excel.get_cell_value(x, y)
        if header == 'yes':
            return data_conf.get_header_value()
        else:
            return None

    def get_request_method(self, x):
        '''获取请求方式'''
        y = data_conf.get_request_type()
        request_method = self.op_excel.get_cell_value(x, y)
        return request_method

    def get_request_url(self, x):
        '''获取请求地址'''
        y = data_conf.get_url()
        request_url = self.op_excel.get_cell_value(x, y)
        return request_url

    def get_request_data(self, x):
        '''获取请求数据'''
        y = data_conf.get_request_data()
        request_data = self.op_excel.get_cell_value(x, y)
        if request_data == '':
            return None
        return request_data

    def get_data_for_json(self, x):
        '''通过excel中的关键字去获取json数据'''
        op_json = operationJson()
        data = op_json.get_key_words(self.get_request_data(x))
        return data

    def get_expect_data(self, x):
        '''获取预期结果数据'''
        y = data_conf.get_expect_result()
        expect_data = self.op_excel.get_cell_value(x, y)
        if expect_data == '':
            return None
        return expect_data

post、get基类的封装

在base包下创建run_methopy文件，在文件中重新编写对get丶post请求方式的代码封装，具体如下

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/21 11:19'
import requests


class RunMain(object):

    def get_main(self, url, data=None, header=None):
        res = None
        if header is not None:
            res = requests.get(url=url, data=data, headers=header).json()
        else:
            res = requests.get(url=url, data=data).json()
        return res

    def post_main(self, url, data, header=None):
        res = None
        if header is not None:
            res = requests.post(url=url, data=data, headers=header).json()
        else:
            res = requests.post(url=url, data=data).json()
        return res


    def run_main(self, url, method, data=None, header=None):
        res = None
        if method.lower() == 'post':
            res = self.post_main(url, data, header)
        elif method.lower() == 'get':
            res = self.get_main(url, data, header)
        else:
            return 'what ?????'
        return res

主流程封装及错误解决调试

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/21 11:57'
from base.run_method import RunMain
from data.data_get import getData


class RunTest(object):
    def __init__(self):
        self.runmain = RunMain()
        self.data = getData()

    def run(self):
        res = None
        row_counts = self.data.get_case_lines()  # 获取excel表格行数
        # print(row_counts) 5
        for row_count in range(1, row_counts):
            # print(row_count) 1,2,3,4
            url = self.data.get_request_url(row_count)  # y行不变遍历获取x列的请求地址
            method = self.data.get_request_method(row_count)  # y行不变遍历获取x列的请求方式
            is_run = self.data.get_is_run(row_count)  # y行不变遍历获取x列的是否运行
            data = self.data.get_data_for_json(row_count)  # y行不变遍历获取x列的请求数据，这里面时三次调用，依次分别是get_data_for_json丶get_key_words丶get_request_data
            header = self.data.get_is_header(row_count)
            print('url:', url)
            print('method:', method)
            print('is_run:', is_run)
            print('data:', data)
            print('header:', header)

            if is_run:
                res = self.runmain.run_main(url,method,data,header)
                print('*'*60+'分割线'+'*'*60)
        return res


if __name__ == '__main__':
    print('res:', RunTest().run())

运行run_test，成功的将excel以及json数据正确打印出来，返回res服务器返回结果，需要说明的是excel表中的所有数据都不是真实存在的，包括json文档数据也是，这里主要是测试整个框架的正确性读取excel以及json文档数据，并正确的发送请求获得相应数据

运行结果出现红色的内容，是由requests模块发送请求的安全请求警告，如不想显示此警告，可以在run_methopy发送请求核心代码进行禁用，禁用代码如下

重新运行run_test，安全请求警告不再显示

根据代码运行结果，对比excel表以及json数据文档内容，数据正确无误

0.返回数据格式处理以及调错

为了测试返回的接口的响应数据，博主这里在excel文档以及json文档中添加了一条数据

因为在excel文档中小说的接口不携带header所以在向接口发送请求数据核心代码块，进行了如下修改，因为在excel文档中的最后一个接口时真实的，所以只需要对最后一个接口url返回的字典类型的响应数据进行转换成json格式的数据，并按照关键字进行排序

运行run_test，在最后一个接口中成功打印出我们想要的数据

获取接口返回状态

在发送请求数据核心代码中，进行打印返回的状态码status_code即可，最后一个接口比较特殊，返回的响应数据中没有status_code，所以需要对返回的json数据中的status进行判断，并向其返回数据中添加我们所要的status_code的值

运行代码，当返回的状态码为404表示接口不存在，只要是存在响应数据，则status_code为200，必须说明一点的就是status_code为200不一定表示接口存在，有些大型网站会对其域名下不存在的接口返回200的错误页面显示，所以在测试文档中会体现预期结果和实际结果两项数据需要一致才能表示测试通过

通过预期结果判断case是否执行成功

进行接下来的测试，博主这里重新准备了另一个excel表来进行测试，需要对json文件中的数据进行添加，在excel表Book-05中的请求数据book5关键字对应的json文件的数据故意为空，可以对测试结果有一个对比

在数据获取核心类中定义了一个方法来获取excel表模块字段的数据

回过头在启动文件中获取模块名预期结果并进行打印

运行启动文件，查看运行结果

# -*- coding: utf-8 -*-
__author__ = 'cdtaogang'
__date__ = '2019/6/21 18:43'
import json

class CommonUtil(object):
    def is_contains(self, expect, reality):
        flag = None
        reality = json.loads(reality)
        expect = json.loads(expect)
        if expect['status'] == reality['pub']['status']:
            flag = True
        else:
            flag = False
        return flag

在启动文件中需要注释掉res响应数据，利于查看测试结果，还需要在启动文件调用is_contains方法来根据其返回值判断测试是否通过

运行启动文件，查看测试结果

将测试结果写入到excel中

首先在op_excepy中定义一个方法，该方法实现读取excel的数据，并进行copy复制，然后再write方法将数据写入到坐标位置

然后在data_get.py中需要定义一个方法，在该方法中核心逻辑为获取y坐标的值

最后在启动文件中，调用data_get模块中的write_reality_data方法，并将剩余的x坐标的值以及data数据传递给最终的核心方法write_reality_result_data来完成对excel表中的实际结果数据的写入

文章为作者独立观点，不代表股票程序化软件自动交易接口观点

程序化交易券商接口视频-接口和抽象类详解 接口继承 实现接口 抽象类继承

程序化交易券商接口视频-接口和抽象类详解接口继承实现接口抽象类继承