pythonテストhttp、websocketインターフェース
22832 ワード
テスト環境には毎日データを作成して、できるだけテスト環境と生産環境データ量が一致するようにする項目があります。注文書を作成してから、業者は注文書を受け取って完成します。以下はコードです。主にインターフェースで完成します。
注文コードを完成しました。
注文コードを完成しました。
# -*- coding: utf-8 -*
import urllib3
import requests
import logging
import time
import os
from TestAllocation.dcpay.wss import wss
from TestAllocation.dcpay.yaml_ini import read_yaml
urllib3.disable_warnings()
time_log = time.strftime('%Y-%m-%d %H_%M_%S', time.localtime(time.time()))
# log_path = r'E:\PythonWork\StructuralData\TestAllocation\dcpay\logs' + '\\' + time_log + '.log'
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
# logging.basicConfig(filename=log_path, filemode='w+', level=logging.INFO, format='%(asctime)s - %(message)s')
key_dc = read_yaml()['key_dc']
url_login = read_yaml()['url_login']
url_open_take = read_yaml()['url_open_take']
url_rob = read_yaml()['url_rob']
url_pay = read_yaml()['url_pay']
# url_upon = read_yaml()['url_upon']
dc_head = read_yaml()['dc_head']
# backstage_head = read_yaml()['backstage_head']
dc_phone = read_yaml()['dc_phone']
def login():
list_uid = []
list_head = []
for line in dc_phone:
try:
# requests.adapters.DEFAULT_RETRIES = 10
# s = requests.session()
# s.keep_alive = False
sessionRequest = requests.session()
#
app_login = app_login
r_login = sessionRequest.post(url=url_login, data=data, headers=dc_head, verify=False, timeout=5)
# print(r_login.url)
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 = head
list_uid.append(uid)
list_head.append(head)
# logging.info(u'HEAD: ' + str(list_head) + '
')
# logging.info(u'HEAD: ' + str(list_head) + '
')
except BaseException as msg:
logging.info(u'---ERROR:' + str(msg) + '
')
return list_uid, list_head
def accomplishOrderDc():
list_data = login()
list_uid = list_data[0]
list_head = list_data[1]
while True:
for uid, head in zip(list_uid, list_head):
try:
# logging.info(uid)
# logging.info(head)
sessionRequest = requests.session()
# #
# merchant_pay_list = requests.get(url=url_code_list, headers=head, verify=False)
# # id
# code_wx_id = merchant_pay_list.json()['data'][0]['id']
# code_zfb_id = merchant_pay_list.json()['data'][1]['id']
# r_start_grab = requests.post(url=url_start_grab, data=data},
# headers=head, verify=False)
# logging.info(' :' + r_start_grab.text)
r_start = sessionRequest.post(url=url_open_take, data=data, headers=head, verify=False,
timeout=5)
# logging.info(' :' + r_start.text)
# # r_buy_off = requests.post(url=url_buy, params=data, headers=head, verify=False)
# # logging.info(' :' + r_buy_off.text)
# #
# # url_sell_off = requests.post(url=url_buy, params=data,
# # headers=head,
# # verify=False)
# # logging.info(' :' + url_sell_off.text)
#
# r_buy = requests.post(url=url_buy, data=data, headers=head, verify=False)
# logging.info(' :' + r_buy.text)
#
# r_sell = requests.post(url=url_buy, data=data, headers=head,
# verify=False)
# logging.info(' :' + r_sell.text)
try:
if wss(uid) is not None:
#
order_id = wss(uid)
r_rob = sessionRequest.post(url=url_rob, data=data, headers=head, verify=False,
timeout=10)
# logging.info(u' :' + r_rob.text)
amount = r_rob.json()['data']['amount']['reality']
order_zid = r_rob.json()['data']['id']
#
r_pay = sessionRequest.post(url=url_pay, data=data, headers=head,
verify=False, timeout=5)
logging.info(r_pay.text + ' uid' + uid + ' orderId: ' + order_zid + '
')
except BaseException as msg:
logging.info(u'ERROR:' + str(msg) + '
')
except BaseException as msg:
logging.info(u'ERROR:' + str(msg) + '
')
if __name__ == '__main__':
accomplishOrderDc()
受注コード:# coding=utf-8
import re
from websocket import create_connection
from TestAllocation.dcpay.yaml_ini import read_yaml
def wss(uid):
try:
ws = create_connection(read_yaml()['url_wss'])
ws.send('data')
num = 0
while True:
ws.send('data')
send_data = ws.recv()
if 'invoice' in str(send_data):
# print(send_data)
order_id = re.findall(r'', send_data)
# print(order_id)
return order_id
if num == 100:
break
num = num + 1
except BaseException as msg:
print('ERROR: ' + msg)
if __name__ == '__main__':
wss('uid')