import pymysql
import datetime
import json
import decimal
import requests
import warnings
"""
超市供应链数据测试:1、product表和supplier表拆分重组与supply_introduce表对比
2、接口返回数据和supply_introduce表数据是否一致
ps:测试账号拥有所有区域、所有商品品类权限
"""
class DecimalEncoder(json.JSONEncoder):
def default(self,obj):
if isinstance(obj,decimal.Decimal):#decimal类型转换,使其可以转换为json格式数据
return float(obj)
elif isinstance(obj,datetime.datetime):#datetime类型转换,使其可以转换为json格式数据
return obj.__str__()
return super(DecimalEncoder,self).default(obj)
class RUNSQL(object):
def run_sql(self):
db=pymysql.connect(host='192.168.147.192',port=3306,user='order_base',password='JMZet0h0fmKsTt2n',db='supply_chain',charset='utf8')
self.db_cursor1=db.cursor()
self.db_cursor2=db.cursor()
self.db_cursor3=db.cursor()
self.db_cursor4=db.cursor()
self.db_cursor1.execute("SELECT id,cooperate_area_ids FROM `supplier`;")
self.db_cursor2.execute("SELECT id,supplier_id FROM `product`;")
self.db_cursor3.execute("SELECT product_id,area_id FROM `supply_introduce`;")
self.db_cursor4.execute("SELECT product_id,area_id,product_status FROM supply_introduce;")
db.close()
def run_data(self):
self.run_sql()
sup_data = []
new_sup_data=[]
#操作supplier表
for i in self.db_cursor1.fetchall():
if '1' in i[1] and '2' not in i[1]:
str_s=json.dumps(i[1],cls=DecimalEncoder,ensure_ascii=False)
str_s=str_s.replace('1','2')
sup_data.append([i[0],str_s.strip('"')])
elif '1'in i[1]and '2'in i[1]:
str_s = json.dumps(i[1], cls=DecimalEncoder, ensure_ascii=False)
if '1,' in i[1]:
str_s1=str_s.replace('1,','')
sup_data.append([i[0], str_s1.strip('"')])
if ',1' in i[1]:
str_s2=str_s.replace(',1','')
sup_data.append([i[0],str_s2.strip('"')])
else:
sup_data.append([i[0],i[1]])
# print(data)
# print(len(data))
for line in sup_data:
if ','in line[1]:
lines=line[1].split(',')
# print(lines)
for j in lines:
new_sup_data.append([line[0],j])
else:
new_sup_data.append([line[0],line[1]])
# print(new_sup_data)
# print(len(new_sup_data))
#操作product表,先验证supplier表和product表的关联id是否都存在
a,b=[],[]
sql2=self.db_cursor2.fetchall()
for i in sql2:
a.append(i[1])
for i in new_sup_data:
b.append(i[0])
for i in a:
if i not in b:
print('supplier表中不存在的id:',i)
for i in b:
if i not in a:
print('product表中不存在的supplier_id:',i)
pro_data=[]
for i in sql2:
for line in new_sup_data:
if i[1]==line[0]:#存在相等就添加至列表
# pro_data.append([i[0],line[0],line[1]])
pro_data.append([i[0],line[1]])
else:
pass
# print(pro_data)
# print(len(pro_data))
#操作supply_introduce表
supin_data=[]
for i in self.db_cursor3.fetchall():
supin_data.append([i[0],str(i[1])])
# print(supin_data)
# print(len(supin_data))
#对比数据
for i in pro_data:
if i not in supin_data:
print('pro_data对比supin_data(supply_introduce中不含的数据):',i)
else:
pass
for i in supin_data:
if i not in pro_data:
print('supin_data对比pro_data(product表中不含的数据):',i)
else:
pass
#管理后台数据与supply_introduce表数据对比
supin_datas=[]
for i in self.db_cursor4.fetchall():
supin_datas.append([i[0],str(i[1]),i[2]])
warnings.filterwarnings('ignore')
url = 'https://dev-hlj.rainbowcn.com/api/login-adapter/user/login'
data = {
"account": "127458",
"password": "LU123456"
}
session = requests.session() # 保持登陆状态
session.post(url, verify=False, json=data)
# url_text=session.post(url,json=data)
# print(json.dumps(url_text.json(),indent=4,ensure_ascii=False,sort_keys=True))
# Token=url_text.json()['data']['token']
# Token={'token':Token}
# return Token
all_data = []
url_1 = 'https://dev-hlj.rainbowcn.com/scm-admin-api/supplier/listByPager'
url_text_1 = session.post(url_1, verify=False, json={})
for i in range(1, int(url_text_1.json()["pagerInfo"]["pages"]) + 1):
url_text_all = session.post(url_1, verify=False, json={"pageNum": i})
for j in range(len(url_text_all.json()["pagerInfo"]["list"])):
# print(url_text_all.json()["pagerInfo"]["list"][j]["productId"],url_text_all.json()["pagerInfo"]["list"][j]["cooperateAreaIds"],url_text_all.json()["pagerInfo"]["list"][j]["cooperateStatus"])
all_data.append([url_text_all.json()["pagerInfo"]["list"][j]["productId"],
url_text_all.json()["pagerInfo"]["list"][j]["cooperateAreaIds"],
url_text_all.json()["pagerInfo"]["list"][j]["cooperateStatus"]])
url_text_2 = session.post(url_1, verify=False, json={"cooperateStatus": 100})
for i in range(1, int(url_text_2.json()["pagerInfo"]["pages"]) + 1):
url_text_all = session.post(url_1, verify=False, json={"cooperateStatus": 100, "pageNum": i})
for j in range(len(url_text_all.json()["pagerInfo"]["list"])):
# print(url_text_all.json()["pagerInfo"]["list"][j]["productId"],url_text_all.json()["pagerInfo"]["list"][j]["cooperateAreaIds"],url_text_all.json()["pagerInfo"]["list"][j]["cooperateStatus"])
all_data.append([url_text_all.json()["pagerInfo"]["list"][j]["productId"],
url_text_all.json()["pagerInfo"]["list"][j]["cooperateAreaIds"],
url_text_all.json()["pagerInfo"]["list"][j]["cooperateStatus"]])
# print(sorted(all_data))
# print(sorted(supin_datas))
if sorted(all_data)==sorted(supin_datas):
print('管理后台数据与supply_introduce表数据一致')
else:
if len(all_data)>=len(supin_datas):
for i in all_data:
if i not in supin_datas:
print('supply_introduce表无数据:',i)
else:
for i in supin_datas:
if i not in all_data:
print('管理后台无数据:',i)
if __name__=='__main__':
s=RUNSQL()
s.run_data()
接口返回数据与数据数据的对比、数据表之间的拆分重组对比
猜你喜欢
转载自blog.csdn.net/lssrain/article/details/85062582
今日推荐
周排行