Python+Requests+Pytest接口自动化测试微信接口实例

[pytest]
log_cli=true
log_level=NOTSET
log_format = %(asctime)s %(levelname)s %(message)s
log_date_format = %Y-%m-%d %H:%M:%S

addopts = -vs --alluredir ./temp -m 'file'

log_file = ./log/test.log
log_file_level = info
log_file_format = %(asctime)s %(levelname)s %(message)s
log_file_date_format = %Y-%m-%d %H:%M:%S

testpaths = ./testcase
python_files = test_*.py
python_classes = Test*
python_functions = test_*
markers =
    file : test_create_flag

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2023-03-21 15:59
# @Author : hexiuxiu
# @File : all.py
'''运行所有用例'''
import os

import pytest
if __name__ == '__main__':
    pytest.main()
    os.system("allure generate temp -o reports --clean")

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2023-03-28 10:54
# @Author : hexiuxiu
# @File : assert_util.py
'''处理断言'''

from jsonpath import jsonpath

class AssertUtil:
    '''自定义断言'''
#     def assert__util(self,validate, res):
#         for i in validate:
#             if "eq" in i.keys():
#                 yaml_result = i.get("eq")[0]
#                 actual_result = jsonpath(res.json(), yaml_result)
#                 expect_result = i.get("eq")[1]
#                 # print(actual_result)
#                 print("实际结果：%s" % actual_result[0])
#                 print("期望结果：%s" % expect_result)
#
#                 assert actual_result[0] == expect_result

    '''另一个断言方法，更完善，更全面'''

    def assert_result(self, yaml_validate, result, res_status):
        flag_all = 0
        # 循环得到yaml文件的数据列表
        for y in yaml_validate:
            # 循环输出字典中的键值对
            for key, values in y.items():
                # 如果键等于equals
                if key == "equals":
                    flag = self.assert_equals(values, res_status, result)
                    flag_all = flag_all + flag
                elif key == "contains":
                    flag = self.assert_contains(values, result)
                    flag_all = flag_all + flag
                else:
                    print("该断言还未添加，暂不支持")
        assert flag_all == 0

    # 断言-等于
    def assert_equals(self, values, res_status, result):
        flag = 0
        for key, value in values.items():
            # 断言状态是否与预期一致
            if key == "status_code":
                if res_status != value:
                    flag = flag + 1
                    print(f"断言失败,预期状态码{value}与返回状态码{res_status}不一致")
            else:
                assert_value = jsonpath(result, f"$..{key}")
                if assert_value:
                    if value not in assert_value:
                        flag = flag + 1
                        print(f"断言失败,{key}不等于{value}")
                else:
                    flag = flag + 1
                    print(f"{key}不存在")
            return flag

    # 断言_包含
    def assert_contains(self, values, result):
        flag = 0
        # values=jsonpath(result,)
        if values not in str(result):
            flag = flag + 1
            print(f"断言失败{values}不存在{result}中")
        return flag

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2023-03-21 17:22
# @Author : hexiuxiu
# @File : yaml_util.py
'''读取yaml文件'''
import os
import random

import yaml

class YamlUtil():
    # 获取项目路径
    def get_progect_path(self):
        realpath = os.path.dirname(__file__).split('common')[0]
        return realpath

    # 获得随机数
    def get_randon_number(self, min, max):
        return random.randint(int(min), int(max))

    #读取extract.util文件
    def read_extract_yaml(self,key):
        with open(os.path.dirname(os.path.dirname(__file__))+'/data/extract.yaml',mode='r',encoding='utf-8') as f:
            value=yaml.load(stream=f,Loader=yaml.FullLoader)
            return value[key]

    #写入extract.util文件
    def write_extract_yaml(self,data):
        with open(os.path.dirname(os.path.dirname(__file__))+'/data/extract.yaml',mode='a',encoding='utf-8') as f:
            yaml.dump(data=data,stream=f,allow_unicode=True)

    #清除extract.util文件
    def clear_extract_yaml(self):
        with open(os.path.dirname(os.path.dirname(__file__))+'/data/extract.yaml',mode='w',encoding='utf-8') as f:
            f.truncate()

    #读取testcase文件
    def read_testcase_yaml(self,yaml_name):
        with open(os.path.dirname(os.path.dirname(__file__))+'/testcase/'+yaml_name,mode='r',encoding='utf-8') as f:
            value=yaml.load(stream=f,Loader=yaml.FullLoader)
            return value

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2023-03-21 15:42
# @Author : hexiuxiu
# @File : test_send_request.py
'''测试用例'''
import os

