feat: 添加JoyHub运费模板和Banner管理接口用例

This commit is contained in:
2026-05-06 10:54:03 +08:00
parent cc6733a8fb
commit 86f4e8288e
89 changed files with 11557 additions and 3 deletions

View File

@@ -0,0 +1,176 @@
import logging
import allure
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_log = logging.getLogger("logger")
class BannerManage(DlzhanInterface):
def __init__(self):
super().__init__()
@allure.step("创建Banner")
def kw_joyhub_banner_create_post(self, position, platform, lang, rank_num, banner_type, status, interval_time, id=0, title="", sub_title="", image=None, link="", cover_image=None):
"""
创建Banner管理业务关键字
:param id: 主键新增为0
:param position: 位置
:param title: 标题
:param sub_title: 副标题
:param image: 图片,格式{'url': 'https://xxx', 'name': 'xxx', 'mime_type': 'image/webp'}
:param link: 链接
:param platform: 适用平台(1PC 2手机)
:param lang: 语言(en 英语 de 德语 ja 日语)
:param rank_num: 排序号
:param banner_type: 类型
:param status: 状态 1-启用2-停用
:param interval_time: 轮播时间(秒)
:param cover_image: 缩略图
:return: 响应结果
"""
obj_log.info(f"创建Banner - position: {position}, platform: {platform}, lang: {lang}, rankNum: {rank_num}, type: {banner_type}, status: {status}")
if image is None:
image = {'url': 'https://www.toendi.com/static/image/cd94c191561c4a37a04c78fca2913851.webp', 'name': 'Test Banner Image', 'mime_type': 'image/webp'}
params = {
"id": id,
"position": position,
"title": title,
"subTitle": sub_title,
"image": image,
"link": link,
"platform": platform,
"lang": lang,
"rankNum": rank_num,
"type": banner_type,
"status": status,
"intervalTime": interval_time,
"coverImage": cover_image
}
resp = self.kw_in_joyhub_banner_create_post(**params)
obj_log.info(f"创建Banner响应: {resp}")
return resp
@allure.step("删除Banner")
def kw_joyhub_banner_delete_delete(self, banner_id):
"""
删除Banner管理业务关键字
:param banner_id: Banner编号
:return: 响应结果
"""
obj_log.info(f"删除Banner - id: {banner_id}")
resp = self.kw_in_joyhub_banner_delete_delete(banner_id=banner_id)
obj_log.info(f"删除Banner响应: {resp}")
return resp
@allure.step("批量删除Banner")
def kw_joyhub_banner_delete_list_delete(self, banner_ids):
"""
批量删除Banner管理业务关键字
:param banner_ids: Banner编号列表
:return: 响应结果
"""
obj_log.info(f"批量删除Banner - ids: {banner_ids}")
resp = self.kw_in_joyhub_banner_delete_list_delete(ids=banner_ids)
obj_log.info(f"批量删除Banner响应: {resp}")
return resp
@allure.step("获得Banner详情")
def kw_joyhub_banner_get_get(self, banner_id):
"""
获得Banner管理详情业务关键字
:param banner_id: Banner编号
:return: 响应结果
"""
obj_log.info(f"获得Banner详情 - id: {banner_id}")
resp = self.kw_in_joyhub_banner_get_get(banner_id=banner_id)
obj_log.info(f"获得Banner详情响应: {resp}")
return resp
@allure.step("获得Banner分页列表")
def kw_joyhub_banner_page_get(self, page_no=1, page_size=10, **kwargs):
"""
获得Banner管理分页业务关键字
:param page_no: 页码
:param page_size: 每页条数
:param position: 位置
:param title: 标题
:param sub_title: 副标题
:param platform: 适用平台(1PC 2手机)
:param lang: 语言
:param banner_type: 类型
:param status: 状态
:return: 响应结果
"""
obj_log.info(f"获得Banner分页列表 - pageNo: {page_no}, pageSize: {page_size}")
params = {
"pageNo": page_no,
"pageSize": page_size,
"position": kwargs.get("position", ""),
"title": kwargs.get("title", ""),
"subTitle": kwargs.get("sub_title", ""),
"platform": kwargs.get("platform", ""),
"lang": kwargs.get("lang", ""),
"type": kwargs.get("banner_type", ""),
"status": kwargs.get("status", "")
}
resp = self.kw_in_joyhub_banner_page_get(**params)
obj_log.info(f"获得Banner分页列表响应: {resp}")
return resp
@allure.step("更新Banner")
def kw_joyhub_banner_update_put(self, banner_id, position, platform, lang, rank_num, banner_type, status, interval_time, title="", sub_title="", image=None, link="", cover_image=None):
"""
更新Banner管理业务关键字
:param banner_id: 主键
:param position: 位置
:param title: 标题
:param sub_title: 副标题
:param image: 图片
:param link: 链接
:param platform: 适用平台(1PC 2手机)
:param lang: 语言(en 英语 de 德语 ja 日语)
:param rank_num: 排序号
:param banner_type: 类型
:param status: 状态 1-启用2-停用
:param interval_time: 轮播时间(秒)
:param cover_image: 缩略图
:return: 响应结果
"""
obj_log.info(f"更新Banner - id: {banner_id}, position: {position}, platform: {platform}, lang: {lang}, rankNum: {rank_num}, type: {banner_type}, status: {status}")
if image is None:
image = {'url': 'https://www.toendi.com/static/image/cd94c191561c4a37a04c78fca2913851.webp', 'name': 'Updated Banner Image', 'mime_type': 'image/webp'}
params = {
"id": banner_id,
"position": position,
"title": title,
"subTitle": sub_title,
"image": image,
"link": link,
"platform": platform,
"lang": lang,
"rankNum": rank_num,
"type": banner_type,
"status": status,
"intervalTime": interval_time,
"coverImage": cover_image
}
resp = self.kw_in_joyhub_banner_update_put(**params)
obj_log.info(f"更新Banner响应: {resp}")
return resp

