Skip to content
GitLab
Projects
Groups
Snippets
/
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Sign in / Register
Toggle navigation
Menu
Open sidebar
test
autotest-airtest-web-spd
Commits
cae6b4e3
Commit
cae6b4e3
authored
Jul 16, 2024
by
xiao-hesheng
Browse files
数据库验证流程十一脚本编写
parent
bd4f8d84
Changes
2
Hide whitespace changes
Inline
Side-by-side
air_case/b6_spd3_core_business_flow_database_check/a_b4流程十一请领业务流_二级库向直送_数据库验证.air/a_b4流程十一请领业务流_二级库向直送_数据库验证.py
View file @
cae6b4e3
...
...
@@ -101,7 +101,7 @@ def check_mcms_dept_buy_car(type=1):
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查二级库请领单审批前的子表数据
def
mcms_dept_buy_detail
(
type
=
16
):
branch_id
=
get_branch_id
()
# 获取主表主键
...
...
@@ -182,7 +182,7 @@ def mcms_dept_buy_detail(type=16):
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查二级库请领单审批前的主表数据
def
check_mcms_dept_buy
(
type
=
16
):
branch_id
=
get_branch_id
()
sql
=
"""select * from mcms_dept_buy where branch_id='%s' and bill_mode='%s' order by create_time desc LIMIT 1;
...
...
@@ -285,7 +285,7 @@ def check_mcms_dept_buy(type=16):
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查二级库请领单审批后的主表数据
def
check_mcms_dept_buy_approval_after
(
type
=
16
):
branch_id
=
get_branch_id
()
sql
=
"""select * from mcms_dept_buy where branch_id='%s' and bill_mode='%s' order by create_time desc LIMIT 1;
...
...
@@ -390,7 +390,7 @@ def check_mcms_dept_buy_approval_after(type=16):
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查二级库请领单审批后的子表数据
def
mcms_dept_buy_detail_approval_after
(
type
=
16
):
branch_id
=
get_branch_id
()
# 获取主表主键
...
...
@@ -474,7 +474,7 @@ def mcms_dept_buy_detail_approval_after(type=16):
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查采购计划主表
def
check_mcms_mcms_pur_plan
(
type
=
16
):
branch_id
=
get_branch_id
()
# 获取请领单主键id
...
...
@@ -574,7 +574,7 @@ def check_mcms_mcms_pur_plan(type=16):
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查采购计划子表
def
check_mcms_mcms_pur_plan_detail
(
id
,
type
,
planOrderNo
):
branch_id
=
get_branch_id
()
# 根据采购计划主表的id获取采购计划单子表主键
...
...
@@ -712,6 +712,241 @@ def check_mcms_mcms_pur_plan_detail(id,type,planOrderNo):
else
:
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
#检查采购订单
def
check_mcms_purchase
(
type
=
16
):
branch_id
=
get_branch_id
()
# 获取采购订单主键id
sql
=
"select id from mcms_purchase where branch_id='%s' and bill_mode='%s' order by create_time desc LIMIT 1;"
%
(
branch_id
,
type
)
print
(
sql
)
id1
=
process_tuple
(
get_dept_buy_id
(
sql
))
sql
=
"select order_no from mcms_purchase where branch_id='%s' and bill_mode='%s'and id='%s' order by create_time desc LIMIT 1;"
%
(
branch_id
,
type
,
id1
)
print
(
sql
)
buyOrderNo1
=
process_tuple
(
get_dept_buy_id
(
sql
))
# 要验证的数据sql
sql1
=
"select * from mcms_purchase where branch_id='%s' and bill_mode='%s' and id='%s' order by create_time desc LIMIT 1;"
%
(
branch_id
,
type
,
id1
)
# 获取实际值
actual
=
check_mcms_pur_plan_sql
(
sql1
)
print
(
'actual'
,
actual
)
hos_id
=
get_hosid
()
dept_id1
=
FileUtils
().
r_info8
(
"b2_herp3_bs"
,
"所有科室id"
,
'message6'
)[
'deptid1'
]
dept_name1
=
FileUtils
().
r_info8
(
"b2_herp3_bs"
,
"所有科室name"
,
'message6'
)[
'deptid1'
]
dept_name2
=
FileUtils
().
r_info8
(
"b2_herp3_bs"
,
"所有科室name"
,
'message6'
)[
'deptid2'
]
rec_org_id
=
process_tuple
(
get_id
(
"select id from sys_org where branch_id='%s' and ename='%s';"
%
(
branch_id
,
dept_name2
)))
rec_org_name
=
dept_name2
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
=
'默认库区'
source_id
=
id1
rec_user
=
get_login_user_uxid
()
rec_addr
=
'自动化测试'
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
,
id1
)
print
(
sql
)
bill_expdt_date
=
get_expdt_date
(
sql
)
buyBillId
=
id1
planBillId
=
id2
buyOrderNo
=
buyOrderNo1
create_user
=
rec_user
create_time
=
get_create_time
(
"select create_time from mcms_pur_plan where branch_id='%s' and source_id='%s' order by create_time desc LIMIT 1;"
%
(
branch_id
,
id1
))
last_modified
=
create_time
last_modified_user
=
create_user
planOrderNo
=
PLAN_order_no
if
type
==
16
:
# 验证子表数据
check_mcms_mcms_pur_plan_detail
(
id2
,
type
,
planOrderNo
)
expected
=
commonFuc
().
get_business_data
(
'b6_spd3_core_business_flow_database_check'
,
'mcms_pur_plan'
,
id2
,
PLAN_order_no
,
hos_id
,
get_branch_id
(),
dept_id1
,
dept_name1
,
rec_org_id
,
rec_org_name
,
area_code
,
area_name
,
source_id
,
rec_user
,
rec_addr
,
bill_expdt_date
,
buyBillId
,
planBillId
,
planOrderNo
,
buyOrderNo
,
create_user
,
create_time
,
last_modified
,
last_modified_user
)
elif
type
==
20
:
check_mcms_mcms_pur_plan_detail
(
id2
,
type
,
planOrderNo
)
expected
=
commonFuc
().
get_business_data
(
'b6_spd3_core_business_flow_database_check'
,
'mcms_pur_plan2'
,
id2
,
PLAN_order_no
,
hos_id
,
get_branch_id
(),
dept_id1
,
dept_name1
,
rec_org_id
,
rec_org_name
,
area_code
,
area_name
,
source_id
,
rec_user
,
rec_addr
,
bill_expdt_date
,
buyBillId
,
planBillId
,
planOrderNo
,
buyOrderNo
,
create_user
,
create_time
,
last_modified
,
last_modified_user
)
elif
type
==
66
:
check_mcms_mcms_pur_plan_detail
(
id2
,
type
,
planOrderNo
)
expected
=
commonFuc
().
get_business_data
(
'b6_spd3_core_business_flow_database_check'
,
'mcms_pur_plan3'
,
id2
,
PLAN_order_no
,
hos_id
,
get_branch_id
(),
dept_id1
,
dept_name1
,
rec_org_id
,
rec_org_name
,
area_code
,
area_name
,
source_id
,
rec_user
,
rec_addr
,
bill_expdt_date
,
buyBillId
,
planBillId
,
planOrderNo
,
buyOrderNo
,
create_user
,
create_time
,
last_modified
,
last_modified_user
)
else
:
expected
=
None
actual_value1
=
get_process_list2
(
actual
)
expected_value1
=
get_process_list2
(
expected
)
actual_value2
=
actual_value1
.
replace
(
" "
,
""
)
expected_value2
=
expected_value1
.
replace
(
" "
,
""
)
# print('actual_value2',actual_value2)
print
(
'expected_value2'
)
print
(
expected_value2
)
print
(
'actual_value2'
)
print
(
actual_value2
)
result
=
(
compare_text_index
(
actual_value2
,
expected_value2
))
print
(
'result'
,
result
)
if
actual_value2
==
expected_value2
:
print
(
'ok'
)
commonFuc
().
check_text_exist_result_text
(
'succees'
,
'succees'
)
else
:
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
def
check_mcms_purchase_detail
(
id
,
type
,
planOrderNo
):
branch_id
=
get_branch_id
()
# 根据采购计划主表的id获取采购计划单子表主键
sql
=
"select id from mcms_pur_plan_detail where pid='%s' order by create_time desc LIMIT 1;"
%
(
id
)
id2
=
process_tuple
(
get_dept_buy_id
(
sql
))
# 采购计划单子表主键
sql
=
"select order_no from mcms_pur_plan where id='%s' order by create_time desc LIMIT 1;"
%
(
id2
)
PLAN_order_no
=
process_tuple
(
get_dept_buy_id
(
sql
))
# 采购计划单子表order_no
# 要验证的数据sql
sql1
=
"select * from mcms_pur_plan_detail where id='%s' order by create_time desc LIMIT 1;"
%
(
id2
)
# 获取实际值
actual
=
check_mcms_pur_plan_sql
(
sql1
)
print
(
'actual'
,
actual
)
hos_id
=
get_hosid
()
dept_id1
=
FileUtils
().
r_info8
(
"b2_herp3_bs"
,
"所有科室id"
,
'message6'
)[
'deptid1'
]
dept_name1
=
FileUtils
().
r_info8
(
"b2_herp3_bs"
,
"所有科室name"
,
'message6'
)[
'deptid1'
]
dept_name2
=
FileUtils
().
r_info8
(
"b2_herp3_bs"
,
"所有科室name"
,
'message6'
)[
'deptid2'
]
rec_org_id
=
process_tuple
(
get_id
(
"select id from sys_org where branch_id='%s' and ename='%s';"
%
(
branch_id
,
dept_name2
)))
rec_user
=
get_login_user_uxid
()
create_user
=
rec_user
# last_modified = create_time
last_modified_user
=
create_user
pid
=
id
if
type
==
16
:
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
)))
# 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
)))
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
)))
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
create_time
=
get_create_time
(
"select create_time from mcms_pur_plan_detail where id='%s';"
%
(
id2
))
last_modified
=
create_time
sub_prov_name
=
prov_name
source_id
=
process_tuple
(
get_id
(
"select source_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';"
%
(
hos_goods_id
,
id2
)))
expected
=
commonFuc
().
get_business_data
(
'b6_spd3_core_business_flow_database_check'
,
'mcms_pur_plan_detail'
,
id2
,
pid
,
planOrderNo
,
source_id
,
hos_goods_id
,
mdm_goods_code
,
goods_code
,
pkg_def_id
,
supply_id
,
prov_id
,
prov_name
,
sub_prov_id
,
sub_prov_name
,
create_user
,
create_time
,
last_modified
,
last_modified_user
)
elif
type
==
20
:
hos_goods_id
=
FileUtils
().
r_info8
(
'b5_spd3_core_business_flow'
,
'产品审核信息'
,
'message1'
)[
"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
)))
# 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
)))
pkg_def_id
=
process_tuple
(
get_id
(
"select pkg_def_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';"
%
(
hos_goods_id
,
id2
)))
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
)))
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
create_time
=
get_create_time
(
"select create_time from mcms_pur_plan_detail where id='%s';"
%
(
id2
))
last_modified
=
create_time
sub_prov_name
=
prov_name
source_id
=
process_tuple
(
get_id
(
"select source_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';"
%
(
hos_goods_id
,
id2
)))
expected
=
commonFuc
().
get_business_data
(
'b6_spd3_core_business_flow_database_check'
,
'mcms_pur_plan_detail2'
,
id2
,
pid
,
planOrderNo
,
source_id
,
hos_goods_id
,
mdm_goods_code
,
goods_code
,
supply_id
,
prov_id
,
prov_name
,
sub_prov_id
,
sub_prov_name
,
create_user
,
create_time
,
last_modified
,
last_modified_user
)
else
:
hos_goods_id
=
FileUtils
().
r_info8
(
'b5_spd3_core_business_flow'
,
'产品审核信息'
,
'message3'
)[
"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
)))
# 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
)))
pkg_def_id
=
process_tuple
(
get_id
(
"select pkg_def_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';"
%
(
hos_goods_id
,
id2
)))
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
)))
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
create_time
=
get_create_time
(
"select create_time from mcms_pur_plan_detail where id='%s';"
%
(
id2
))
last_modified
=
create_time
sub_prov_name
=
prov_name
source_id
=
process_tuple
(
get_id
(
"select source_id from mcms_pur_plan_detail where hos_goods_id='%s' and id='%s';"
%
(
hos_goods_id
,
id2
)))
expected
=
commonFuc
().
get_business_data
(
'b6_spd3_core_business_flow_database_check'
,
'mcms_pur_plan_detail3'
,
id2
,
pid
,
planOrderNo
,
source_id
,
hos_goods_id
,
mdm_goods_code
,
goods_code
,
pkg_def_id
,
supply_id
,
prov_id
,
prov_name
,
sub_prov_id
,
sub_prov_name
,
create_user
,
create_time
,
last_modified
,
last_modified_user
)
actual_value1
=
get_process_list2
(
actual
)
expected_value1
=
get_process_list2
(
expected
)
actual_value2
=
actual_value1
.
replace
(
" "
,
""
)
expected_value2
=
expected_value1
.
replace
(
" "
,
""
)
# print('actual_value2',actual_value2)
print
(
'expected_value2'
)
print
(
expected_value2
)
print
(
'actual_value2'
)
print
(
actual_value2
)
result
=
(
compare_text_index
(
actual_value2
,
expected_value2
))
print
(
'result'
,
result
)
if
actual_value2
==
expected_value2
:
print
(
'ok'
)
commonFuc
().
check_text_exist_result_text
(
'succees'
,
'succees'
)
else
:
print
(
'error'
)
commonFuc
().
check_text_exist_result_text
(
'error'
,
'succees'
)
try
:
pro_path
=
commonFuc
().
get_pro_path2
()
print
(
'pro_path'
,
pro_path
)
...
...
@@ -751,15 +986,20 @@ try:
# 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_mcms_pur_plan
(
int
(
i
))
# 采购模块处理
purchase_module_process
()
for
i
in
range
(
3
):
# 三个单据共审批3次
approval_center
()
check_mcms_purchase
(
int
(
i
))
order_dp
(
1
)
# 1低值
order_dp
(
2
)
# 2高值
order_dp
(
3
)
# 3试剂
...
...
data/b6_spd3_core_business_flow_database_check/data
View file @
cae6b4e3
...
...
@@ -34,3 +34,6 @@
"mcms_pur_plan_detail": '{"id": "%s", "pid": "%s", "order_no": "%s", "source_id": "%s", "hos_goods_id": "%s", "mdm_goods_code": "%s", "goods_code": "%s", "pur_mode": 10, "pkg_def_id": "%s", "pkg_def_name": "100", "pkg_def_qty": Decimal("100.00"), "plan_pkg_qty": Decimal("1.00"), "plan_qty": Decimal("100.00"), "pkg_qty": Decimal("1.00"), "price": Decimal("99.999900"), "qty": Decimal("100.00"), "in_settlement": None, "supply_id": "%s", "prov_id": "%s", "prov_name": "%s", "sub_send_flag": 10, "sub_prov_id": "%s", "sub_prov_name": "%s", "contract_id": None, "remark": None, "close_status": 0, "tb_status": 10, "pur_task_id": None, "create_user": "%s", "create_time": %s, "last_modified": %s, "last_modified_user": "%s", "version": 0, "coding_type": 0}'
"mcms_pur_plan_detail2": '{"id": "%s", "pid": "%s", "order_no": "%s", "source_id": "%s", "hos_goods_id": "%s", "mdm_goods_code": "%s", "goods_code": "%s", "pur_mode": 20, "pkg_def_id": None, "pkg_def_name": None, "pkg_def_qty": Decimal("1.00"), "plan_pkg_qty": Decimal("1.00"), "plan_qty": Decimal("1.00"), "pkg_qty": Decimal("1.00"), "price": Decimal("99.999900"), "qty": Decimal("1.00"), "in_settlement": None, "supply_id": "%s", "prov_id": "%s", "prov_name": "%s", "sub_send_flag": 10, "sub_prov_id": "%s", "sub_prov_name": "%s", "contract_id": None, "remark": None, "close_status": 0, "tb_status": 10, "pur_task_id": None, "create_user": "%s", "create_time": %s, "last_modified": %s, "last_modified_user": "%s", "version": 0, "coding_type": 0}'
"mcms_pur_plan_detail3": '{"id": "%s", "pid": "%s", "order_no": "%s", "source_id": "%s", "hos_goods_id": "%s", "mdm_goods_code": "%s", "goods_code": "%s", "pur_mode": 60, "pkg_def_id": "%s", "pkg_def_name": "100", "pkg_def_qty": Decimal("100.00"), "plan_pkg_qty": Decimal("1.00"), "plan_qty": Decimal("100.00"), "pkg_qty": Decimal("1.00"), "price": Decimal("99.999900"), "qty": Decimal("100.00"), "in_settlement": None, "supply_id": "%s", "prov_id": "%s", "prov_name": "%s", "sub_send_flag": 10, "sub_prov_id": "%s", "sub_prov_name": "%s", "contract_id": None, "remark": None, "close_status": 0, "tb_status": 10, "pur_task_id": None, "create_user": "%s", "create_time": %s, "last_modified": %s, "last_modified_user": "%s", "version": 0, "coding_type": 0}'
"mcms_purchase": '{"id": "Ch034720240716000019", "order_no": "CG2024071621466", "bill_mode": "16", "rec_type": "HDI", "hos_id": "h0347", "branch_id": "h0347-3755", "dept_id": "63c734f07fcc422ea80f49ca8e5bbd95", "dept_name": "设备科001", "rec_org_id": "b12c7c6ca56e4b46b418b319cbadbea5", "rec_org_name": "二级检验科002", "area_code": "areah03472781", "area_name": "默认库区", "bill_expdt_date": datetime.datetime(2024, 8, 16, 0, 0), "source_id": "Ph034720240716029571", "source_type": "CGJH", "in_settlement": None, "rec_user": "自动化测试w&vqF", "rec_phone": "None", "rec_addr": "自动化测试", "prov_id": "p1e35e", "prov_name": "国药集团公司S57ui", "sub_send_flag": 10, "exe_prov_id": "p1e35e", "exe_prov_name": "国药集团公司S57ui", "bill_relation_json": "{"buyBillId":"Bh03472024071600022","planBillId":"Ph034720240716029571","planOrderNo":"PLAN2024071632926","buyOrderNo":"BUY2024071609399","purOrderNo":"CG2024071621466","purBillId":"Ch034720240716000019"}", "ext_info": "{"epidemicFlag":0,"purMode":10}", "barcode_flag": 1, "pkg_flag": 1, "remark": "", "priority": 0, "close_status": 0, "close_date": datetime.datetime(2024, 7, 17, 15, 20, 54), "up_status": 0, "tb_status": 30, "create_user": "h0347_UID-051420", "create_time": datetime.datetime(2024, 7, 16, 15, 20, 54), "last_modified": datetime.datetime(2024, 7, 16, 15, 20, 54), "last_modified_user": None, "version": 0, "del_flag": 0, "last_auditor": "自动化测试w&vqF", "last_audit_time": datetime.datetime(2024, 7, 16, 15, 20, 54)}'
\ No newline at end of file
Write
Preview
Supports
Markdown
0%
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment