企业微信接口测试实战2

测试用例编写

import re

import pytest
import requests


@pytest.fixture(scope="session")
def test_token():
    res = None
    # 获取 token
    corpid = "wwe653983e4c732493"
    corpsecret = "T72_Vgw9TaNS-FLDU2gJlw6AteerMXsuMval9kGNZbc"
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&"
                       f"corpsecret={corpsecret}")
    return res.json()["access_token"]


def test_get(userid, test_token):
    # 根据 user-id查询成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token={test_token}&userid={userid}")
    return res.json()


def test_create(userid, name, mobile, test_token):
    # 添加成员
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
        "department": [1],
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/create?access_token={test_token}",
                        json=data
                        )
    return res.json()


def test_update(userid, name, mobile, test_token):
    # 更新成员
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/update?access_token={test_token}",
                        json=data)
    return res.json()


def test_delete(userid, test_token):
    # 删除成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/delete?access_token={test_token}&userid={userid}")
    return res.json()


@pytest.mark.parametrize("userid, name, mobile", [("zhangsan123456", "小白", "13811110009")])
def test_all(userid, name, mobile, test_token):
    # 可能发生创建失败
    try:
        assert "created" == test_create(userid, name, mobile, test_token)["errmsg"]
    except AssertionError as e:
        if "mobile existed" in e.__str__():
            # 如果手机号被使用了,找出使用手机号的 userid ,进行删除
            re_userid = re.findall(":(.*)'$", e.__str__())[0]
            assert "deleted" == test_delete(re_userid, test_token)['errmsg']
            assert 60111 == test_get(re_userid, test_token)['errcode']
            assert "created" == test_create(userid, name, mobile, test_token)["errmsg"]
    # 可能发生userid不存在异常
    assert name == test_get(userid, test_token)['name']
    assert "updated" == test_update(userid, "xxxxxxx", mobile, test_token)['errmsg']
    assert "xxxxxxx" == test_get(userid, test_token)['name']
    assert "deleted" == test_delete(userid, test_token)['errmsg']
    assert 60111 == test_get(userid, test_token)['errcode']

并行

pytest-xdist

import re

import pytest
import requests
from filelock import FileLock


@pytest.fixture(scope="session")
def test_token():
    res = None
    # 获取 token
    while FileLock("session.lock"):
        corpid = "wwe653983e4c732493"
        corpsecret = "T72_Vgw9TaNS-FLDU2gJlw6AteerMXsuMval9kGNZbc"
        res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&"
                           f"corpsecret={corpsecret}")
    return res.json()["access_token"]


def test_get(userid, test_token):
    # 根据 user-id查询成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token={test_token}&userid={userid}")
    return res.json()


def test_create(userid, name, mobile, test_token):
    # 添加成员
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
        "department": [1],
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/create?access_token={test_token}",
                        json=data
                        )
    return res.json()


def test_update(userid, name, mobile, test_token):
    # 更新成员
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/update?access_token={test_token}",
                        json=data)
    return res.json()


def test_delete(userid, test_token):
    # 删除成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/delete?access_token={test_token}&userid={userid}")
    return res.json()


def test_create_data():
    data = [("wu123fff" + str(x), "zhangsan", "138%08d" % x) for x in range(20)]
    return data


@pytest.mark.parametrize("userid, name, mobile", test_create_data())
def test_all(userid, name, mobile, test_token):
    # 可能发生创建失败
    try:
        assert "created" == test_create(userid, name, mobile, test_token)["errmsg"]
    except AssertionError as e:
        if "mobile existed" in e.__str__():
            # 如果手机号被使用了,找出使用手机号的 userid ,进行删除
            re_userid = re.findall(":(.*)", e.__str__())[0]
            if re_userid.endswith("'") or re_userid.endswith('"'):
                re_userid = re_userid[:-1]
            assert "deleted" == test_delete(re_userid, test_token)['errmsg']
            assert 60111 == test_get(re_userid, test_token)['errcode']
            assert "created" == test_create(userid, name, mobile, test_token)["errmsg"]
    # 可能发生userid不存在异常
    assert name == test_get(userid, test_token)['name']
    assert "updated" == test_update(userid, "xxxxxxx", mobile, test_token)['errmsg']
    assert "xxxxxxx" == test_get(userid, test_token)['name']
    assert "deleted" == test_delete(userid, test_token)['errmsg']
    assert 60111 == test_get(userid, test_token)['errcode']

配置文件:pytest.ini

[pytest]
addopts = -n auto

课间作业

利用 pytest-xdist 完成并行代码

数据驱动

image

课后反馈表:https://hogwarts.jinshuju.com/f/M1PWLd

课间作业

import random
import re

import pytest
import requests

@pytest.fixture(scope="session")
def test_get_token():
    pyload = {
        "corpid": "ww653e65d70cbfb643",
        "corpsecret": "1WH5MioXvKPka5wIKjo-tReIYbNrlbLe9537vpxZZEc"
    }
    r = requests.get("https://qyapi.weixin.qq.com/cgi-bin/gettoken", params=pyload)
    token = r.json()["access_token"]
    return token

def test_get_member(userid, test_get_token):
    pyload = {
        "access_token": test_get_token,
        "userid": userid
    }
    r = requests.get("https://qyapi.weixin.qq.com/cgi-bin/user/get", params=pyload)
    return r.json()

def test_create_member(userid, name, mobile, test_get_token):
    datas = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
        "department": [1]
    }
    r = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/create?access_token={test_get_token}", json=datas)
    return r.json()

def test_change_member(userid, name, mobile, test_get_token):
    datas = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
    }
    r = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/update?access_token={test_get_token}", json=datas)
    return r.json()

def test_delete_member(userid, test_get_token):
    r = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/delete?access_token={test_get_token}&userid={userid}")
    return r.json()

def set_datas():
    datas = [("wy"+str(x), "无忧"+str(x), "1382064%04d" %x) for x in range(10)]
    # print(datas)
    return datas

@pytest.mark.parametrize("userid, name, mobile", set_datas())
def test_all(userid, name, mobile, test_get_token):
    try:
        assert 'created' == test_create_member(userid, name, mobile, test_get_token)["errmsg"]
    except AssertionError as e:
        s = e.__str__()
        re_userid: str = ""
        if "mobile existed" in s:
            re_userid = re.findall(":(.*)", s)[0]
            if re_userid.endswith('"') or re_userid.endswith("'"):
                re_userid = re_userid[:-1]
        assert "deleted" == test_delete_member(re_userid, test_get_token)["errmsg"]
        assert 'created' == test_create_member(userid, name, mobile, test_get_token)["errmsg"]
    assert name == test_get_member(userid, test_get_token)['name']
    assert "updated" == test_change_member(userid, '李李', mobile, test_get_token)["errmsg"]
    assert '李李' == test_get_member(userid, test_get_token)['name']
    assert "deleted" == test_delete_member(userid,test_get_token)["errmsg"]
    assert 60111 == test_get_member(userid,test_get_token)["errcode"]


课件作业,使用部门管理接口, 利用 pytest-xdist 完成并行代码

import pytest
import requests


@pytest.fixture(scope="session")
def test_access_token():
    corpid = 'ww197063f50a78f00e'
    corpsecert = 'KjeanuIGls0J7_mwx0KNHaw2Jp_yDO7NZ9fBZWkJBcM'
    url = f'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecert}'
    r = requests.get(url)
    return r.json()['access_token']


def test_get_dept(id, test_access_token):
    url = f'https://qyapi.weixin.qq.com/cgi-bin/department/list?access_token={test_access_token}&id={id}'
    r = requests.get(url)
    return r.json()


def test_add_dept(name, id, test_access_token):
    data = {
        "name": name,
        "parentid": 1,
        "id": id
    }
    url = f'https://qyapi.weixin.qq.com/cgi-bin/department/create?access_token={test_access_token}'
    r = requests.post(url, json=data)
    return r.json()


# @pytest.mark.skip
def test_update_dept(name_new, id, test_access_token):
    data = {
        "id": id,
        "name": name_new,
        "parentid": 1,
        "order": 1
    }
    url = f'https://qyapi.weixin.qq.com/cgi-bin/department/update?access_token={test_access_token}'
    r = requests.post(url, json=data)
    return r.json()


# @pytest.mark.skip
def test_delete_dept(id, test_access_token):
    url = f'https://qyapi.weixin.qq.com/cgi-bin/department/delete?access_token={test_access_token}&id={id}'
    r = requests.get(url)
    return r.json()


@pytest.mark.parametrize("name, id, name_new", [("tester1", "6", 'ceshi1'), ("tester2", "7", 'ceshi2')])
def test_all(name, id, name_new, test_access_token):
    # 异常处理:如果新增部门时,部门id已存在,则删除该部门,然后再创建部门
    try:
        assert 'created' == test_add_dept(name, id, test_access_token)['errmsg']
    except AssertionError as e:
        if 'department existed' in e.__str__():
            assert 'deleted' == test_delete_dept(id, test_access_token)['errmsg']
            assert 'created' == test_add_dept(name, id, test_access_token)['errmsg']
    assert 'updated' == test_update_dept(name_new, id, test_access_token)['errmsg']
    assert name_new == test_get_dept(id, test_access_token)['department'][0]['name']
    assert 'deleted' == test_delete_dept(id, test_access_token)['errmsg']
 
import re

import pytest
import requests

from jsonpath import jsonpath
base_url = 'https://qyapi.weixin.qq.com'
token = {'corpid': 'wwf2dbb0a93f2eac33', 'corpsecret': 'fmDbF_Ll4GPiYmrqHrqtztTTVkbG6Z1NqXjiJ-eQ1gc'}


@pytest.fixture(scope='session')
def get_token():
    r = requests.get(f'{base_url}/cgi-bin/gettoken', params=token)
    access_token = jsonpath(r.json(), '$.access_token')[0]
    return access_token

def test_create_data():
    data = [('xsd{0}'.format(x), 'xiaoming{}'.format(x), '189%08d' % x) for x in range(10)]
    return data
class Test_req():



    # 添加成员
    @pytest.mark.parametrize("userid,name,mobile",test_create_data())
    def test_create(self,get_token,userid,name,mobile):

        json={
  "userid": userid,
  "name": name,
  "alias": "jackzhangw",
  "mobile": mobile,
  "department": [1],
  "order": [10, 40],
  "position": "产品经理",
  "gender": "1",
  # "email": "zhankkge@gzmmv.com",
  "is_leader_in_dept": [1],
  "enable": 1,
  "telephone": "020-123456",
  "address": "广州市海珠区新港中路",
  "extattr": {
    "attrs": [{
        "type": 0,
        "name": "文本名称",
        "text": {
          "value": "文本"
        }
      },
      {
        "type": 1,
        "name": "网页名称",
        "web": {
          "url": "http://www.test.com",
          "title": "标题"
        }
      }
    ]
  },
  "to_invite": True,
  "external_position": "产品经理"
}


        
        url = f'{base_url}/cgi-bin/user/create?access_token='+get_token

        r=requests.post(url,json=json)

        print(r.json())
        assert  'created'==jsonpath(r.json(), '$.errmsg')[0]
        assert jsonpath(r.json(), '$.errcode')[0] == 0


# 查询成员
    def test_get_mem(self,get_token,userid,name=None):

        url=f'{base_url}/cgi-bin/user/get?access_token={get_token}&userid={userid}'
        r=requests.get(url)
        print(r.json())
        assert jsonpath(r.json(), '$.errcode')[0] == 0
        if name:
            assert jsonpath(r.json(),'$.name')[0]==name
    # 更新成员
    def test_update_mem(self,get_token,userid,name):

        url=f'{base_url}/cgi-bin/user/update?access_token={get_token}'
        json={
        "userid": userid,
        "name": name,
        "department": [1]
        }
        r = requests.post(url,json=json)
        print(r.text)
        assert jsonpath(r.json(), '$.errcode')[0] == 0
    # 删除成员
    def test_delete_mem(self,get_token,userid):
        url=f'{base_url}/cgi-bin/user/delete?access_token={get_token}&userid={userid}'
        r = requests.get(url)
        assert jsonpath(r.json(), '$.errcode')[0] == 0

    #成员列表
    def test_simplelist(self,get_token):
        url = f'{base_url}/cgi-bin/user/simplelist?access_token={get_token}&department_id=1&fetch_child=0'
        r = requests.get(url)
        assert jsonpath(r.json(), '$.errcode')[0] == 0

    @pytest.mark.parametrize("userid,name,mobile", test_create_data())
    def test_all(self, get_token, userid, name, mobile):
        try:
            self.test_create(get_token, userid, name, mobile)
        except AssertionError as e:
            if "mobile existed" in e.__str__():
                re_userid = re.findall(":(.*)'$", e.__str__())[0]
                self.test_delete_mem(get_token, re_userid)
                self.test_create(get_token, userid, name, mobile)
        self.test_get_mem(get_token, userid)
        self.test_update_mem(get_token, userid, '小米')
        self.test_get_mem(get_token, userid, '小米')
        self.test_delete_mem(get_token, userid)
···
import pytest
import requests

# 获取token
@pytest.fixture(scope='session')
def test_get_token():
    corpid = 'ww65cd3275dad0af38'
    corpsecret = 'ml7ZFg_LGtGK7g-mxwZ8rFbDnBG87JNPw2MtIxpkBgw'
    r = requests.get(f'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}')
    print(r.json()['access_token'])
    return r.json()['access_token']


# 创建部门
def test_department_create(name, test_get_token):
    payload = {
        "name": name,
        "parentid": 1
    }
    r = requests.post(f'https://qyapi.weixin.qq.com/cgi-bin/department/create?access_token={test_get_token}',
                      json=payload)
    print(r.json())
    return r.json()


# 更新部门
def test_department_update(department_id, department_name, test_get_token):
    payload = {
        "id": department_id,
        "name": department_name
    }

    r = requests.post(f'https://qyapi.weixin.qq.com/cgi-bin/department/update?access_token={test_get_token}',
                      json=payload)
    print(r.json())
    return r.json()


# 删除部门
def test_department_delete(department_id, test_get_token):
    r = requests.get(
        f'https://qyapi.weixin.qq.com/cgi-bin/department/delete?access_token={test_get_token}&id={department_id}')
    print(r.json())
    return r.json()


# 获取部门列表,返回部门列表
def test_department_list(test_get_token, department_id=1):
    r = requests.get(
        f'https://qyapi.weixin.qq.com/cgi-bin/department/list?access_token={test_get_token}&id={department_id}')
    print(r.json())
    return r.json()


@pytest.mark.parametrize('department_name', ["测试部门0706" + str(x) for x in range(1, 30)])
def test_all(test_get_token, department_name):
    try:
        # 可能会存在部门名称已存在的异常
        creat_res = test_department_create(department_name, test_get_token)
        assert 'created' == creat_res['errmsg']
    except AssertionError as e:
        # 如果异常信息中有部门已存在的信息,则进行处理
        if "department existed" in e.__str__():
            # 获取所有部门信息,遍历,如果有部门名称与参数的部门名称相同,则执行删除对应id的方法
            department_list = test_department_list(test_get_token)["department"]
            for department in department_list:
                if department_name == department["name"]:
                    test_department_delete(department["id"])
    new_department_name = '111' + department_name
    assert "updated" == test_department_update(creat_res['id'], new_department_name, test_get_token)['errmsg']
    assert new_department_name == test_department_list(test_get_token, creat_res['id'])["department"][0]["name"]
    assert "deleted" == test_department_delete(creat_res['id'], test_get_token)['errmsg']

##课间作业

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import pytest
import requests


# 获取token
from filelock import FileLock


@pytest.fixture(scope='session')
def test_token():
    with FileLock("session.lock"):
        corpid = 'wwb9dd59b36dfb3ab0'
        corpsecret = 'Bg-l59yCtimW-H8rBSiRC_IWDsF07awtF0UGTtb3KuU'
        req = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}")
    return req.json()['access_token']


# 创建部门
# @pytest.mark.parametrize("name, name_en, ID", [('上海研发中心2', 'Develop2', 2)])
def test_create_dep(name, name_en, ID, test_token):
    data = {
        "name": name,
        "name_en": name_en,
        "parentid": 1,
        "id": ID
    }
    req = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/department/create?access_token={test_token}",
                        json=data
                        )
    return req.json()


# 修改部门信息
def test_edit_dep(ID, name, test_token):
    data = {
        "id": ID,
        "name": name,
    }
    req = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/department/update?access_token={test_token}",
                        json=data
                        )
    return req.json()


# 获取部门列表
def test_get_dep(ID, test_token):
    req = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/department/list?access_token={test_token}&id={ID}")
    # print(req.json())
    return req.json()


# 删除部门
def test_del_dep(ID, test_token):
    req = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/department/delete?access_token={test_token}&id={ID}")
    # print(req.json())
    return req.json()


# 根据部门名称获取部门ID
# @pytest.mark.parametrize("name", ['上海研发中心'])
def test_get_DepID(name, test_token):
    get_all_deps = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/department/list?access_token={test_token}")
    list_deps = get_all_deps.json()['department']
    for dep in list_deps:
        if name == dep['name']:
            del_id = dep['id']
            return del_id


# 准备测试数据
def test_create_data():
    data = [("上海研发中心" + str(i),
             "Develop" + str(i),
             i) for i in range(3, 10)]
    return data


# 测试用例
@pytest.mark.parametrize("name, name_en, ID", test_create_data())
def test_all(name, name_en, ID, test_token):
    try:
        assert "created" == test_create_dep(name, name_en, ID, test_token)['errmsg']
    except AssertionError as e:
        # 如果部门名称重复或者部门ID重复,找到同名的部门id并删除
        if "department existed" in e.__str__():
            del_id = test_get_DepID(name, test_token)
            if del_id != 'None':
                assert "deleted" == test_del_dep(del_id, test_token)['errmsg']
                assert "created" == test_create_dep(name, name_en, ID, test_token)['errmsg']
            else:
                assert "deleted" == test_del_dep(ID, test_token)['errmsg']
                assert "created" == test_create_dep(name, name_en, ID, test_token)['errmsg']
    assert name == test_get_dep(ID, test_token)['department'][0]['name']
    assert "updated" == test_edit_dep(ID, '北京研发中心', test_token)['errmsg']
    assert "北京研发中心" == test_get_dep(ID, test_token)['department'][0]['name']
    assert "deleted" == test_del_dep(ID, test_token)['errmsg']
    assert 60123 == test_del_dep(ID, test_token)['errcode']

课间作业

import re
from filelock import FileLock
import pytest
import requests

# request的基本使用
# 功能测试基础之上根据接口文档进行的接口测试
# 实际测试需根据的接口之间的关联情况进行设计,此处暂无做关联

@pytest.fixture(scope="session")
#dist中此处的session无作用,所以需要使用锁来实现单个session的限制
def test_token():
    res = None
    # 获取 token
    corpid = "ww12e3ecd647cf1230"
    corpsecret = "hFIkHWRIEQdOGrJe6kv-ggC7bluk22OzVpDWRYasBcI"
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&corpsecret={corpsecret}")
    return res.json()["access_token"]

def test_createmember(userid, name, mobile, test_token):
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
        "department": [1]
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/create?access_token={test_token}", json=data)
    return res.json()


def test_getmember(userid, test_token):
    # userid 查询成语言
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token={test_token}&userid={userid}")
    return res.json()


def test_updatemember(userid, name, mobile, test_token):
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/update?access_token={test_token}", json=data)
    return res.json()

def test_deletemember(userid, test_token):
    # 根据userid删除成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/delete?access_token={test_token}&userid={userid}")
    return res.json()

def test_create_data():
    data=[("xiaobai"+str(x),"AAA","138%08d"%x) for x in range (10)]
    # data = [("wu123fff" + str(x), "zhangsan", "138%08d" % x) for x in range(10)]
    return data

@pytest.mark.parametrize("userid, name, mobile",test_create_data())
def test_all(userid, name, mobile, test_token):
    # 一般在每个test中进行断言,此处暂写在一起
    # 异常处理:创建失败
    try:
        assert "created" == test_createmember(userid, name, mobile, test_token)["errmsg"]
    except AssertionError as e:
        if "mobile existed" in e.__str__():
            # 如果手机号被使用了,找出使用手机号的 userid ,进行删除
            re_userid = re.findall(":(.*)", e.__str__())[0]
            if re_userid.endswith("'") or re_userid.endswith('"'):
                re_userid = re_userid[:-1]
            assert "deleted" == test_deletemember(re_userid,test_token)["errmsg"]
            assert 60111 == test_getmember(re_userid,test_token)['errcode']
            assert "created" == test_createmember(userid, name, mobile, test_token)["errmsg"]
    # 成员添加无异常
    assert name == test_getmember(userid, test_token)['name']
    assert "updated" == test_updatemember(userid, "ABC", mobile, test_token)["errmsg"]
    assert "ABC" == test_getmember(userid, test_token)['name']
    assert "deleted" == test_deletemember(userid,test_token)["errmsg"]
    assert 60111 == test_getmember(userid,test_token)["errcode"]
关闭