View File

@@ -0,0 +1,174 @@
import logging
import allure
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_log = logging.getLogger("logger")
class DeptManage(DlzhanInterface):
def __init__(self):
super().__init__()
@allure.step("创建部门")
def kw_joyhub_dept_create_post(self, name, sort, status=1, parent_id=None, leader_user_id=None, phone=None, email=None, code=None):
"""
创建部门业务关键字
:param name: 部门名称
:param sort: 显示顺序
:param status: 状态
:param parent_id: 父部门ID
:param leader_user_id: 负责人用户编号
:param phone: 联系电话
:param email: 邮箱
:param code: 部门编码
:return: 响应结果
"""
obj_log.info(f"创建部门 - name: {name}, sort: {sort}, status: {status}")
params = {
"name": name,
"sort": sort,
"status": status
}
if parent_id is not None:
params["parentId"] = parent_id
if leader_user_id is not None:
params["leaderUserId"] = leader_user_id
if phone:
params["phone"] = phone
if email:
params["email"] = email
if code:
params["code"] = code
resp = self.kw_in_joyhub_dept_create_post(**params)
obj_log.info(f"创建部门响应: {resp}")
return resp
@allure.step("删除部门")
def kw_joyhub_dept_delete_post(self, dept_id):
"""
删除部门业务关键字
:param dept_id: 部门编号
:return: 响应结果
"""
obj_log.info(f"删除部门 - id: {dept_id}")
resp = self.kw_in_joyhub_dept_delete_post(dept_id)
obj_log.info(f"删除部门响应: {resp}")
return resp
@allure.step("批量删除部门")
def kw_joyhub_dept_delete_list_post(self, ids):
"""
批量删除部门业务关键字
:param ids: 部门编号列表
:return: 响应结果
"""
obj_log.info(f"批量删除部门 - ids: {ids}")
resp = self.kw_in_joyhub_dept_delete_list_post(ids)
obj_log.info(f"批量删除部门响应: {resp}")
return resp
@allure.step("获得部门信息")
def kw_joyhub_dept_get_get(self, dept_id):
"""
获得部门信息业务关键字
:param dept_id: 部门编号
:return: 响应结果
"""
obj_log.info(f"获得部门信息 - id: {dept_id}")
resp = self.kw_in_joyhub_dept_get_get(dept_id)
obj_log.info(f"获得部门信息响应: {resp}")
return resp
@allure.step("获取部门列表")
def kw_joyhub_dept_list_get(self, name=None, status=None):
"""
获取部门列表业务关键字
:param name: 部门名称(模糊匹配)
:param status: 状态
:return: 响应结果
"""
obj_log.info(f"获取部门列表 - name: {name}, status: {status}")
params = {}
if name:
params["name"] = name
if status is not None:
params["status"] = status
resp = self.kw_in_joyhub_dept_list_get(**params)
obj_log.info(f"获取部门列表响应: {resp}")
return resp
@allure.step("获取部门精简信息列表")
def kw_joyhub_dept_list_all_simple_get(self):
"""
获取部门精简信息列表业务关键字
:return: 响应结果
"""
obj_log.info("获取部门精简信息列表")
resp = self.kw_in_joyhub_dept_list_all_simple_get()
obj_log.info(f"获取部门精简信息列表响应: {resp}")
return resp
@allure.step("获取部门精简信息列表")
def kw_joyhub_dept_simple_list_get(self):
"""
获取部门精简信息列表业务关键字(另一个接口)
:return: 响应结果
"""
obj_log.info("获取部门精简信息列表(simple-list)")
resp = self.kw_in_joyhub_dept_simple_list_get()
obj_log.info(f"获取部门精简信息列表响应: {resp}")
return resp
@allure.step("更新部门")
def kw_joyhub_dept_update_put(self, dept_id, name, sort, status=1, parent_id=None, leader_user_id=None, phone=None, email=None):
"""
更新部门业务关键字
:param dept_id: 部门编号
:param name: 部门名称
:param sort: 显示顺序
:param status: 状态
:param parent_id: 父部门ID
:param leader_user_id: 负责人用户编号
:param phone: 联系电话
:param email: 邮箱
:return: 响应结果
"""
obj_log.info(f"更新部门 - id: {dept_id}, name: {name}, sort: {sort}, status: {status}")
params = {
"id": dept_id,
"name": name,
"sort": sort,
"status": status
}
if parent_id is not None:
params["parentId"] = parent_id
if leader_user_id is not None:
params["leaderUserId"] = leader_user_id
if phone:
params["phone"] = phone
if email:
params["email"] = email
resp = self.kw_in_joyhub_dept_update_put(**params)
obj_log.info(f"更新部门响应: {resp}")
return resp

View File

