Skip to content
GitLab
Commit b80be0f2 authored by xiao-hesheng's avatar xiao-hesheng
Browse files

数据库验证流程十一脚本编写

parent 1db7efb0
Hide whitespace changes
Inline Side-by-side
air_case/b6_spd3_core_business_flow_database_check/a_b4流程十一请领业务流_二级库向直送_数据库验证.air/a_b4流程十一请领业务流_二级库向直送_数据库验证.py
View file @ b80be0f2
......@@ -618,16 +618,16 @@ def check_mcms_mcms_pur_plan_detail(id, type, planOrderNo):
hos_goods_id = FileUtils().r_info8('b5_spd3_core_business_flow', '产品审核信息', 'message2')["hosGoodsId"]
mdm_goods_code = process_tuple(
get_id("select mdm_goods_code from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';" % (
hos_goods_id, id2)))
hos_goods_id, id2)))
# print(mdm_goods_code)
goods_code = process_tuple(
get_id("select goods_code from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';" % (
hos_goods_id, id2)))
hos_goods_id, id2)))
pkg_def_id = process_tuple(
get_id("select pkg_def_id from mcms_pkg where hos_goods_id='%s' LIMIT 1;" % hos_goods_id))
supply_id = process_tuple(
get_id("select supply_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';" % (
hos_goods_id, id2)))
hos_goods_id, id2)))
prov_id = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_corpId"]
prov_name = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"]
sub_prov_id = prov_id
......@@ -768,11 +768,14 @@ def check_mcms_purchase(type=16):
info = FileUtils().r_info8("b5_spd3_core_business_flow", "库区信息", 'message22')
list_a = info['list_a']
area_code = commonFuc().analysis_json('shelfCode', list_a)
area_name = '默认库区'
sql1 = "select area_name from mcms_purchase where branch_id='%s' and id='%s' order by create_time desc LIMIT 1;" % (
branch_id, id1)
area_name = process_tuple(get_id(sql1))
rec_user = FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1']
# rec_user = get_login_user_uxid()
rec_addr = '自动化测试'
sql1 = "select rec_addr from mcms_purchase where branch_id='%s' and id='%s' order by create_time desc LIMIT 1;" % (
branch_id, id1)
rec_addr = process_tuple(get_id(sql1))
sql = "select bill_expdt_date from mcms_pur_plan where branch_id='%s' and source_id='%s' order by create_time desc LIMIT 1;" % (
branch_id, buyBillId)
print(sql)
......@@ -786,8 +789,8 @@ def check_mcms_purchase(type=16):
"select last_modified from mcms_purchase where branch_id='%s' and id='%s' order by create_time desc LIMIT 1;" % (
branch_id, id1))
last_modified_user = create_user
last_auditor=rec_user
last_audit_time= get_create_time(
last_auditor = rec_user
last_audit_time = get_create_time(
"select last_audit_time from mcms_purchase where branch_id='%s' and id='%s' order by create_time desc LIMIT 1;" % (
branch_id, id1))
......@@ -816,7 +819,7 @@ def check_mcms_purchase(type=16):
close_date = get_create_time(
"select close_date from mcms_purchase where id='%s';" % (
id1))
source_id=planBillId
source_id = planBillId
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'mcms_purchase', id1, order_no, hos_id,
get_branch_id(), dept_id1, dept_name1, rec_org_id, rec_org_name,
......@@ -946,16 +949,16 @@ def check_mcms_purchase_detail(id, type, planOrderNo):
hos_goods_id = FileUtils().r_info8('b5_spd3_core_business_flow', '产品审核信息', 'message2')["hosGoodsId"]
mdm_goods_code = process_tuple(
get_id("select mdm_goods_code from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';" % (
hos_goods_id, id2)))
hos_goods_id, id2)))
# print(mdm_goods_code)
goods_code = process_tuple(
get_id("select goods_code from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';" % (
hos_goods_id, id2)))
hos_goods_id, id2)))
pkg_def_id = process_tuple(
get_id("select pkg_def_id from mcms_pkg where hos_goods_id='%s' LIMIT 1;" % hos_goods_id))
supply_id = process_tuple(
get_id("select supply_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';" % (
hos_goods_id, id2)))
hos_goods_id, id2)))
prov_id = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_corpId"]
prov_name = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"]
sub_prov_id = prov_id
......@@ -1058,20 +1061,22 @@ def check_mcms_purchase_detail(id, type, planOrderNo):
else:
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_mcms_purchase_out_and_in(type=1):
if type==1:
sql="select * from mcms_purchase order by create_time desc LIMIT 3;"
if type == 1:
sql = "select * from mcms_purchase order by create_time desc LIMIT 3;"
else:
sql = "select * from mcms_purchase_detail order by create_time desc LIMIT 3;"
result_in=check_in_check_mcms_purchase(sql)
result_out=check_out_check_mcms_purchase(sql)
result_in = check_in_check_mcms_purchase(sql)
result_out = check_out_check_mcms_purchase(sql)
result = (compare_text_index(result_in, result_out))
print(result)
print(result_in)
print(result_out)
i=0
i = 0
while True:
if i==1000:
if i == 1000:
print('采购订单同步到外网已经过去16分钟未完成，请检查同步是否出现问题')
commonFuc().check_text_exist_result_text('error', 'succees')
break
......@@ -1087,11 +1092,13 @@ def check_mcms_purchase_out_and_in(type=1):
i = 1 + 10
continue
def check_hdi_distr(type=16):
from common.db.sql.sql_tools_out import check_mcms_pur_plan_sql,get_id
from common.db.sql.sql_tools_out import check_mcms_pur_plan_sql, get_id
branch_id = get_branch_id()
#验证配送单主表
sql="select * from hdi_distr where branch_id = '%s' and bill_mode='%s'order by create_time desc LIMIT 1;"%(branch_id,type)
# 验证配送单主表
sql = "select * from hdi_distr where branch_id = '%s' and bill_mode='%s'order by create_time desc LIMIT 1;" % (
branch_id, type)
# 获取实际值
actual = check_mcms_pur_plan_sql(sql)
print('actual', actual)
......@@ -1099,33 +1106,37 @@ def check_hdi_distr(type=16):
# 获取采购订单id
purchase_id = get_id(
"select id from mcms_purchase where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
branch_id, type))
# 查询配送单主键
distr_main_key = get_id(
"select id from hdi_distr a where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
branch_id, type))
# 查询配送单order_no
distr_order_no= get_id(
distr_order_no = get_id(
"select order_no from hdi_distr a where id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
distr_main_key, type))
#从采购订单获取source_id
source_id= get_id(
distr_main_key, type))
# 从采购订单获取source_id
source_id = get_id(
"select id from mcms_purchase where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
pur_bill_id=source_id
rec_user='自动化测试w&vqF'
branch_id, type))
pur_bill_id = source_id
rec_user = '自动化测试w&vqF'
sql = "select id from spd3_herp_test2.mcms_dept_buy where branch_id='%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type)
print(sql)
buyBillId = process_tuple(get_dept_buy_id(sql))
purBillDate
distrBillId
distrOrderNo
distrUser = 'p1e35e:UID-052165'
create_user = 'p1e35e:UID-052165'
create_user_name = '国药集团公司S57ui'
create_time
last_modified_user
purBillDate = get_expdt_date(
"select create_time from mcms_purchase where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
distrBillId = distr_main_key
distrOrderNo = distr_order_no
distrUser = get_login_user_uxid()
create_user = distrUser
create_user_name = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"]
create_time = get_create_time(
"select create_time from mcms_purchase where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
last_modified_user = create_time
if type == 16:
# 验证子表数据
# check_mcms_mcms_pur_plan_detail(id2, type, planOrderNo)
......@@ -1133,13 +1144,13 @@ def check_hdi_distr(type=16):
planBillId = process_tuple(
get_id(
"select pid from mcms_pur_plan_detail where hos_goods_id='%s' order by create_time desc LIMIT 1;" % (
"select pid from spd3_herp_test2.mcms_pur_plan_detail where hos_goods_id='%s' order by create_time desc LIMIT 1;" % (
hos_goods_id)))
planOrderNo = process_tuple(
get_id(
"select order_no from mcms_pur_plan_detail where hos_goods_id='%s' order by create_time desc LIMIT 1;" % (
"select order_no from spd3_herp_test2.mcms_pur_plan_detail where hos_goods_id='%s' order by create_time desc LIMIT 1;" % (
hos_goods_id)))
sql = "select order_no from mcms_dept_buy where branch_id='%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
sql = "select order_no from spd3_herp_test2.mcms_dept_buy where branch_id='%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type)
print(sql)
buyOrderNo = process_tuple(get_dept_buy_id(sql))
......@@ -1156,9 +1167,20 @@ def check_hdi_distr(type=16):
prov_name = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"]
exe_prov_id = prov_id
exe_prov_name = prov_name
pur_dept_id = FileUtils().r_info8("b2_herp3_bs", "所有科室id", 'message6')['deptid1']
pur_dept_name = FileUtils().r_info8("b2_herp3_bs", "所有科室name", 'message6')['deptid1']
rec_org_id = process_tuple(
get_id("select id from spd3_herp_test2.sys_org where branch_id='%s' and ename='%s';" % (branch_id, pur_dept_name)))
rec_org_name = pur_dept_name
last_modified=create_time
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'mcms_purchase', distr_main_key,distr_order_no,hos_id,get_hosname(),get_branch_id(),source_id,pur_bill_id,prov_id,prov_name,exe_prov_id,exe_prov_name,pur_dept_id,pur_dept_name,rec_org_id,rec_org_name,rec_user
'mcms_purchase', distr_main_key, distr_order_no, hos_id, get_hosname(),
get_branch_id(), source_id, pur_bill_id, prov_id, prov_name,
exe_prov_id, exe_prov_name, pur_dept_id, pur_dept_name, rec_org_id,
rec_org_name, rec_user, buyBillId, planBillId, planOrderNo, buyOrderNo,
purOrderNo, purBillId, purBillDate, distrBillId, distrOrderNo,
distrUser, create_user, create_user_name, create_time, last_modified,
last_modified_user
)
elif type == 20:
......@@ -1249,6 +1271,7 @@ def check_hdi_distr(type=16):
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
try:
pro_path = commonFuc().get_pro_path2()
print('pro_path', pro_path)
......@@ -1266,49 +1289,49 @@ try:
from a_b4流程十一请领业务流_二级库向直送 import check_accept_order
for i in range(1):
goods_move2() # 设置产品出库渠道为直送
# # 低值
second_dept_Request(FileUtils().r_info8("b5_spd3_core_business_flow", "产品审核信息", 'message2')['hosGoodsId'], 1)
check_mcms_dept_buy_car(1)
# # 高值
second_dept_Request(FileUtils().r_info8("b5_spd3_core_business_flow", "产品审核信息", 'message1')['hosGoodsId'], 2)
check_mcms_dept_buy_car(2)
# # 试剂
second_dept_Request(FileUtils().r_info8("b5_spd3_core_business_flow", "产品审核信息", 'message3')['hosGoodsId'], 3)
check_mcms_dept_buy_car(3)
BuyCarSubmit() # 请领车提交
list_data = ['16', '20', '66']
for i in list_data:
check_mcms_dept_buy(int(i))
mcms_dept_buy_detail(int(i))
for i in range(6): # 三个单据共审批6次
approval_center()
list_data = ['16', '20', '66']
for i in list_data:
check_mcms_dept_buy_approval_after(int(i))
mcms_dept_buy_detail_approval_after(int(i))
# 检查自动生成的采购计划数据
list_data = ['16', '20', '66']
for i in list_data:
check_mcms_mcms_pur_plan(int(i))
# 采购模块处理
purchase_module_process()
for i in range(3): # 三个单据共审批3次
approval_center()
list_data = ['16', '20', '66']
for i in list_data:
check_mcms_purchase(int(i))
#检查同步到外网的采购订单数据跟内网的是否一致
check_mcms_purchase_out_and_in(1)
check_mcms_purchase_out_and_in(2)
order_dp(1) # 1低值
order_dp(2) # 2高值
order_dp(3) # 3试剂
# goods_move2() # 设置产品出库渠道为直送
# # # 低值
# second_dept_Request(FileUtils().r_info8("b5_spd3_core_business_flow", "产品审核信息", 'message2')['hosGoodsId'], 1)
# check_mcms_dept_buy_car(1)
# # # 高值
# second_dept_Request(FileUtils().r_info8("b5_spd3_core_business_flow", "产品审核信息", 'message1')['hosGoodsId'], 2)
# check_mcms_dept_buy_car(2)
# # # 试剂
# second_dept_Request(FileUtils().r_info8("b5_spd3_core_business_flow", "产品审核信息", 'message3')['hosGoodsId'], 3)
# check_mcms_dept_buy_car(3)
# BuyCarSubmit() # 请领车提交
#
# list_data = ['16', '20', '66']
# for i in list_data:
# check_mcms_dept_buy(int(i))
# mcms_dept_buy_detail(int(i))
# for i in range(6): # 三个单据共审批6次
# approval_center()
# list_data = ['16', '20', '66']
# for i in list_data:
# check_mcms_dept_buy_approval_after(int(i))
# mcms_dept_buy_detail_approval_after(int(i))
#
# # 检查自动生成的采购计划数据
# list_data = ['16', '20', '66']
# for i in list_data:
# check_mcms_mcms_pur_plan(int(i))
# # 采购模块处理
# purchase_module_process()
#
# for i in range(3): # 三个单据共审批3次
# approval_center()
#
# list_data = ['16', '20', '66']
# for i in list_data:
# check_mcms_purchase(int(i))
#
# #检查同步到外网的采购订单数据跟内网的是否一致
# check_mcms_purchase_out_and_in(1)
# check_mcms_purchase_out_and_in(2)
# order_dp(1) # 1低值
# order_dp(2) # 2高值
# order_dp(3) # 3试剂
# 上面暂时注释=========================================
check_hdi_distr()
check_accept_order(1) # 低值
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment