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
1a6e8495
Commit
1a6e8495
authored
Apr 14, 2025
by
liguangyu06
Browse files
增加多线程
parent
0923a292
Changes
1
Hide whitespace changes
Inline
Side-by-side
air_case/b2_herp3_bs/生成结算数据.air/生成结算数据.py
View file @
1a6e8495
...
@@ -9,6 +9,7 @@ from common.db.sql.sql_del_kind_goodsinfo import delKindGoodsinfo
...
@@ -9,6 +9,7 @@ from common.db.sql.sql_del_kind_goodsinfo import delKindGoodsinfo
from
common.db.sql.sql_del_notices_info
import
delByNoticeName
from
common.db.sql.sql_del_notices_info
import
delByNoticeName
from
common.fileUtls
import
FileUtils
from
common.fileUtls
import
FileUtils
from
common.timeUtils
import
timeUtils
from
common.timeUtils
import
timeUtils
import
threading
__author__
=
"xiaohesheng"
__author__
=
"xiaohesheng"
...
@@ -25,106 +26,113 @@ import json
...
@@ -25,106 +26,113 @@ import json
module
=
"b2_herp3_bs"
module
=
"b2_herp3_bs"
u_id
=
str
(
uuid
.
uuid4
())
def
c_pur
():
u_random
=
random
.
randint
(
1000
,
100000000000000
)
u_id
=
str
(
uuid
.
uuid4
())
u_random
=
random
.
randint
(
1000
,
100000000000000
)
# 内网登录
username
=
commonFuc
().
get_business_data
(
module
,
"username411"
)
## print('医院名称',username)
password
=
commonFuc
().
get_business_data
(
module
,
"password411"
)
# print(username, password,'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')
# 获取token和projectCode
token
,
projectCode
,
uxid
,
corpId
,
info
=
login_system
(
username
,
password
).
get_token
()
# print( token,projectCode,uxid,corpId,info)
headers
=
commonFuc
().
get_business_data
(
module
,
"json_headers_pur"
,
token
)
# 第二步发起站点资料变更
url
=
commonFuc
().
get_api_url
()
+
commonFuc
().
get_business_data
(
module
,
"pur_get_goods_url"
)
# print(url)
request_body
=
commonFuc
().
get_business_data
(
module
,
"payload_pur"
)
"""
场景: 站点资料申请变更功能验证
用例名称:站点资料申请变更功能验证
输出:{"success":true,"code":"200","message":"OK","data":"success"}
"""
# 内网登录
username
=
commonFuc
().
get_business_data
(
module
,
"username411"
)
## print('医院名称',username)
password
=
commonFuc
().
get_business_data
(
module
,
"password411"
)
# print(username, password,'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')
# 获取token和projectCode
token
,
projectCode
,
uxid
,
corpId
,
info
=
login_system
(
username
,
password
).
get_token
()
# print( token,projectCode,uxid,corpId,info)
headers
=
commonFuc
().
get_business_data
(
module
,
"json_headers_pur"
,
token
)
print
(
headers
)
# 第二步发起站点资料变更
url
=
commonFuc
().
get_api_url
()
+
commonFuc
().
get_business_data
(
module
,
"pur_get_goods_url"
)
print
(
url
)
request_body
=
commonFuc
().
get_business_data
(
module
,
"payload_pur"
)
"""
场景: 站点资料申请变更功能验证
用例名称:站点资料申请变更功能验证
输出:{"success":true,"code":"200","message":"OK","data":"success"}
"""
# 发送请求
result
=
requests
.
post
(
url
,
json
=
request_body
,
headers
=
headers
)
# 获取接口响应时间
result
=
json
.
loads
(
result
.
content
)
print
(
len
(
result
[
"data"
][
"data"
]))
# len = random.randint(0, len(result["data"]["data"]) - 1)
# # 获取产品信息
# detail_list = result["data"]["data"][len]
# print(detail_list["hosGoodsCode"])
for
i
in
result
[
"data"
][
"data"
]:
pkg_def_id
=
i
[
"pkgDefVOList"
][
0
][
"pkgDefId"
]
print
(
i
[
"hosGoodsCode"
])
i
[
"inSettlement"
]
=
1
# pkg_def_id = result["data"]["data"][0]["pkgDefVOList"][0]["pkgDefId"]
# detail_list["inSettlement"] = 1
initStockBatchVOS
=
[
{
"batchCode"
:
u_random
,
"productDate"
:
"2025-04-01"
,
"expdtDate"
:
"2029-04-30"
,
"stockQty"
:
1
,
"viewId"
:
u_id
,
"pkgDefId"
:
pkg_def_id
,
"pkgDefName"
:
"大包"
,
"pkgDefQty"
:
20
}
]
batchVOS
=
[
{
"batchCode"
:
u_random
,
"productDate"
:
"2025-04-01"
,
"expdtDate"
:
"2029-04-30"
,
"stockQty"
:
1
,
"viewId"
:
u_id
,
"pkgDefId"
:
pkg_def_id
,
"pkgDefName"
:
"大包"
,
"pkgDefQty"
:
20
}
]
i
[
"initStockBatchVOS"
]
=
initStockBatchVOS
i
[
"batchVOS"
]
=
batchVOS
i
[
"stockQty"
]
=
1
json1
=
{
"areaCode"
:
"area0379"
,
"deptId"
:
"391d5e06d0ba4cb39a3a7ee2e7934ae7"
,
"provId"
:
"p00002a0c"
,
"purMode"
:
"60"
,
"remark"
:
""
,
"sourceType"
:
3
,
"stockId"
:
"b278f988a25c4678a1448fb08c3b4a41"
,
"subProvId"
:
"p00002a0c"
,
"attachmentList"
:
[]
}
json1
[
"detailVOS"
]
=
[
i
]
payload_pur
=
json1
# 手工入库
url
=
commonFuc
().
get_api_url
()
+
commonFuc
().
get_business_data
(
module
,
"pur_submit_url"
)
request_body
=
json1
# 发送请求
# 发送请求
result
=
requests
.
post
(
url
,
json
=
request_body
,
headers
=
headers
)
result
=
requests
.
post
(
url
,
json
=
request_body
,
headers
=
headers
)
# 获取接口响应时间
# 获取接口响应时间
result
=
json
.
loads
(
result
.
content
)
result
=
json
.
loads
(
result
.
content
)
print
(
result
)
# print(len(result["data"]["data"]))
# len = random.randint(0, len(result["data"]["data"]) - 1)
# # 获取产品信息
# detail_list = result["data"]["data"][len]
# print(detail_list["hosGoodsCode"])
for
i
in
result
[
"data"
][
"data"
]:
pkg_def_id
=
i
[
"pkgDefVOList"
][
0
][
"pkgDefId"
]
# print(i["hosGoodsCode"])
i
[
"inSettlement"
]
=
1
# pkg_def_id = result["data"]["data"][0]["pkgDefVOList"][0]["pkgDefId"]
# detail_list["inSettlement"] = 1
initStockBatchVOS
=
[
{
"batchCode"
:
u_random
,
"productDate"
:
"2025-04-01"
,
"expdtDate"
:
"2029-04-30"
,
"stockQty"
:
1
,
"viewId"
:
u_id
,
"pkgDefId"
:
pkg_def_id
,
"pkgDefName"
:
"大包"
,
"pkgDefQty"
:
20
}
]
batchVOS
=
[
{
"batchCode"
:
u_random
,
"productDate"
:
"2025-04-01"
,
"expdtDate"
:
"2029-04-30"
,
"stockQty"
:
1
,
"viewId"
:
u_id
,
"pkgDefId"
:
pkg_def_id
,
"pkgDefName"
:
"大包"
,
"pkgDefQty"
:
20
}
]
i
[
"initStockBatchVOS"
]
=
initStockBatchVOS
i
[
"batchVOS"
]
=
batchVOS
i
[
"stockQty"
]
=
1
json1
=
{
"areaCode"
:
"area0379"
,
"deptId"
:
"391d5e06d0ba4cb39a3a7ee2e7934ae7"
,
"provId"
:
"p00002a0c"
,
"purMode"
:
"60"
,
"remark"
:
""
,
"sourceType"
:
3
,
"stockId"
:
"b278f988a25c4678a1448fb08c3b4a41"
,
"subProvId"
:
"p00002a0c"
,
"attachmentList"
:
[]
}
json1
[
"detailVOS"
]
=
[
i
]
payload_pur
=
json1
# 手工入库
url
=
commonFuc
().
get_api_url
()
+
commonFuc
().
get_business_data
(
module
,
"pur_submit_url"
)
request_body
=
json1
# 发送请求
result
=
requests
.
post
(
url
,
json
=
request_body
,
headers
=
headers
)
# 获取接口响应时间
result
=
json
.
loads
(
result
.
content
)
# print(result)
threads
=
[]
for
_
in
range
(
10
):
t
=
threading
.
Thread
(
target
=
c_pur
)
threads
.
append
(
t
)
t
.
start
()
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