import requests
import pytest
from common import yaml_util,assert_util
from random import randint
import logging


class TestRendRequest():
    number_flag=randint(1000,9999)
    jsValue=''
    logger = logging.getLogger(__name__)

    @pytest.mark.file
    @pytest.mark.parametrize('caseinfo',yaml_util.YamlUtil().read_testcase_yaml('get_token.yaml'))
    def test_get_token(self,caseinfo):
        '''获取token值'''
        data = caseinfo['test']['request']['params']
        url = caseinfo['test']['request']['url']
        method=caseinfo['test']['request']['method']
        validate = caseinfo['test']['validate']
        rep = requests.request(method=method,url=url, params=data)
        result = rep.json()
        print(result)
        res_status = rep.status_code
        print(res_status)
        # print(result['access_token'])
        # print({'access_token':rep.json()['access_token']})

        if 'access_token' in result:
            yaml_util.YamlUtil().write_extract_yaml({'access_token':rep.json()['access_token']})
            # 断言
            # assert_util.AssertUtil().assert__util(validate,rep)
            assert_util.AssertUtil().assert_result(validate, result, res_status)
            TestRendRequest.logger.info(f'正常返回请求结果{result}')
        elif result['errcode']==45009:
            print('今天获取token次数已满，明天再试')
            TestRendRequest.logger.info('今天获取token次数已满，明天再试')
        else:
            print('异常')
            TestRendRequest.logger.error('出现其他错误，执行错误用例')

    @pytest.mark.file
    @pytest.mark.parametrize('caseinfo',yaml_util.YamlUtil().read_testcase_yaml('get_flag.yaml'))
    def test_get_flag(self,caseinfo):
        '''获取公众号已创建的标签'''
        value=yaml_util.YamlUtil().read_extract_yaml('access_token')
        url=caseinfo['test']['request']['url']+f'?access_token={value}'
        method = caseinfo['test']['request']['method']
        validate=caseinfo['test']['validate']
        rep=requests.request(method=method,url=url)
        result = rep.json()
        res_status=rep.status_code
        # print(result)
        #获取其中一组id值，用于后期编辑标签使用
        TestRendRequest.jsValue=result['tags'][1]['id']
        if 'tags' in result:
            yaml_util.YamlUtil().write_extract_yaml({'jsValue': result['tags'][1]['id']})
            assert_util.AssertUtil().assert_result(validate, result, res_status)
            TestRendRequest.logger.info('成功获取公众号已创建的标签')
            TestRendRequest.logger.info('jsValue:%s' % result['tags'][1]['id'])
        else:
            TestRendRequest.logger.error('没有获取到公众号已创建的标签')
        print(TestRendRequest.jsValue)

    # @pytest.mark.file
    @pytest.mark.parametrize('caseinfo',yaml_util.YamlUtil().read_testcase_yaml('create_flag.yaml'))
    def test_create_flag(self,caseinfo):
        '''创建标签'''
        value = yaml_util.YamlUtil().read_extract_yaml('access_token')
        url=caseinfo['test']['request']['url']+f'?access_token={value}'
        # headers = {
        #     'Content-Type': 'application/json'
        # }
        data={"tag":{"name":f"桃花{TestRendRequest.number_flag}"}}
        headers=caseinfo['test']['request']['headers']
        # data=caseinfo['test']['request']['data']
        print(data)
        method = caseinfo['test']['request']['method']
        validate=caseinfo['test']['validate']
        rep=requests.request(method=method,url=url,json=data,headers=headers)
        res_status=rep.status_code
        rep.encoding='utf-8'
        result=rep.text
        # result = rep.json()
        # result = rep.content
        assert_util.AssertUtil().assert_result(validate, result, res_status)
        TestRendRequest.logger.info('创建标签')
        print(result)

    # @pytest.mark.file
    @pytest.mark.parametrize('caseinfo', yaml_util.YamlUtil().read_testcase_yaml('edit_flag.yaml'))
    def test_edit_flag(self,caseinfo):
        '''编辑标签'''
        value = yaml_util.YamlUtil().read_extract_yaml('access_token')
        url=caseinfo['test']['request']['url']+f'?access_token={value}'
        headers = caseinfo['test']['request']['headers']
        method = caseinfo['test']['request']['method']
        validate = caseinfo['test']['validate']
        jsValue = yaml_util.YamlUtil().read_extract_yaml('jsValue')
        # data = {"tag" : {"id" :TestRendRequest.jsValue,"name" : f"taohua{TestRendRequest.number_flag}"} }
        data = {"tag" : {"id" : jsValue,"name" : f"taohua{TestRendRequest.number_flag}"} }
        rep = requests.request(method=method,url=url, json=data, headers=headers)
        result=rep.json()
        res_status=rep.status_code
        assert_util.AssertUtil().assert_result(validate,result,res_status)
        TestRendRequest.logger.info('编辑标签')
        print(rep.json())

    # @pytest.mark.file
    @pytest.mark.parametrize('caseinfo', yaml_util.YamlUtil().read_testcase_yaml('delect_flag.yaml'))
    def test_delect_flag(self,caseinfo):
        '''删除标签'''
        value = yaml_util.YamlUtil().read_extract_yaml('access_token')
        jsValue = yaml_util.YamlUtil().read_extract_yaml('jsValue')
        url=caseinfo['test']['request']['url']+f'?access_token={value}'
        headers = caseinfo['test']['request']['headers']
        method=caseinfo['test']['request']['method']
        validate = caseinfo['test']['validate']
        data = {"tag" : {"id" :jsValue} }
        rep = requests.request(method=method,url=url, json=data, headers=headers)
        result = rep.json()
        res_status = rep.status_code
        assert_util.AssertUtil().assert_result(validate, result, res_status)
        TestRendRequest.logger.info('删除标签')
        print(rep.json())

    # @pytest.mark.file
    @pytest.mark.parametrize('caseinfo', yaml_util.YamlUtil().read_testcase_yaml('updata_photo.yaml'))
    def test_file_upload(self,caseinfo):
        '''上传图片'''
        value = yaml_util.YamlUtil().read_extract_yaml('access_token')

        url = caseinfo['test']['request']['url'] + f'?access_token={value}'
        method = caseinfo['test']['request']['method']
        validate = caseinfo['test']['validate']
        # data = caseinfo['test']['request']['data']
        # print(data)
        data={
            "media":open(r'./data/1.jpg',"rb")
        }
        rep=requests.request(method=method,url=url,files=data)
        result = rep.json()
        res_status = rep.status_code
        print(rep.json())
        assert_util.AssertUtil().assert_result(validate, result, res_status)
        TestRendRequest.logger.info('上传图片')

