测试环境有个项目需要每天构造数据,来尽量保证测试环境和生产环境数据量保持一致。需要生成订单后商家接单完成,以下是代码,主要是用接口完成

创建订单

# coding=utf-8

import requests
import time
import uuid
import hashlib
import hmac
import urllib3
# from config import read_yaml urllib3.disable_warnings() def shunfeng():
# 时间戳
millis = "%d" % (time.time() * 1000) # 订单号
order_id = ''.join((str(uuid.uuid1()).split('-'))) url = 'https://test'
mid = 'test'
secret = 'test' sec_data = {"data": mid,
"data": 'HmacSHA256',
"data": '',
"data": millis,
"data": order_id,
"orderType": "",
"data": "",
"data": "WechatPay",
"data": "",
"data": "110.110.110.110",
"data": "https://test.com",
"data": "CNY",
"data": order_id} sec_data_str = ''
for k in sorted(sec_data):
sec_data_str += k + '=' + sec_data[k] + '&'
sec_data_str = sec_data_str[:-1]
print('\nsing排序后参数:\n'+sec_data_str) signature_str = hmac.new(secret.encode('utf-8'), sec_data_str.encode('utf-8'), digestmod=hashlib.sha256).digest()
sign = signature_str.hex().upper()
print('\nsign:\n' + sign) body_data = {"data": mid,
"data": 'HmacSHA256',
"data": '',
"data": millis,
"data": order_id,
"data": "",
"data": "",
"data": "WechatPay",
"data": "",
"data": "110.110.110.110",
"data": "https://test.com",
"data": "CNY",
"data": order_id,
"data": sign} body_str = ''
for i in body_data.items():
body_str += i[0] + '=' + i[1] + '&'
body_str = body_str[:-1]
print('\nbody参数:\n' + body_str) head = {"Content-Type": "application/x-www-form-urlencoded"} req = requests.post(url=url, data=body_str, headers=head)
print('\n请求url:\n' + req.url)
print('\n下单返回结果:' + req.text) if __name__ == '__main__':
shunfeng()

接收订单推送

# coding=utf-8

import urllib3
import re
from websocket import create_connection urllib3.disable_warnings() def wss(uid): url = 'wss://test'
ws = create_connection(url) ws.send('{"test":2,"tese":16,"test":1,"body":{"test":"test","test":"test","test":"1.0","test":"1.0"}}') while True: ws.send('{"test":5,"test":16,"test":2,"test":{"userId":' + '"' + uid + '"' + '}}')
send_data = ws.recv()
if 'test' in str(send_data): str1 = str(send_data).split('\\') str2 = str(str1).split(' '' ') str3 = ''.join(str(str2).split('\\')) str4 = re.sub("[\"\'[, ]", "", str3) str5 = re.findall(r'invoice:(.*?)=', str4) order_id = ''.join(str5) + '='
# print(order_id) return order_id if __name__ == '__main__':
wss('uid')

登陆-抢单-完成