@@ -0,0 +1,197 @@
import logging
import allure
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_log = logging.getLogger("logger")
class PostManage(DlzhanInterface):
def __init__(self):
super().__init__()
@allure.step("创建岗位")
def kw_joyhub_post_create_post(self, name, code, sort, status, remark=""):
"""
创建岗位业务关键字
:param name: 岗位名称
:param code: 岗位编码
:param sort: 显示顺序
:param status: 状态
:param remark: 备注
:return: 响应结果
"""
obj_log.info(f"创建岗位 - name: {name}, code: {code}, sort: {sort}, status: {status}, remark: {remark}")
params = {
"name": name,
"code": code,
"sort": sort,
"status": status
}
if remark:
params["remark"] = remark
resp = self.kw_in_joyhub_post_create_post(**params)
obj_log.info(f"创建岗位响应: {resp}")
return resp
@allure.step("删除岗位")
def kw_joyhub_post_delete_post(self, post_id):
"""
删除岗位业务关键字
:param post_id: 岗位编号
:return: 响应结果
"""
obj_log.info(f"删除岗位 - id: {post_id}")
resp = self.kw_in_joyhub_post_delete_post(id=post_id)
obj_log.info(f"删除岗位响应: {resp}")
return resp
@allure.step("批量删除岗位")
def kw_joyhub_post_delete_list_post(self, ids):
"""
批量删除岗位业务关键字
:param ids: 岗位编号列表
:return: 响应结果
"""
obj_log.info(f"批量删除岗位 - ids: {ids}")
resp = self.kw_in_joyhub_post_delete_list_post(ids=ids)
obj_log.info(f"批量删除岗位响应: {resp}")
return resp
@allure.step("岗位管理导出")
def kw_joyhub_post_export_get(self, pageNo=1, pageSize=10, code="", name="", status=None):
"""
岗位管理导出业务关键字
:param pageNo: 页码
:param pageSize: 每页条数
:param code: 岗位编码(模糊匹配)
:param name: 岗位名称(模糊匹配)
:param status: 状态
:return: 响应结果
"""
obj_log.info(f"岗位管理导出 - pageNo: {pageNo}, pageSize: {pageSize}, code: {code}, name: {name}, status: {status}")
params = {
"pageNo": pageNo,
"pageSize": pageSize
}
if code:
params["code"] = code
if name:
params["name"] = name
if status is not None:
params["status"] = status
resp = self.kw_in_joyhub_post_export_get(**params)
obj_log.info(f"岗位管理导出响应: {resp}")
return resp
@allure.step("获得岗位信息")
def kw_joyhub_post_get_get(self, post_id):
"""
获得岗位信息业务关键字
:param post_id: 岗位编号
:return: 响应结果
"""
obj_log.info(f"获得岗位信息 - id: {post_id}")
resp = self.kw_in_joyhub_post_get_get(id=post_id)
obj_log.info(f"获得岗位信息响应: {resp}")
return resp
@allure.step("获取岗位全列表")
def kw_joyhub_post_list_all_simple_get(self):
"""
获取岗位全列表业务关键字
:return: 响应结果
"""
obj_log.info("获取岗位全列表")
resp = self.kw_in_joyhub_post_list_all_simple_get()
obj_log.info(f"获取岗位全列表响应: {resp}")
return resp
@allure.step("获得岗位分页列表")
def kw_joyhub_post_page_get(self, pageNo=1, pageSize=10, code="", name="", status=None):
"""
获得岗位分页列表业务关键字
:param pageNo: 页码
:param pageSize: 每页条数
:param code: 岗位编码(模糊匹配)
:param name: 岗位名称(模糊匹配)
:param status: 状态
:return: 响应结果
"""
obj_log.info(f"获得岗位分页列表 - pageNo: {pageNo}, pageSize: {pageSize}, code: {code}, name: {name}, status: {status}")
params = {
"pageNo": pageNo,
"pageSize": pageSize
}
if code:
params["code"] = code
if name:
params["name"] = name
if status is not None:
params["status"] = status
resp = self.kw_in_joyhub_post_page_get(**params)
obj_log.info(f"获得岗位分页列表响应: {resp}")
return resp
@allure.step("获取岗位精简列表")
def kw_joyhub_post_simple_list_get(self):
"""
获取岗位精简列表业务关键字
:return: 响应结果
"""
obj_log.info("获取岗位精简列表")
resp = self.kw_in_joyhub_post_simple_list_get()
obj_log.info(f"获取岗位精简列表响应: {resp}")
return resp
@allure.step("修改岗位")
def kw_joyhub_post_update_put(self, name, code, sort, status, post_id=None, remark=""):
"""
修改岗位业务关键字
:param name: 岗位名称
:param code: 岗位编码
:param sort: 显示顺序
:param status: 状态
:param post_id: 岗位编号(可选,更新时需要)
:param remark: 备注
:return: 响应结果
"""
obj_log.info(f"修改岗位 - id: {post_id}, name: {name}, code: {code}, sort: {sort}, status: {status}, remark: {remark}")
params = {
"name": name,
"code": code,
"sort": sort,
"status": status
}
if post_id:
params["id"] = post_id
if remark:
params["remark"] = remark
resp = self.kw_in_joyhub_post_update_put(**params)
obj_log.info(f"修改岗位响应: {resp}")
return resp

View File