-   test:
        name: 创建标签:正常填写参数
        request:
            headers:
                Postman-Token: 9445b6b0-4a08-42dd-ab80-4984efe4d538
                User-Agent: PostmanRuntime/7.26.8
                Content-Type: application/json
            method: POST
            url: https://api.weixin.qq.com/cgi-bin/tags/create
#            data: {"tag": {"name": 'f"{TestRendRequest.number_flag}"'}}
            #${rand_str(5, 10)}生成5到10为随机字符串
        validate:
        -   equals:
              status_code:
               200
#        -   equals:
#              errcode:
#               45056
#        -   contains: too many tag now

-   test:
        name: 创建标签:正常填写参数
        request:
            headers:
                Postman-Token: 9445b6b0-4a08-42dd-ab80-4984efe4d538
                User-Agent: PostmanRuntime/7.26.8
                Content-Type: application/json
            method: POST
            url: https://api.weixin.qq.com/cgi-bin/tags/create
#            data: {"tag": {"name": 'f"{TestRendRequest.number_flag}"'}}
            #${rand_str(5, 10)}生成5到10为随机字符串
        validate:
        -   equals:
              status_code:
               200
#        -   equals:
#              errcode:
#               45056
#        -   contains: too many tag now

-   test:
        name: 编辑标签:正常填写参数
        request:
            headers:
                Postman-Token: 9445b6b0-4a08-42dd-ab80-4984efe4d538
                User-Agent: PostmanRuntime/7.26.8
                Content-Type: application/json
            method: POST
            url: https://api.weixin.qq.com/cgi-bin/tags/update
        validate:
        -   equals:
              status_code:
               200
        -   equals:
              errcode:
               0
        -   contains: ok

