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

流程八脚本提交

parent 7cab8e9c
Hide whitespace changes
Inline Side-by-side
air_case/b6_spd3_core_business_flow_database_check/a_b1流程八_高值跟台业务流程_数据库验证.air/a_b1流程八_高值跟台业务流程_数据库验证.py
View file @ c994d095
......@@ -22,18 +22,18 @@ python runner_test.py tag id2303-11 debug mdm3
"""
def check_mcms_operation_apply():#检查跟台申请单
def check_mcms_operation_apply(): # 检查跟台申请单
branch_id = get_branch_id()
sql = " select * from mcms_operation_apply where branch_id='%s' order by CREATE_TIME desc LIMIT 1;" % (
branch_id)
branch_id)
actual = check_mcms_dept_buy_sql(sql)
print(actual)
#获取主表主键ID
# 获取主表主键ID
sql = " select id from mcms_operation_apply where branch_id='%s' order by CREATE_TIME desc LIMIT 1;" % (
branch_id)
branch_id)
ID = get_id(sql)
#获取order_no
# 获取order_no
sql = " select order_no from mcms_operation_apply where id='%s';" % (ID)
order_no = get_id(sql)
rec_dept_id = FileUtils().r_info8("b2_herp3_bs", "所有科室id", 'message6')['deptid2']
......@@ -41,7 +41,6 @@ def check_mcms_operation_apply():#检查跟台申请单
target_stock_id = get_stockId_fromdb(rec_dept_name + '库房')
rec_address_id = get_stockAreaId_new_fromDb(rec_dept_name, 2)
source_corp_id = get_hosid()
source_branch_id = get_branch_id()
source_dept_id = FileUtils().r_info8("b2_herp3_bs", "所有科室id", 'message6')['deptid1']
......@@ -52,21 +51,21 @@ def check_mcms_operation_apply():#检查跟台申请单
create_user = get_login_user_uxid_bydb(FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1'])
create_time = get_create_time("select create_time from mcms_operation_apply where id='%s';" % ID)
last_modified = get_create_time("select last_modified from mcms_operation_apply where id='%s';" % ID)
oper_date= get_create_time("select oper_date from mcms_operation_apply where id='%s';" % ID)
expire_date= get_create_time("select expire_date from mcms_operation_apply where id='%s';" % ID)
oper_date = get_create_time("select oper_date from mcms_operation_apply where id='%s';" % ID)
expire_date = get_create_time("select expire_date from mcms_operation_apply where id='%s';" % ID)
last_modified_user = create_user
hos_id=get_hosid()
hos_name=get_hosname()
branch_name=get_branch_name()
#子表检查
hos_id = get_hosid()
hos_name = get_hosname()
branch_name = get_branch_name()
# 子表检查
check_mcms_operation_apply_list(ID)
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'check_mcms_operation_apply', ID, hos_id,hos_name,branch_id,branch_name,order_no,
rec_dept_id,rec_dept_name,rec_address_id,oper_date,expire_date
, create_user, create_time, last_modified, last_modified_user
)
'check_mcms_operation_apply', ID, hos_id, hos_name, branch_id, branch_name,
order_no,
rec_dept_id, rec_dept_name, rec_address_id, oper_date, expire_date
, create_user, create_time, last_modified, last_modified_user
)
print('expected', expected)
print('aaaaaaaaaaaaaaaaa')
......@@ -87,6 +86,7 @@ def check_mcms_operation_apply():#检查跟台申请单
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_mcms_operation_apply_list(pid):
branch_id = get_branch_id()
sql = " select * from mcms_operation_apply_list where pid='%s';" % (
......@@ -100,9 +100,8 @@ def check_mcms_operation_apply_list(pid):
ID = get_id(sql)
hos_goods_id = FileUtils().r_info8('b5_spd3_core_business_flow', '产品审核信息', 'message1')["hosGoodsId"]
prov_id=get_prov_id()
sub_prov_id=get_prov_id()
prov_id = get_prov_id()
sub_prov_id = get_prov_id()
create_user = get_login_user_uxid_bydb(FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1'])
create_time = get_create_time("select create_time from mcms_operation_apply_list where id='%s';" % ID)
......@@ -110,8 +109,9 @@ def check_mcms_operation_apply_list(pid):
last_modified_user = create_user
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'check_mcms_operation_apply_list', ID, pid, hos_goods_id, prov_id, sub_prov_id,
create_user, create_time, last_modified, last_modified_user
'check_mcms_operation_apply_list', ID, pid, hos_goods_id, prov_id,
sub_prov_id,
create_user, create_time, last_modified, last_modified_user
)
......@@ -134,18 +134,19 @@ def check_mcms_operation_apply_list(pid):
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_mcms_operation_apply_approval_after():#检查跟台申请单审批后的
def check_mcms_operation_apply_approval_after(): # 检查跟台申请单审批后的
branch_id = get_branch_id()
sql = " select * from mcms_operation_apply where branch_id='%s' order by CREATE_TIME desc LIMIT 1;" % (
branch_id)
branch_id)
actual = check_mcms_dept_buy_sql(sql)
print(actual)
#获取主表主键ID
# 获取主表主键ID
sql = " select id from mcms_operation_apply where branch_id='%s' order by CREATE_TIME desc LIMIT 1;" % (
branch_id)
branch_id)
ID = get_id(sql)
#获取order_no
# 获取order_no
sql = " select order_no from mcms_operation_apply where id='%s';" % (ID)
order_no = get_id(sql)
rec_dept_id = FileUtils().r_info8("b2_herp3_bs", "所有科室id", 'message6')['deptid2']
......@@ -153,7 +154,6 @@ def check_mcms_operation_apply_approval_after():#检查跟台申请单审批后
target_stock_id = get_stockId_fromdb(rec_dept_name + '库房')
rec_address_id = get_stockAreaId_new_fromDb(rec_dept_name, 2)
source_corp_id = get_hosid()
source_branch_id = get_branch_id()
source_dept_id = FileUtils().r_info8("b2_herp3_bs", "所有科室id", 'message6')['deptid1']
......@@ -164,21 +164,21 @@ def check_mcms_operation_apply_approval_after():#检查跟台申请单审批后
create_user = get_login_user_uxid_bydb(FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1'])
create_time = get_create_time("select create_time from mcms_operation_apply where id='%s';" % ID)
last_modified = get_create_time("select last_modified from mcms_operation_apply where id='%s';" % ID)
oper_date= get_create_time("select oper_date from mcms_operation_apply where id='%s';" % ID)
expire_date= get_create_time("select expire_date from mcms_operation_apply where id='%s';" % ID)
oper_date = get_create_time("select oper_date from mcms_operation_apply where id='%s';" % ID)
expire_date = get_create_time("select expire_date from mcms_operation_apply where id='%s';" % ID)
last_modified_user = create_user
hos_id=get_hosid()
hos_name=get_hosname()
branch_name=get_branch_name()
#子表检查
hos_id = get_hosid()
hos_name = get_hosname()
branch_name = get_branch_name()
# 子表检查
check_mcms_operation_apply_list(ID)
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'check_mcms_operation_apply_approval_after', ID, hos_id,hos_name,branch_id,branch_name,order_no,
rec_dept_id,rec_dept_name,rec_address_id,oper_date,expire_date
, create_user, create_time, last_modified, last_modified_user
)
'check_mcms_operation_apply_approval_after', ID, hos_id, hos_name,
branch_id, branch_name, order_no,
rec_dept_id, rec_dept_name, rec_address_id, oper_date, expire_date
, create_user, create_time, last_modified, last_modified_user
)
print('expected', expected)
print('aaaaaaaaaaaaaaaaa')
......@@ -199,6 +199,7 @@ def check_mcms_operation_apply_approval_after():#检查跟台申请单审批后
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_mcms_operation_apply_list_apply_approval_after(pid):
branch_id = get_branch_id()
sql = " select * from mcms_operation_apply_list where pid='%s';" % (
......@@ -212,9 +213,8 @@ def check_mcms_operation_apply_list_apply_approval_after(pid):
ID = get_id(sql)
hos_goods_id = FileUtils().r_info8('b5_spd3_core_business_flow', '产品审核信息', 'message1')["hosGoodsId"]
prov_id=get_prov_id()
sub_prov_id=get_prov_id()
prov_id = get_prov_id()
sub_prov_id = get_prov_id()
create_user = get_login_user_uxid_bydb(FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1'])
create_time = get_create_time("select create_time from mcms_operation_apply_list where id='%s';" % ID)
......@@ -222,8 +222,9 @@ def check_mcms_operation_apply_list_apply_approval_after(pid):
last_modified_user = create_user
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'check_mcms_operation_apply_list', ID, pid, hos_goods_id, prov_id, sub_prov_id,
create_user, create_time, last_modified, last_modified_user
'check_mcms_operation_apply_list', ID, pid, hos_goods_id, prov_id,
sub_prov_id,
create_user, create_time, last_modified, last_modified_user
)
......@@ -246,6 +247,7 @@ def check_mcms_operation_apply_list_apply_approval_after(pid):
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_mcms_operation_notice():
branch_id = get_branch_id()
sql = " select * from mcms_operation_notice where branch_id='%s' order by CREATE_TIME desc LIMIT 1;" % (
......@@ -260,8 +262,8 @@ def check_mcms_operation_notice():
# 获取order_no
sql = " select order_no from mcms_operation_notice where id='%s';" % (ID)
order_no = get_id(sql)
source_bill_id=get_id("select id from mcms_operation_apply order by create_time desc LIMIT 1;")
source_order_no=get_id("select order_no from mcms_operation_apply order by create_time desc LIMIT 1;")
source_bill_id = get_id("select id from mcms_operation_apply order by create_time desc LIMIT 1;")
source_order_no = get_id("select order_no from mcms_operation_apply order by create_time desc LIMIT 1;")
rec_dept_id = FileUtils().r_info8("b2_herp3_bs", "所有科室id", 'message6')['deptid2']
rec_dept_name = FileUtils().r_info8("b2_herp3_bs", "所有科室name", 'message6')['deptid2']
......@@ -283,16 +285,18 @@ def check_mcms_operation_notice():
hos_id = get_hosid()
hos_name = get_hosname()
branch_name = get_branch_name()
create_user_name= FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1']
prov_id=get_prov_id()
prov_name=get_prov_name()
create_user_name = FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1']
prov_id = get_prov_id()
prov_name = get_prov_name()
# 子表检查
check_mcms_operation_notice_list(ID,order_no)
check_mcms_operation_notice_list(ID, order_no)
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'check_mcms_operation_notice', ID,order_no, source_bill_id,source_order_no,hos_id, hos_name,
'check_mcms_operation_notice', ID, order_no, source_bill_id,
source_order_no, hos_id, hos_name,
branch_id, branch_name,
rec_dept_id, rec_dept_name, oper_date, prov_id
, prov_name,rec_address_id,create_user, create_user_name,create_time, last_modified, last_modified_user
, prov_name, rec_address_id, create_user, create_user_name, create_time,
last_modified, last_modified_user
)
......@@ -315,7 +319,8 @@ def check_mcms_operation_notice():
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_mcms_operation_notice_list(pid,p_order_no):
def check_mcms_operation_notice_list(pid, p_order_no):
branch_id = get_branch_id()
sql = " select * from mcms_operation_notice_list where pid='%s';" % (
pid)
......@@ -326,24 +331,26 @@ def check_mcms_operation_notice_list(pid,p_order_no):
sql = " select id from mcms_operation_notice_list where pid='%s';" % (
pid)
ID = get_id(sql)
source_bill_detail_id=get_id("select id from mcms_operation_apply_list order by create_time desc LIMIT 1;")
pkg_code=p_order_no
source_bill_detail_id = get_id("select id from mcms_operation_apply_list order by create_time desc LIMIT 1;")
pkg_code = p_order_no
hos_goods_id = FileUtils().r_info8('b5_spd3_core_business_flow', '产品审核信息', 'message1')["hosGoodsId"]
goodsCode=get_goods_code1(hos_goods_id)
goodsName=get_goods_name(hos_goods_id)
goodsCode = get_goods_code1(hos_goods_id)
goodsName = get_goods_name(hos_goods_id)
sub_prov_id=get_prov_id()
sub_prov_name=get_prov_name()
create_user_name=FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1']
sub_prov_id = get_prov_id()
sub_prov_name = get_prov_name()
create_user_name = FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1']
create_user= get_login_user_uxid_bydb(FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1'])
create_user = get_login_user_uxid_bydb(FileUtils().r_info8('b2_herp3_bs', "用户名信息", 'message3')['username1'])
create_time = get_create_time("select create_time from mcms_operation_notice_list where id='%s';" % ID)
last_modified = get_create_time("select last_modified from mcms_operation_notice_list where id='%s';" % ID)
last_modified_user = create_user
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'check_mcms_operation_notice_list', ID, pid, source_bill_detail_id,pkg_code,hos_goods_id, sub_prov_id, sub_prov_name,
goodsCode,goodsName,create_user, create_user_name,create_time, last_modified, last_modified_user
'check_mcms_operation_notice_list', ID, pid, source_bill_detail_id,
pkg_code, hos_goods_id, sub_prov_id, sub_prov_name,
goodsCode, goodsName, create_user, create_user_name, create_time,
last_modified, last_modified_user
)
......@@ -366,6 +373,7 @@ def check_mcms_operation_notice_list(pid,p_order_no):
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
def check_hdi_distr(type=25):
from common.db.sql.sql_tools_out import check_mcms_pur_plan_sql, get_id
branch_id = get_branch_id()
......@@ -393,7 +401,8 @@ def check_hdi_distr(type=25):
rec_user = info['username1']
distrBillId = distr_main_key
distrOrderNo = distr_order_no
distrUser = get_login_user_uxid_bydb(FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"])
distrUser = get_login_user_uxid_bydb(
FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"])
create_user = distrUser
create_user_name = FileUtils().r_info8('b5_spd3_core_business_flow', '供货关系申请2', 'message')["e_username"]
create_time = get_create_time(
......@@ -415,29 +424,28 @@ def check_hdi_distr(type=25):
branch_id, FileUtils().r_info8("b2_herp3_bs", "所有科室name", 'message6')['deptid2'])))
rec_org_name = FileUtils().r_info8("b2_herp3_bs", "所有科室name", 'message6')['deptid2']
last_modified = get_create_time(
"select last_modified from hdi_distr where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
ext_info = get_ext_info(branch_id, type)
operationApplyOrderNo=get_id("select order_no from spd3_herp_test2.mcms_operation_apply where branch_id='%s' order by create_time desc LIMIT 1;"%branch_id)
operationNoticeBillId=get_id("select id from spd3_herp_test2.mcms_operation_notice where branch_id='%s' order by create_time desc LIMIT 1;"%branch_id)
operationNoticeOrderNo=get_id("select order_no from spd3_herp_test2.mcms_operation_notice where branch_id='%s' order by create_time desc LIMIT 1;"%branch_id)
operationApplyBillId=get_id("select id from spd3_herp_test2.mcms_operation_apply where branch_id='%s' order by create_time desc LIMIT 1;"%branch_id)
bill_relation_json = get_id(
"select bill_relation_json from hdi_distr a where id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
distr_main_key, type))
print(bill_relation_json)
purBillDate = reGetString(bill_relation_json, r'purBillDate":"', r'""distrBillId')
"select last_modified from hdi_distr where branch_id = '%s' and bill_mode='%s' order by create_time desc LIMIT 1;" % (
branch_id, type))
operationApplyOrderNo = get_id(
"select order_no from spd3_herp_test2.mcms_operation_apply where branch_id='%s' order by create_time desc LIMIT 1;" % branch_id)
operationNoticeBillId = get_id(
"select id from spd3_herp_test2.mcms_operation_notice where branch_id='%s' order by create_time desc LIMIT 1;" % branch_id)
operationNoticeOrderNo = get_id(
"select order_no from spd3_herp_test2.mcms_operation_notice where branch_id='%s' order by create_time desc LIMIT 1;" % branch_id)
operationApplyBillId = get_id(
"select id from spd3_herp_test2.mcms_operation_apply where branch_id='%s' order by create_time desc LIMIT 1;" % branch_id)
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'hdi_distr_0911', 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, operationApplyOrderNo, operationNoticeBillId, distrBillId, operationNoticeOrderNo,
distrOrderNo, operationApplyBillId,distrUser, create_user, create_user_name,
rec_org_name, operationApplyOrderNo, operationNoticeBillId, distrBillId,
operationNoticeOrderNo,
distrOrderNo, operationApplyBillId, distrUser, create_user,
create_user_name,
create_time, last_modified,
last_modified_user
)
actual_value1 = get_process_list2(actual)
expected_value1 = get_process_list2(expected)
actual_value2 = actual_value1.replace(" ", "")
......@@ -458,7 +466,7 @@ def check_hdi_distr(type=25):
def check_hdi_distr_detail(distr_main_key, type):
branch_id=get_branch_id()
branch_id = get_branch_id()
sql = "select * from hdi_distr_detail where pid = '%s';" % distr_main_key
# 获取实际值
actual = check_mcms_pur_plan_sql(sql)
......@@ -467,7 +475,8 @@ def check_hdi_distr_detail(distr_main_key, type):
distr_detail_main_key = get_id(
"select id from hdi_distr_detail where pid = '%s';" % (
distr_main_key))
source_id = get_id("select id from mcms_operation_notice where branch_id='%s' order by create_time desc LIMIT 1;"% (
source_id = get_id(
"select id from mcms_operation_notice where branch_id='%s' order by create_time desc LIMIT 1;" % (
branch_id))
source_detail_id = get_id("select id from mcms_operation_notice_list order by create_time desc LIMIT 1;")
pur_detail_id = source_detail_id
......@@ -512,19 +521,19 @@ def check_hdi_distr_detail(distr_main_key, type):
"select last_modified from hdi_distr_detail where id = '%s';" % (
distr_detail_main_key))
last_modified_user = create_user
operationNoticeBillId=source_id
operationNoticeBillDetailId=source_detail_id
operationApplyBillDetailId=source_detail_id
operationNoticeBillId = source_id
operationNoticeBillDetailId = source_detail_id
operationApplyBillDetailId = source_detail_id
expected = commonFuc().get_business_data('b6_spd3_core_business_flow_database_check',
'hdi_distr_detail_0911', distr_detail_main_key, distr_main_key, source_id,
source_detail_id, pur_detail_id
, hos_goods_id, mdm_goods_code, goods_code, goods_name, goods_gg,
mfrs_id, expdt_date, product_date
, pkg_def_id, supply_id, sub_prov_id, sub_prov_name, operationNoticeBillId,
distrDetailId, distrBillId, operationNoticeBillDetailId, operationApplyBillDetailId,
create_user,
create_time, last_modified, last_modified_user)
'hdi_distr_detail_0911', distr_detail_main_key, distr_main_key, source_id,
source_detail_id, pur_detail_id
, hos_goods_id, mdm_goods_code, goods_code, goods_name, goods_gg,
mfrs_id, expdt_date, product_date
, pkg_def_id, supply_id, sub_prov_id, sub_prov_name, operationNoticeBillId,
distrDetailId, distrBillId, operationNoticeBillDetailId,
operationApplyBillDetailId,
create_user,
create_time, last_modified, last_modified_user)
# comparison_result(actual, expected)
actual_value1 = get_process_list2(actual)
......@@ -544,11 +553,14 @@ def check_hdi_distr_detail(distr_main_key, type):
else:
print('error')
commonFuc().check_text_exist_result_text('error', 'succees')
pro_path = commonFuc().get_pro_path2()
print('pro_path', pro_path)
sys.path.append(pro_path + r'/air_case/b5_spd3_core_business_flow/a_b1流程八_高值跟台业务流程.air')
using(pro_path + r'//air_case/b5_spd3_core_business_flow/a_b1流程八_高值跟台业务流程.air')
from a_b1流程八_高值跟台业务流程 import operationApply,second_dept_in_warehouse,out_return_gt
from a_b1流程八_高值跟台业务流程 import operationApply, second_dept_in_warehouse, out_return_gt
try:
# set_dept_lead() # 设置二级科室负责人
# # # # # 设置默认科室,切换到二级科室,使用科室管理功能
......@@ -571,7 +583,7 @@ try:
# orderno = Search_OrderNO() # 手术通知单接口查询单号
# height_value_order_dp(orderno) # 高值跟台外网配送
# time.sleep(6)
check_hdi_distr() #外网配送单数据库检查
check_hdi_distr() # 外网配送单数据库检查
check_accept_order() # 内网验收
updateUserLoginDefaultRange(get_listUserMgrRangePage(2)) # 设置登录账号默认权限是二级库
second_dept_in_warehouse() # 二级科室入库
......@@ -582,4 +594,3 @@ except Exception as e:
# 打印错误信息
print(f"发生错误: {e}")
commonFuc().check_text_exist_result_text('error', 'succees')
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