练习api：接口文档 - 企业微信开发者中心
jsonpath: https://github.com/json-path/JsonPath
代码地址:https://github.com/yuruotong1/lagou-1
作业：实现课上内容，使用 po 封装
/api/Baseapi.py
import json
import time
import allure
import requests
import yaml
class BaseApi:
def yaml_steps(self, path, module):
with allure.step("获取token"):
access_token = self.get_token(module)
with open(path) as f:
request = yaml.safe_load(f)
# print(request)
for r in request:
print(r["comment"])
with allure.step(r["comment"]):
try:
params = {"access_token": access_token}
data = {}
if r['params'] is not None:
for key in r['params'].keys():
params[key] = r['params'][key]
if r['data'] is not None:
data = r['data']
if "use_var" in r.keys():
for var in save_var:
data = json.dumps(data)
data = data.replace(var, save_var[var])
data = json.loads(data)
print(data)
re = requests.request(method=r["method"], url=r["url"], params=params, json=data)
# if r["method"] == "get":
# re = requests.get(url=url, params=params, json=data)
# if r["method"] == "post":
# re = requests.post(url=url, params=params, json=data)
if "save_var" in r.keys():
save_var = r["save_var"]
for key in save_var.keys():
if key == "tagid":
save_var[key] = re.json()["tag_group"][0]["tag"][save_var[key]]["id"]
print(re.json())
if r['assert'] is not None:
for key in r['assert'].keys():
assert r['assert'][key] == re.json()[key]
except Exception:
print("failed")
continue
def get_token(self, module) -> str:
with open("../api/baseconfig.yaml") as f:
x = yaml.safe_load(f)
y = 0 # y记录当前列表指针
for d in x:
if d['module'] == module:
id, secret, access_token, last_time = d['id'], d['secret'], d['access_token'], d['last_time']
break
else:
y += 1
now = int(time.time())
# print(d)
if now - last_time > 7200:
url = f'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={id}&corpsecret={secret}'
r = requests.get(url=url)
print(r.json())
d['access_token'] = r.json()['access_token']
d['last_time'] = now
x[y] = d
with open("../api/baseconfig.yaml", "w") as f:
yaml.dump(x, f)
return d['access_token']
else:
return access_token
def send_api(self, req: dict):
r = requests.request(**req).json()
print(json.dumps(r, indent=2))
return r
/api/baseconfig.yaml
- access_token: cf5rrw6K4B4PmMsjCwud6vnGohZf--6njBdsUYaKC5__kkStSWDEyq0SCMon3w-OxUdO0pLJPMyAtYsmYrJze_052KBWlwb81bNZWj9Z-ccTaI5hejibhxMEM2XpiuEwd2Mj1cB_bhtbFsNTKsnk00tuRRe_ywRoIF3cHftFi1soe_B_QW6lG-1VMpirxIEPCzRdSWFIBT95Xd-XiqGfuw
id: ww6cb5b7542d579ade
last_time: 1590073406
module: address
secret: H3vRXGeRKJ-OOypBGMvv0sq87c0aQIgHbzePpPr7bhI
- access_token: SDolpJOybOJ0xFQ9aAaA129DuxXUqd4vzxDZcitJ-18trHj27E6-EfnvhP5QvuCB5AfS1TqmuVrE5a7gn40LHXXbqVMo-ixzjrXwJJjp5pdngl-SCrurVRCVD4bAKZGkVTZp8K4xrPhFbWXAHlAdToNYoh7P_ASWvP4qNRMpVUVRn-s770K0S93hgqG951nVl2Kf7_-qvBw_zreU311laQ
id: ww6cb5b7542d579ade
last_time: 1590073449
module: externalcontact
secret: J-oExWZiWylBmnct05KFYfuX3odiuaFPdJcmAJ48VSk
/api/externalContact.py
import json
import re
from jsonpath import jsonpath
from httpapi.workweixin.POcase.api.BaseApi import BaseApi
class externalContact(BaseApi):
module = "externalcontact"
def __init__(self):
self.token = self.get_token(module=self.module)
def add_corp_tag(self, tag_name):
data = {
"method": "post",
"url": 'https://qyapi.weixin.qq.com/cgi-bin/externalcontact/add_corp_tag',
"params": {
"access_token": self.token
},
"json": {
"group_name": "test",
"tag": [{"name": tag_name}]
}
}
return self.send_api(data)
def get_corp_tag_list(self):
data = {
"method": "post",
"url": "https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get_corp_tag_list",
"params": {
"access_token": self.token
},
"json": {
"tag_id": []
}
}
return self.send_api(data)
def del_corp_tag(self, tag_id):
data = {
"method": "post",
"url": "https://qyapi.weixin.qq.com/cgi-bin/externalcontact/del_corp_tag",
"params": {
"access_token": self.token
},
"json": {
"tag_id": [
tag_id
]
}
}
return self.send_api(data)
def edit_corp_tag(self, tag_id, name):
data = {
"method": "post",
"url": "https://qyapi.weixin.qq.com/cgi-bin/externalcontact/edit_corp_tag",
"params": {
"access_token": self.token
},
"json": {
"id": tag_id,
"name": name
}
}
return self.send_api(data)
def get_tag_id_by_name(self, name=""):
# if name == "":
# tag_id = jsonpath(self.get_corp_tag_list(), '$..tag[0].id')
# else:
l = self.get_corp_tag_list()
tag_id = jsonpath(l, f'$..tag[?(@.name=="{name}")].id')
if tag_id is False:
raise Exception("name not is exist")
return tag_id[0]
def get_allname(self) -> list:
r = self.get_corp_tag_list()
if not r["tag_group"]:
return None
else:
pattern = re.compile('"name": "(.+?)"') # 设定正则模式
result = pattern.findall(json.dumps(r)) # 获得结果
return result
/testcase/test_externalcontact.py
import random
import pytest
from httpapi.workweixin.POcase.api.externalcontact import externalContact
class Test_externalContact:
def setup_class(self):
self.tag = externalContact()
@pytest.mark.parametrize("name", ["zhangsan2"])
def test_add_not_exist(self, name):
# name不存在直接添加，存在的话先删除再添加
names = self.tag.get_allname()
if names is not None and name in names:
self.tag.del_corp_tag(self.tag.get_tag_id_by_name(name))
r = self.tag.add_corp_tag(name)
assert r["errcode"] == 0
assert r["errmsg"] == "ok"
assert r["tag_group"]["tag"][0]["name"] == name
@pytest.mark.parametrize("name", ["zhangsan3"])
def test_add_is_exist(self, name):
# name存在直接添加，不存在的话先添加再添加
names = self.tag.get_allname()
if names is None or name not in names:
self.tag.add_corp_tag(name)
r = self.tag.add_corp_tag(name)
assert r["errcode"] in (40068, 40071)
def test_get(self):
assert self.tag.get_corp_tag_list()["errcode"] == 0
def test_edit(self):
old_name = "zhangsan3"
new_name = "lisi"
names = self.tag.get_allname()
if old_name not in names:
self.tag.add_corp_tag(old_name)
if new_name not in names:
new_name = "lisi" + str(random.randint(1, 1000))
tag_id = self.tag.get_tag_id_by_name(old_name)
assert self.tag.edit_corp_tag(tag_id, new_name)["errcode"] == 0
names = self.tag.get_allname()
assert new_name in names
def test_del(self):
names = self.tag.get_allname()
if names is None:
self.tag.add_corp_tag("lisi")
names = self.tag.get_allname()
for name in names:
self.tag.del_corp_tag(self.tag.get_tag_id_by_name(name))
assert self.tag.get_allname() is None
def test_with_yaml(self):
self.tag.yaml_steps(path="test_externalContact.yaml", module=self.tag.module)
/testcase/test_externalContact.yaml
- comment: 添加企业客户标签
method: post
params:
url: https://qyapi.weixin.qq.com/cgi-bin/externalcontact/add_corp_tag
data:
group_name: test
tag:
- name: zhangsan
assert:
errcode: 0
errmsg: ok
- comment: 获取企业标签库
method: post
params:
url: https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get_corp_tag_list
data:
assert:
errcode: 0
errmsg: ok
save_var:
tagid: 0
- comment: 编辑企业客户标签
method: post
params:
url: https://qyapi.weixin.qq.com/cgi-bin/externalcontact/edit_corp_tag
use_var: True
data:
id: tagid
name: wangwu
assert:
errcode: 0
errmsg: ok
- comment: 删除企业客户标签
method: post
params:
url: https://qyapi.weixin.qq.com/cgi-bin/externalcontact/del_corp_tag
use_var: True
data:
tag_id:
- tagid
assert:
errcode: 0
errmsg: ok
https://github.com/gqxjones/MyprojectHogwarts/tree/master/httpTest
po封装：git目录下的api&test_case目录
测试结果：
base_api
import requests
class BaseAPI:
def send_api(self, req):
return requests.request(**req).json()
wechat_work
from study_requests.wechat.po.base_api import BaseAPI
class WechatWork(BaseAPI):
ID = 'ww03972ce56e110e69'
def get_token(self, corpsecret):
req = {
'method': 'get',
'url': 'https://qyapi.weixin.qq.com/cgi-bin/gettoken',
'params': {
'corpid': self.ID,
'corpsecret': corpsecret
}
}
return self.send_api(req)['access_token']
tag
from study_requests.wechat.po.wechat_work import WechatWork
class Tag(WechatWork):
tagSecret = 'YGFVXAaX70np5UbRPliymNHJHQMiWCg1O5l5zLTVfOk'
def __init__(self):
self.token = self.get_token(self.tagSecret)
def get_tags(self):
req = {
'method': 'post',
'url': 'https://qyapi.weixin.qq.com/cgi-bin/externalcontact/get_corp_tag_list',
'params': {
'access_token': self.token,
}
}
return self.send_api(req)
def add_tag(self, groupName, tagName):
req = {
'method': 'post',
'url': 'https://qyapi.weixin.qq.com/cgi-bin/externalcontact/add_corp_tag',
'params': {
'access_token': self.token,
},
'json': {
"group_name": groupName,
"tag": [{
"name": tagName
}]
}
}
return self.send_api(req)
def delete_tag(self, tagId):
req = {
'method': 'post',
'url': 'https://qyapi.weixin.qq.com/cgi-bin/externalcontact/del_corp_tag',
'params': {
'access_token': self.token,
},
'json': {
"tag_id": [
tagId
]
}
}
return self.send_api(req)
def update_tag(self, tagId, tagName):
req = {
'method': 'post',
'url': 'https://qyapi.weixin.qq.com/cgi-bin/externalcontact/edit_corp_tag',
'params': {
'access_token': self.token,
},
'json': {
"id": tagId,
"name": tagName
}
}
return self.send_api(req)
test_tag
import json
from jsonpath import jsonpath
from study_requests.wechat.po.tag import Tag
class TestTag:
def setup(self):
self.tag = Tag()
def test_get_tags(self):
# 格式化json
print(json.dumps(self.tag.get_tags(), indent=2))
def test_add_tag(self):
print(self.tag.add_tag('testGroup', 'tag3'))
def test_delete_tag(self):
tagId = jsonpath(self.tag.get_tags(), '$..tag[?(@.name=="tag2")].id')[0]
print(self.tag.delete_tag(tagId))
def test_update_tag(self):
tagId = jsonpath(self.tag.get_tags(), '$..tag[?(@.name=="tag3")].id')[0]
print(self.tag.update_tag(tagId, 'newTag3'))
def test_link(self):
tags = self.tag.get_tags()
print(tags)
tagId = jsonpath(tags, '$..tag[?(@.name=="tag3")].id')
# 如果已存在
if tagId:
# 删除
print(self.tag.delete_tag(tagId[0]))
# 添加
res = self.tag.add_tag('testGroup', 'tag3')
print(res)
newTagId = jsonpath(res, '$..id')
print(newTagId)
# 更新
print(self.tag.update_tag(newTagId[0], 'updateTag3'))
api:
https://github.com/mengqiD11/Http/blob/master/api/base_api.py
https://github.com/mengqiD11/Http/blob/master/api/department.py
https://github.com/mengqiD11/Http/blob/master/api/depart_all.yaml
testcases:
https://github.com/mengqiD11/Http/blob/master/testcases/test_wk_conlabl.py
https://github.com/mengqiD11/Http/blob/master/testcases/test_department_data.yaml
https://github.com/testQx/interface/blob/master/baseapi/baseapi.py
https://github.com/testQx/interface
https://github.com/tim8709/hogwarts_testing/blob/master/test_requests/api/base_api.py
https://github.com/tim8709/hogwarts_testing/blob/master/test_requests/api/wework.py
https://github.com/tim8709/hogwarts_testing/blob/master/test_requests/api/tag.py
代码github地址：
https://github.com/Amber-huangby/request_project_1
testcase：
class TestTag():
def setup_class(self):
self.tag = Tag()
@pytest.mark.parametrize('group_id, old_tag_name, new_tag_name', [('g1', 't1', 't1111')])
def test_tag(self, group_id, old_tag_name, new_tag_name):
# 首先删除垃圾数据
json_data = self.tag.get_alltag()
names=jsonpath(json_data, f"$..tag[*].name")
if old_tag_name in names:
tag_id = jsonpath(json_data, f"$..tag[?(@.name=='{old_tag_name}')].id")[0]
assert self.tag.delete_tag(tag_id)['errcode'] == 0
if new_tag_name in names:
tag_id = jsonpath(json_data, f"$..tag[?(@.name=='{new_tag_name}')].id")[0]
assert self.tag.delete_tag(tag_id)['errcode'] == 0
# 新增企业标签
assert self.tag.add_tag(group_id, old_tag_name)['errcode'] == 0
# 更新企业标签
json_data = self.tag.get_alltag()
tag_id = jsonpath(json_data, f"$..tag[?(@.name=='{old_tag_name}')].id")[0]
assert self.tag.update_tag(tag_id, new_tag_name)['errcode'] == 0