-   test:
        name: 获取公众号已创建的标签:正常获取
        request:
            headers:
                Postman-Token: 9445b6b0-4a08-42dd-ab80-4984efe4d538
                User-Agent: PostmanRuntime/7.26.8
            method: GET
            url: https://api.weixin.qq.com/cgi-bin/tags/get
        validate:
        -   equals:
              status_code:
               200
        -   equals:
#            - $.tags[0].id
#            - 1
              tags[0].id:
               2

-   test:
        name: 获取Access_token值:正常填写参数
        request:
            headers:
                Postman-Token: d56d528a-8c84-4127-9df7-da8a32555898
                User-Agent: PostmanRuntime/7.26.8
            method: GET
            params:
                appid: wx6b11b3efd1cdc290
                grant_type: client_credential
                secret: 106a9c6157c4db5f6029918738f9529d
            url: https://api.weixin.qq.com/cgi-bin/token
#        extract:
#            tokens: content.access_token
        validate:
        -   equals:
              status_code:
                200
        -   contains: access_token
#              headers.Content-Type:
#                application/json; encoding=utf-8
        -   equals:
#              - $.expires_in
              expires_in:
                7200

-   test:
        name: 获取Access_token值:appid为空获取
        request:
            headers:
                Postman-Token: d56d528a-8c84-4127-9df7-da8a32555898
                User-Agent: PostmanRuntime/7.26.8
            method: GET
            params:
                appid:
                grant_type: client_credential
                secret: 106a9c6157c4db5f6029918738f9529d
            url: https://api.weixin.qq.com/cgi-bin/token
        validate:
        -   equals:
             status_code:
               200
        -   contains: appid missing rid
        -   equals:
             errcode:
               41002

-   test:
        name: 上传图片:正常填写参数
        request:
            method: POST
            url: https://api.weixin.qq.com/cgi-bin/media/uploadimg
#            data:
#              media:
#                open(r'./data/1.jpg','rb')

        validate:
        -   equals:
              status_code:
               200
        -   contains: http

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2023-03-27 10:29
# @Author : hexiuxiu
# @File : conftest.py
'''
@pytest.fixture() 一般会和 conftest.py 文件一起使用。conftest.py 文件的名称是固定的， 功能很强大。

1. conftest.py 文件是单独存放@pytest.fixture() 方法的， 用处是可以在多个py文件中共享前后置配置。

2. conftest.py 文件里面的方法在调用时不需要导入， 可以直接使用。

3. conftest.py 文件可以有多个。也可以有多个不同的层级

4. 在实际项目中，@pytest.fixture() 和 conftest.py的结合， 可以用来做登录、数据库连接等多种前置功能配置，它的好处就是多个py文件能共享，不用每个类都写一次
'''
import pytest
from common.yaml_util import YamlUtil

@pytest.fixture(scope='session',autouse=True)
def clear_yaml():
    YamlUtil().clear_extract_yaml()

pytest
pytest-html
pytest-xdist
pytest-ordering
pytest-rerunfailures
allure-pytest