import urllib3
import requests
import logging
import time
from web_wss import wss urllib3.disable_warnings() time_log = time.strftime('%Y-%m-%d %H_%M_%S', time.localtime(time.time()))
# logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
logging.basicConfig(filename= 'result.log', filemode= 'w+', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def login():
list_uid = []
list_head = [] phone = phone login_header = {'User-Agent': 'test/1.0.0(build:test;ANDROID 9)',
'bndkey': 'test'} for line in phone:
try:
# requests.adapters.DEFAULT_RETRIES = 10
# s = requests.session()
# s.keep_alive = False sessionRequest = requests.session() # 登陆
app_login = {'account': '86,' + line,
'accountKind': '',
'passwd': 'test'}
r_login = sessionRequest.post(url=url_login, data=app_login, headers=login_header, verify=False, timeout=10)
# print(r_login.url) if '登录成功' in r_login.text:
logging.info(u'登陆成功: ' + line + ' - ' + r_login.json()['data']['uid'])
else:
logging.info(u'登陆失败: ' + r_login.text) # 请求头
uid = r_login.json()['data']['uid']
u_token = dict(r_login.headers)['utoken']
t_token = dict(r_login.headers)['ttoken']
head = {'User-Agent': 'test/1.0.0(build:test;ANDROID 7.1.1)',
'bndkey': 'test',
'utoken': u_token,
'ttoken': t_token} r_start = sessionRequest.post(url=url_open_take, data={'main': ''}, headers=head, verify=False, timeout=10)
if 'true' in r_start.text:
logging.info(u'接单开启成功: uid - ' + str(uid) + ' - ' + r_start.text)
else:
logging.info(u'接单开启失败: uid - ' + str(uid) + ' - ' + r_start.text) r_sell = requests.post(url=url_sell, data={'merchantDeposit': '', 'playerDeposit': ''}, headers=head, verify=False, timeout=10)
if 'true' in r_sell.text:
logging.info(u'卖币开启成功: uid - ' + str(uid) + ' - ' + r_sell.text)
else:
logging.info(u'卖币开启失败: uid - ' + str(uid) + ' - ' + r_sell.text) list_uid.append(uid)
list_head.append(head) logging.info(u'uid: ' + str(uid) + ' - utoken: ' + str(u_token) + ' - ttoken: ' + str(t_token) + '\n')
# logging.info(u'HEAD: ' + str(list_head) + '\n') except BaseException as msg:
logging.info(u'---ERROR:' + str(msg) + '\n' + '\n' + '\n') return list_uid, list_head def accomplishOrder():
list_data = login()
list_uid = list_data[0]
list_head = list_data[1] for uid, head in zip(list_uid, list_head):
try:
sessionRequest = requests.session() # list_order_id = []
# list_amount = [] try:
if wss(uid) is not None:
# 抢单
order_id_sec = wss(uid)
r_rob = sessionRequest.post(url=url_rob, data={'id': order_id_sec}, headers=head, verify=False, timeout=10)
logging.info(u'抢单: ' + r_rob.text) # 查询进行中订单
r_get_order = sessionRequest.get(url=url_run_order, headers=head, verify=False, timeout=10) if r_get_order.json()['data'] is not None:
order_data = r_get_order.json()
for i in range(len(r_get_order.json()['data'])):
order_id = order_data['data'][i]['id']
amount = order_data['data'][i]['amount']['reality'] # list_order_id.append(order_id)
# list_amount.append(amount)
# # logging.info('uid: ' + uid + ' - orderId: ' + order_id + '\n')
# return list_order_id, list_amount # 完成订单
r_pay = sessionRequest.post(url=url_pay, data={'id': order_id, 'amount': amount}, headers=head, verify=False, timeout=10)
if 'true' in r_pay.text:
logging.info(u'完成订单: uid - ' + uid + ' orderId: ' + order_id + '\n')
else:
logging.info(u'订单未完成: uid - ' + uid + r_pay.text)
else:
logging.info(u'暂时没有进行中订单: ' + 'uid: ' + uid) except BaseException as msg:
logging.info(u'---ERROR:' + str(msg) + '\n')
except BaseException as msg:
logging.info(u'--- ERROR:' + str(msg) + '\n') if __name__ == '__main__':
while True:
accomplishOrder()

python测试http、websocket接口的更多相关文章

  1. python测试基于websocket协议的即时通讯接口

    随着html5的广泛应用,基于websocket协议的即时通讯有了越来越多的使用场景,本文使用python中的websocket-client模块来做相关的接口测试 import webclient ...

  2. 关于python测试webservice接口的视频分享

    现在大公司非常流行用python做产品的测试框架,还有对于一些快速原型产品的开发也好,很好地支持OO编程,代码易读.Python的更新挺快的,尤其是第三方库. 对于测试人员,代码基础薄弱,用pytho ...

  3. python测试api接口

    在开发中,需要测试web-api的接口 spring mvc 使用单元测试非常方便,但是,受不了单元测试的启动速度.用python写了一个小脚本,用于测试接口, 测试脚本配置文件 api.yaml s ...

  4. jmeter 测试websocket接口(一)

    jmeter 测试websocket接口时,需要对jmeter添加测试websocket的jar包. 下载地址: https://download.csdn.net/download/qq_14913 ...

  5. python、java大作战,python测试dubbo接口

    很多小伙伴都反映公司要求测dubbo(dubbo是一个java的分布式开源框架)接口,不会写java代码,怎么测,能不能用python来调dubbo接口.当然是可以的了,最近研究了一下,很简单,分享给 ...

  6. python爬虫----爬取阿里数据银行websocket接口

    业务需求:爬取阿里品牌数据银行的自定义模块==>>>人群透视==>>>查看报告==>>数据 发现:数据通过websocket接口传递,此类型接口的详细理 ...

  7. 从零使用Python测试。Testing Using Python.

    0. 写在前面 本人使用Python测试已有多年,略有些皮毛经验.每次有新员工入职,都会从头教一遍如何入门上手使用Python进行测试.趁这段有空,整理成文档,也好方便后续新员工学习.文章如有不妥之处 ...

  8. 转载:python + requests实现的接口自动化框架详细教程

    转自https://my.oschina.net/u/3041656/blog/820023 摘要: python + requests实现的接口自动化框架详细教程 前段时间由于公司测试方向的转型,由 ...

  9. Python测试 ——开发工具库

    Web UI测试自动化 splinter - web UI测试工具,基于selnium封装. selenium - web UI自动化测试. mechanize- Python中有状态的程序化Web浏 ...

  10. GPS坐标(WGS84)转换百度坐标(BD09) python测试

    基础知识坐标系说明: WGS84:为一种大地坐标系,也是目前广泛使用的GPS全球卫星定位系统使用的坐标系. GCJ02:是由中国国家测绘局制订的地理信息系统的坐标系统.由WGS84坐标系经加密后的坐标 ...

随机推荐

  1. ※编程随笔※=>☆编程基础☆=>※№ SVN工具自动属性 $Author$ $Date$ $Revision$ $URL$ $Header$替换

    TortoiseSVN设置->常规设置->subversion 配置文件 ->编辑 或者 Windows XP 在 C:\Documents and Settings\Adminis ...

  2. (总结)Linux的chattr与lsattr命令详解

    PS:有时候你发现用root权限都不能修改某个文件,大部分原因是曾经用chattr命令锁定该文件了.chattr命令的作用很大,其中一些功能是由Linux内核版本来支持的,不过现在生产绝大部分跑的li ...

  3. 解决浏览器不支持HTML5和CSS3

    Modernizr插件可用于解决:<响应式web设计>提及! 其他: 第一种方法:IE9以下版本的IE将创建HTML5标签, 非IE浏览器会忽略这段代码,因此不会发生http请求,也就不影 ...

  4. MHD simulation with python

    这里为MHD(磁流体力学)模拟做一些准备,可能现在学习物理的人从某种程度上也得变成程序猿吧.MHD模拟面临的问题是求解一个三维的偏微分方程组,其中涉及的流体元格点非常多.所以希望先从简单的一维模型出发 ...

  5. Android实现系统下拉栏的消息提示——Notification

    Android实现系统下拉栏的消息提示--Notification 系统默认样式 默认通知(通用) 效果图 按钮 <Button android:layout_width="match ...

  6. parcel和parcelable

    Parcel 在英文中有两个意思,其一是名词,为包裹,小包的意思: 其二为动词,意为打包,扎包.邮寄快递中的包裹也用的是这个词.Android采用这个词来表示封装消息数据.这个是通过IBinder通信 ...

  7. OBJ-C中dispatch_once的用法

    见如下代码: +(GameState*)sharedGameState{ static GameState *sharedInstance; static dispatch_once_t onceTo ...

  8. Gmagick convert SVG to PNG with transparent/opacity background

    1 前言 在Ubuntu18.04环境下,用gographics/gmagick API使用了以下代码未有效果 pw := gmagick.NewPixelWand() pw.SetOpacity(1 ...

  9. postgreSQL数据库limit分页、排序

    postgreSQL数据库limit分页.排序 语法: select * from persons limit  A  offset  B; 解释: A就是你需要多少行: B就是查询的起点位置. 示例 ...

  10. hdu-4023-博弈(模拟)

    Game Time Limit: 2000/1000 MS (Java/Others)    Memory Limit: 65768/65768 K (Java/Others)Total Submis ...