@@ -0,0 +1,351 @@
# -*- coding:utf-8 -*-
"""
Author: Auto Generated
Create Date: 2026/04/30
"""
import logging
import os
import sys
current_file_path = os.path.abspath(__file__)
project_root = os.path.abspath(os.path.join(os.path.dirname(current_file_path), '../../../../'))
if project_root not in sys.path:
sys.path.insert(0, project_root)
import time
from base_framework.public_tools import log
from base_framework.public_tools.my_faker import MyFaker
from base_framework.public_tools.runner import Runner
from base_framework.public_tools import utils
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_get_log = log.get_logger()
obj_my_faker = MyFaker()
obj_runner = Runner()
obj_get_way = utils.Tools()
class RoleManage(DlzhanInterface):
def __init__(self):
super().__init__()
joyhub_token = self._read_robot_variable("joyhub_login_token")
if joyhub_token:
self.set_joyhub_token(joyhub_token)
obj_get_log.info("从配置文件读取JoyHub Token成功: {}".format(joyhub_token))
else:
obj_get_log.warning("未从配置文件读取到JoyHub Token")
def _read_robot_variable(self, var_name):
import re
robot_file_path = r'C:\Users\a\PyCharmMiscProject\smart-management-auto-test\dulizhan\test_case\Resource\AdapterKws\hh-qa.robot'
obj_get_log.info("尝试读取配置文件: {}".format(robot_file_path))
try:
with open(robot_file_path, 'r', encoding='utf-8') as f:
content = f.read()
pattern = r'\$\{' + re.escape(var_name) + r'\}\s+(\S+)'
match = re.search(pattern, content)
if match:
obj_get_log.info("匹配到变量 {} = {}".format(var_name, match.group(1)))
return match.group(1)
else:
obj_get_log.warning("未匹配到变量 {}".format(var_name))
except Exception as e:
obj_get_log.error("读取robot配置文件失败: {}".format(str(e)))
return None
def kw_joyhub_role_create_post(self, note, **kwargs):
"""
| 功能说明: | 创建角色 |
| 输入参数: | note | 注释 |
| name | 角色名称 | 必填 |
| code | 角色标志 | 必填 |
| sort | 显示顺序 | 必填 |
| status | 状态 | 必填 |
| remark | 备注 | 非必填 |
| 返回参数: | {"code":0,"msg":"","data":role_id} |
"""
logging.info("==========={0}===========".format(note))
name = kwargs.get("name")
code = kwargs.get("code")
sort = kwargs.get("sort")
status = kwargs.get("status")
remark = kwargs.get("remark")
if not name or not code or sort is None or status is None:
raise Exception("角色名称、角色标志、显示顺序和状态不能为空")
request_params = {
"name": name,
"code": code,
"sort": sort,
"status": status
}
if remark is not None and remark != "":
request_params["remark"] = remark
resp = self.kw_in_joyhub_role_create_post(**request_params)
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_role_delete_post(self, note, **kwargs):
"""
| 功能说明: | 删除角色 |
| 输入参数: | note | 注释 |
| id | 角色编号 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
role_id = kwargs.get("id")
if not role_id:
raise Exception("角色编号不能为空")
resp = self.kw_in_joyhub_role_delete_post(id=role_id)
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_role_delete_list_post(self, note, **kwargs):
"""
| 功能说明: | 批量删除角色 |
| 输入参数: | note | 注释 |
| ids | 角色编号数组 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
ids = kwargs.get("ids")
if not ids or not isinstance(ids, list):
raise Exception("角色编号数组不能为空")
resp = self.kw_in_joyhub_role_delete_list_post(ids=ids)
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_role_get_get(self, note, **kwargs):
"""
| 功能说明: | 获得角色信息 |
| 输入参数: | note | 注释 |
| id | 角色编号 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":{...}} |
"""
logging.info("==========={0}===========".format(note))
role_id = kwargs.get("id")
if not role_id:
raise Exception("角色编号不能为空")
resp = self.kw_in_joyhub_role_get_get(id=role_id)
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_role_page_get(self, note, **kwargs):
"""
| 功能说明: | 获得角色分页 |
| 输入参数: | note | 注释 |
| pageNo | 页码 | 必填 |
| pageSize | 每页条数 | 必填 |
| name | 角色名称 | 非必填 |
| code | 角色标识 | 非必填 |
| status | 状态 | 非必填 |
| 返回参数: | {"code":0,"msg":"","data":{...}} |
"""
logging.info("==========={0}===========".format(note))
page_no = kwargs.get("pageNo")
page_size = kwargs.get("pageSize")
name = kwargs.get("name")
code = kwargs.get("code")
status = kwargs.get("status")
if not page_no or not page_size:
raise Exception("页码和每页条数不能为空")
request_params = {
"pageNo": page_no,
"pageSize": page_size
}
if name is not None and name != "":
request_params["name"] = name
if code is not None and code != "":
request_params["code"] = code
if status is not None:
request_params["status"] = status
resp = self.kw_in_joyhub_role_page_get(**request_params)
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_role_simple_list_get(self, note, **kwargs):
"""
| 功能说明: | 获取角色精简信息列表 |
| 输入参数: | note | 注释 |
| 返回参数: | {"code":0,"msg":"","data":[...]} |
"""
logging.info("==========={0}===========".format(note))
resp = self.kw_in_joyhub_role_simple_list_get()
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_role_list_all_simple_get(self, note, **kwargs):
"""
| 功能说明: | 获取角色精简信息列表 |
| 输入参数: | note | 注释 |
| 返回参数: | {"code":0,"msg":"","data":[...]} |
"""
logging.info("==========={0}===========".format(note))
resp = self.kw_in_joyhub_role_list_all_simple_get()
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_role_update_put(self, note, **kwargs):
"""
| 功能说明: | 修改角色 |
| 输入参数: | note | 注释 |
| id | 角色编号 | 必填 |
| name | 角色名称 | 必填 |
| code | 角色标志 | 必填 |
| sort | 显示顺序 | 必填 |
| status | 状态 | 必填 |
| remark | 备注 | 非必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
role_id = kwargs.get("id")
name = kwargs.get("name")
code = kwargs.get("code")
sort = kwargs.get("sort")
status = kwargs.get("status")
remark = kwargs.get("remark")
if not role_id or not name or not code or sort is None or status is None:
raise Exception("角色编号、角色名称、角色标志、显示顺序和状态不能为空")
request_params = {
"id": role_id,
"name": name,
"code": code,
"sort": sort,
"status": status
}
if remark is not None and remark != "":
request_params["remark"] = remark
resp = self.kw_in_joyhub_role_update_put(**request_params)
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_auth_login_post(self, note, username=None, password=None, max_retries=3, retry_delay=2):
"""
| 功能说明: | 使用账号密码登录(带异常重试机制) |
| 输入参数: | note | 注释 |
| username | 账号 | 必填默认joytest |
| password | 密码 | 必填默认123456 |
| max_retries | 最大重试次数 | 非必填默认3 |
| retry_delay | 重试间隔(秒) | 非必填默认2 |
| 返回参数: | {"code":200,"msg":"success","data":{...}} |
"""
logging.info("==========={0}===========".format(note))
login_username = username if username else "joytest"
login_password = password if password else "123456"
last_exception = None
for attempt in range(max_retries):
try:
logging.info("登录尝试 {}/{}".format(attempt + 1, max_retries))
request_params = {
"username": login_username,
"password": login_password
}
resp = self.kw_in_joyhub_auth_login_post(**request_params)
if resp is not None:
if isinstance(resp, dict) and resp.get('code') == 200:
logging.info("✓ 登录成功")
print("登录成功:", resp)
return resp
elif isinstance(resp, dict) and resp.get('code') == 401:
logging.warning("登录失败账号未登录或token过期重试中...")
else:
logging.warning("登录失败,响应: {}".format(resp))
else:
logging.warning("登录失败,响应为空")
last_exception = Exception("登录失败")
except Exception as e:
last_exception = e
logging.error("登录异常(第{}次尝试): {}".format(attempt + 1, str(e)))
if attempt < max_retries - 1:
logging.info("等待 {} 秒后重试...".format(retry_delay))
time.sleep(retry_delay)
logging.error("登录失败,已达到最大重试次数 {}".format(max_retries))
raise last_exception if last_exception else Exception("登录失败,已达到最大重试次数")
if __name__ == '__main__':
test = RoleManage()
a = test.kw_joyhub_role_page_get(note="测试角色分页", pageNo=1, pageSize=10)
print(a)

View File

@@ -0,0 +1,190 @@
import logging
import allure
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_log = logging.getLogger("logger")
class ShippingTemplateManage(DlzhanInterface):
def __init__(self):
super().__init__()
@allure.step("创建运费模板")
def kw_joyhub_shipping_template_create_post(self, template_name, is_default, calculation_algorithm, currency, status, id=0):
"""
创建运费模板业务关键字
:param template_name: 模板名称
:param is_default: 是否为默认模板1-是2-否
:param calculation_algorithm: 运费计算算法fixed_amount-固定金额percentage-百分比
:param currency: 币种
:param status: 状态 1-启用2-停用
:param id: 主键新增为0
:return: 响应结果
"""
obj_log.info(f"创建运费模板 - templateName: {template_name}, isDefault: {is_default}, calculationAlgorithm: {calculation_algorithm}, currency: {currency}, status: {status}")
params = {
"id": id,
"templateName": template_name,
"isDefault": is_default,
"calculationAlgorithm": calculation_algorithm,
"currency": currency,
"status": status
}
resp = self.kw_in_joyhub_shipping_template_create_post(**params)
obj_log.info(f"创建运费模板响应: {resp}")
return resp
@allure.step("删除运费模板")
def kw_joyhub_shipping_template_delete_post(self, shipping_template_id):
"""
删除运费模板业务关键字
:param shipping_template_id: 运费模板编号
:return: 响应结果
"""
obj_log.info(f"删除运费模板 - id: {shipping_template_id}")
resp = self.kw_in_joyhub_shipping_template_delete_post(shipping_template_id=shipping_template_id)
obj_log.info(f"删除运费模板响应: {resp}")
return resp
@allure.step("批量删除运费模板")
def kw_joyhub_shipping_template_delete_list_post(self, ids):
"""
批量删除运费模板业务关键字
:param ids: 运费模板编号列表
:return: 响应结果
"""
obj_log.info(f"批量删除运费模板 - ids: {ids}")
resp = self.kw_in_joyhub_shipping_template_delete_list_post(ids=ids)
obj_log.info(f"批量删除运费模板响应: {resp}")
return resp
@allure.step("获得运费模板详情")
def kw_joyhub_shipping_template_get_detail_get(self, shipping_template_id):
"""
获得运费模板详情(含规则与子表)业务关键字
:param shipping_template_id: 运费模板编号
:return: 响应结果
"""
obj_log.info(f"获得运费模板详情 - id: {shipping_template_id}")
resp = self.kw_in_joyhub_shipping_template_get_detail_get(shipping_template_id=shipping_template_id)
obj_log.info(f"获得运费模板详情响应: {resp}")
return resp
@allure.step("获得运费模板分页列表")
def kw_joyhub_shipping_template_page_get(self, page_no=1, page_size=10, template_name="", is_default=None, calculation_algorithm="", currency="", status=None):
"""
获得运费模板分页列表业务关键字
:param page_no: 页码
:param page_size: 每页条数
:param template_name: 模板名称
:param is_default: 是否为默认模板1-是2-否
:param calculation_algorithm: 运费计算算法
:param currency: 币种
:param status: 状态
:return: 响应结果
"""
obj_log.info(f"获得运费模板分页列表 - pageNo: {page_no}, pageSize: {page_size}, templateName: {template_name}")
params = {
"pageNo": page_no,
"pageSize": page_size
}
if template_name:
params["templateName"] = template_name
if is_default is not None:
params["isDefault"] = is_default
if calculation_algorithm:
params["calculationAlgorithm"] = calculation_algorithm
if currency:
params["currency"] = currency
if status is not None:
params["status"] = status
resp = self.kw_in_joyhub_shipping_template_page_get(**params)
obj_log.info(f"获得运费模板分页列表响应: {resp}")
return resp
@allure.step("保存运费模板(含规则与子表)")
def kw_joyhub_shipping_template_save_with_children_post(self, template_name, is_default, calculation_algorithm, currency, status, shipping_rules=None, id=0):
"""
保存运费模板信息(含规则与子表)业务关键字
:param template_name: 模板名称
:param is_default: 是否为默认模板1-是2-否
:param calculation_algorithm: 运费计算算法
:param currency: 币种
:param status: 状态
:param shipping_rules: 运费规则列表
:param id: 主键新增为0
:return: 响应结果
"""
obj_log.info(f"保存运费模板(含规则与子表) - templateName: {template_name}, isDefault: {is_default}")
params = {
"id": id,
"templateName": template_name,
"isDefault": is_default,
"calculationAlgorithm": calculation_algorithm,
"currency": currency,
"status": status
}
if shipping_rules:
params["shippingRules"] = shipping_rules
resp = self.kw_in_joyhub_shipping_template_save_with_children_post(**params)
obj_log.info(f"保存运费模板(含规则与子表)响应: {resp}")
return resp
@allure.step("获得运费规则列表")
def kw_joyhub_shipping_template_shipping_rule_list_get(self, shipping_template_id):
"""
获得运费规则列表业务关键字
:param shipping_template_id: 关联的模板ID
:return: 响应结果
"""
obj_log.info(f"获得运费规则列表 - shippingTemplateId: {shipping_template_id}")
resp = self.kw_in_joyhub_shipping_template_shipping_rule_list_get(shipping_template_id=shipping_template_id)
obj_log.info(f"获得运费规则列表响应: {resp}")
return resp
@allure.step("更新运费模板")
def kw_joyhub_shipping_template_update_put(self, id, template_name, is_default, calculation_algorithm, currency, status):
"""
更新运费模板业务关键字
:param id: 模板编号
:param template_name: 模板名称
:param is_default: 是否为默认模板1-是2-否
:param calculation_algorithm: 运费计算算法
:param currency: 币种
:param status: 状态
:return: 响应结果
"""
obj_log.info(f"更新运费模板 - id: {id}, templateName: {template_name}")
params = {
"id": id,
"templateName": template_name,
"isDefault": is_default,
"calculationAlgorithm": calculation_algorithm,
"currency": currency,
"status": status
}
resp = self.kw_in_joyhub_shipping_template_update_put(**params)
obj_log.info(f"更新运费模板响应: {resp}")
return resp

View File

@@ -0,0 +1,451 @@
# -*- coding:utf-8 -*-
"""
Author: Auto Generated
Create Date: 2026/04/28
"""
import logging
import os
import sys
current_file_path = os.path.abspath(__file__)
project_root = os.path.abspath(os.path.join(os.path.dirname(current_file_path), '../../../../'))
if project_root not in sys.path:
sys.path.insert(0, project_root)
import time
from base_framework.public_tools import log
from base_framework.public_tools.my_faker import MyFaker
from base_framework.public_tools.runner import Runner
from base_framework.public_tools import utils
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_get_log = log.get_logger()
obj_my_faker = MyFaker()
obj_runner = Runner()
obj_get_way = utils.Tools()
class UserManage(DlzhanInterface):
def __init__(self):
super().__init__()
# 从robot配置文件读取 JoyHub Token
joyhub_token = self._read_robot_variable("joyhub_login_token")
if joyhub_token:
self.set_joyhub_token(joyhub_token)
obj_get_log.info("从配置文件读取JoyHub Token成功: {}".format(joyhub_token))
else:
obj_get_log.warning("未从配置文件读取到JoyHub Token")
def _read_robot_variable(self, var_name):
"""从robot配置文件读取变量"""
import re
# 使用相对路径定位配置文件
robot_file_path = os.path.join(os.path.dirname(__file__), '../../../../test_case/Resource/AdapterKws/hh-qa.robot')
obj_get_log.info("尝试读取配置文件: {}".format(robot_file_path))
try:
with open(robot_file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 匹配 ${var_name} value 格式
pattern = r'\$\{' + re.escape(var_name) + r'\}\s+(\S+)'
match = re.search(pattern, content)
if match:
obj_get_log.info("匹配到变量 {} = {}".format(var_name, match.group(1)))
return match.group(1)
else:
obj_get_log.warning("未匹配到变量 {}".format(var_name))
except Exception as e:
obj_get_log.error("读取robot配置文件失败: {}".format(str(e)))
return None
def kw_joyhub_user_create_post(self, note, **kwargs):
"""
| 功能说明: | 新增用户 |
| 输入参数: | note | 注释 |
| username | 用户账号 | 必填 |
| nickname | 用户昵称 | 必填 |
| password | 密码 | 必填 |
| email | 用户邮箱 | 非必填 |
| mobile | 手机号码 | 非必填 |
| sex | 用户性别 | 非必填 |
| avatar | 用户头像 | 非必填 |
| remark | 备注 | 非必填 |
| deptId | 部门编号 | 非必填 |
| postIds | 岗位编号数组 | 非必填 |
| 返回参数: | {"code":0,"msg":"","data":user_id} |
"""
logging.info("==========={0}===========".format(note))
username = kwargs.get("username")
nickname = kwargs.get("nickname")
password = kwargs.get("password")
email = kwargs.get("email")
mobile = kwargs.get("mobile")
sex = kwargs.get("sex")
avatar = kwargs.get("avatar")
remark = kwargs.get("remark")
deptId = kwargs.get("deptId")
postIds = kwargs.get("postIds")
if not username or not nickname or not password:
raise Exception("用户账号、用户昵称和密码不能为空")
request_params = {
"username": username,
"nickname": nickname,
"password": password
}
if email is not None and email != "":
request_params["email"] = email
if mobile is not None and mobile != "":
request_params["mobile"] = mobile
if sex is not None:
request_params["sex"] = sex
if avatar is not None and avatar != "":
request_params["avatar"] = avatar
if remark is not None and remark != "":
request_params["remark"] = remark
if deptId is not None:
request_params["deptId"] = deptId
if postIds is not None:
request_params["postIds"] = postIds
resp = self.kw_in_joyhub_user_create_post(**request_params)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_user_delete_post(self, note, **kwargs):
"""
| 功能说明: | 删除用户 |
| 输入参数: | note | 注释 |
| id | 用户编号 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
user_id = kwargs.get("id")
if not user_id:
raise Exception("用户编号不能为空")
resp = self.kw_in_joyhub_user_delete_post(id=user_id)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_user_delete_list_post(self, note, **kwargs):
"""
| 功能说明: | 批量删除用户 |
| 输入参数: | note | 注释 |
| ids | 用户编号数组 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
ids = kwargs.get("ids")
if not ids or not isinstance(ids, list):
raise Exception("用户编号数组不能为空")
resp = self.kw_in_joyhub_user_delete_list_post(ids=ids)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_user_get_get(self, note, **kwargs):
"""
| 功能说明: | 获得用户详情 |
| 输入参数: | note | 注释 |
| id | 用户编号 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":{...}} |
"""
logging.info("==========={0}===========".format(note))
user_id = kwargs.get("id")
if not user_id:
raise Exception("用户编号不能为空")
resp = self.kw_in_joyhub_user_get_get(id=user_id)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_user_page_get(self, note, **kwargs):
"""
| 功能说明: | 获得用户分页列表 |
| 输入参数: | note | 注释 |
| pageNo | 页码 | 必填 |
| pageSize | 每页条数 | 必填 |
| username | 用户账号 | 非必填 |
| nickname | 用户昵称 | 非必填 |
| mobile | 手机号码 | 非必填 |
| status | 用户状态 | 非必填 |
| deptId | 部门编号 | 非必填 |
| 返回参数: | {"code":0,"msg":"","data":{...}} |
"""
logging.info("==========={0}===========".format(note))
page_no = kwargs.get("pageNo")
page_size = kwargs.get("pageSize")
username = kwargs.get("username")
nickname = kwargs.get("nickname")
mobile = kwargs.get("mobile")
status = kwargs.get("status")
deptId = kwargs.get("deptId")
if not page_no or not page_size:
raise Exception("页码和每页条数不能为空")
request_params = {
"pageNo": page_no,
"pageSize": page_size
}
if username is not None and username != "":
request_params["username"] = username
if nickname is not None and nickname != "":
request_params["nickname"] = nickname
if mobile is not None and mobile != "":
request_params["mobile"] = mobile
if status is not None:
request_params["status"] = status
if deptId is not None:
request_params["deptId"] = deptId
resp = self.kw_in_joyhub_user_page_get(**request_params)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_user_update_put(self, note, **kwargs):
"""
| 功能说明: | 修改用户 |
| 输入参数: | note | 注释 |
| id | 用户编号 | 必填 |
| username | 用户账号 | 必填 |
| nickname | 用户昵称 | 必填 |
| password | 密码 | 非必填 |
| email | 用户邮箱 | 非必填 |
| mobile | 手机号码 | 非必填 |
| sex | 用户性别 | 非必填 |
| avatar | 用户头像 | 非必填 |
| remark | 备注 | 非必填 |
| deptId | 部门编号 | 非必填 |
| postIds | 岗位编号数组 | 非必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
user_id = kwargs.get("id")
username = kwargs.get("username")
nickname = kwargs.get("nickname")
password = kwargs.get("password")
email = kwargs.get("email")
mobile = kwargs.get("mobile")
sex = kwargs.get("sex")
avatar = kwargs.get("avatar")
remark = kwargs.get("remark")
deptId = kwargs.get("deptId")
postIds = kwargs.get("postIds")
if not user_id or not username or not nickname:
raise Exception("用户编号、用户账号和用户昵称不能为空")
request_params = {
"id": user_id,
"username": username,
"nickname": nickname
}
if password is not None and password != "":
request_params["password"] = password
if email is not None and email != "":
request_params["email"] = email
if mobile is not None and mobile != "":
request_params["mobile"] = mobile
if sex is not None:
request_params["sex"] = sex
if avatar is not None and avatar != "":
request_params["avatar"] = avatar
if remark is not None and remark != "":
request_params["remark"] = remark
if deptId is not None:
request_params["deptId"] = deptId
if postIds is not None:
request_params["postIds"] = postIds
resp = self.kw_in_joyhub_user_update_put(**request_params)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_user_update_password_put(self, note, **kwargs):
"""
| 功能说明: | 重置用户密码 |
| 输入参数: | note | 注释 |
| id | 用户编号 | 必填 |
| password | 新密码 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
user_id = kwargs.get("id")
password = kwargs.get("password")
if not user_id or not password:
raise Exception("用户编号和新密码不能为空")
resp = self.kw_in_joyhub_user_update_password_put(id=user_id, password=password)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_user_update_status_put(self, note, **kwargs):
"""
| 功能说明: | 修改用户状态 |
| 输入参数: | note | 注释 |
| id | 用户编号 | 必填 |
| status | 用户状态 | 必填 |
| 返回参数: | {"code":0,"msg":"","data":true} |
"""
logging.info("==========={0}===========".format(note))
user_id = kwargs.get("id")
status = kwargs.get("status")
if not user_id or status is None:
raise Exception("用户编号和用户状态不能为空")
resp = self.kw_in_joyhub_user_update_status_put(id=user_id, status=status)
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "操作失败")
return resp_json
def kw_joyhub_user_simple_list_get(self, note, **kwargs):
"""
| 功能说明: | 获取用户精简信息列表 |
| 输入参数: | note | 注释 |
| 返回参数: | {"code":0,"msg":"","data":[...]} |
"""
logging.info("==========={0}===========".format(note))
resp = self.kw_in_joyhub_user_simple_list_get()
# 解析响应
if hasattr(resp, 'json'):
resp_json = resp.json()
else:
resp_json = resp
print(resp_json if resp_json else "查询失败")
return resp_json
def kw_joyhub_auth_login_post(self, note, username=None, password=None, max_retries=3, retry_delay=2):
"""
| 功能说明: | 使用账号密码登录(带异常重试机制) |
| 输入参数: | note | 注释 |
| username | 账号 | 必填默认joytest |
| password | 密码 | 必填默认123456 |
| max_retries | 最大重试次数 | 非必填默认3 |
| retry_delay | 重试间隔(秒) | 非必填默认2 |
| 返回参数: | {"code":200,"msg":"success","data":{...}} |
"""
logging.info("==========={0}===========".format(note))
# 使用默认账号密码
login_username = username if username else "joytest"
login_password = password if password else "123456"
last_exception = None
for attempt in range(max_retries):
try:
logging.info("登录尝试 {}/{}".format(attempt + 1, max_retries))
request_params = {
"username": login_username,
"password": login_password
}
resp = self.kw_in_joyhub_auth_login_post(**request_params)
if resp is not None:
if isinstance(resp, dict) and resp.get('code') == 200:
logging.info("✓ 登录成功")
print("登录成功:", resp)
return resp
elif isinstance(resp, dict) and resp.get('code') == 401:
logging.warning("登录失败账号未登录或token过期重试中...")
else:
logging.warning("登录失败,响应: {}".format(resp))
else:
logging.warning("登录失败,响应为空")
last_exception = Exception("登录失败")
except Exception as e:
last_exception = e
logging.error("登录异常(第{}次尝试): {}".format(attempt + 1, str(e)))
# 如果不是最后一次尝试,等待后重试
if attempt < max_retries - 1:
logging.info("等待 {} 秒后重试...".format(retry_delay))
time.sleep(retry_delay)
logging.error("登录失败,已达到最大重试次数 {}".format(max_retries))
raise last_exception if last_exception else Exception("登录失败,已达到最大重试次数")
if __name__ == '__main__':
test = UserManage()
a = test.kw_joyhub_user_page_get(note="测试用户分页", user='admin', pageNo=1, pageSize=10)
print(a)

View File

@@ -0,0 +1,74 @@
import logging
import allure
from dulizhan.library.Dlizhan_interface import DlzhanInterface
obj_log = logging.getLogger("logger")
class UserProfile(DlzhanInterface):
def __init__(self):
super().__init__()
@allure.step("获得登录用户信息")
def kw_joyhub_user_profile_get_get(self):
"""
获得登录用户信息业务关键字
:return: 响应结果
"""
obj_log.info("获得登录用户信息")
resp = self.kw_in_joyhub_user_profile_get_get()
obj_log.info(f"获得登录用户信息响应: {resp}")
return resp
@allure.step("修改用户个人信息")
def kw_joyhub_user_profile_update_put(self, nickname="", email="", mobile="", sex=None, avatar=""):
"""
修改用户个人信息业务关键字
:param nickname: 用户昵称
:param email: 用户邮箱
:param mobile: 手机号码
:param sex: 用户性别
:param avatar: 用户头像
:return: 响应结果
"""
obj_log.info(f"修改用户个人信息 - nickname: {nickname}, email: {email}, mobile: {mobile}, sex: {sex}")
params = {}
if nickname:
params["nickname"] = nickname
if email:
params["email"] = email
if mobile:
params["mobile"] = mobile
if sex is not None:
params["sex"] = sex
if avatar:
params["avatar"] = avatar
resp = self.kw_in_joyhub_user_profile_update_put(**params)
obj_log.info(f"修改用户个人信息响应: {resp}")
return resp
@allure.step("修改用户个人密码")
def kw_joyhub_user_profile_update_password_put(self, old_password, new_password):
"""
修改用户个人密码业务关键字
:param old_password: 旧密码
:param new_password: 新密码
:return: 响应结果
"""
obj_log.info(f"修改用户个人密码 - oldPassword: {old_password}, newPassword: {new_password}")
params = {
"oldPassword": old_password,
"newPassword": new_password
}
resp = self.kw_in_joyhub_user_profile_update_password_put(**params)
obj_log.info(f"修改用户个人密码响应: {resp}")
return resp

View File

@@ -0,0 +1,7 @@
# -*- coding:utf-8 -*-
"""
JoyHub业务关键字模块
"""
from .UserManage import UserManage
__all__ = ['UserManage']