test_framework
/
youqu
mirror of https://gitee.com/deepin-community/youqu.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: 1、remote模式新增发送用例结果数据到研测平台； 

Description:

Log:
This commit is contained in:
mikigo 2024-05-28 16:44:12 +08:00
parent 979d898595
commit 0b868b88ac
9 changed files with 204 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
RELEASE.md
View File

@ -15,9 +15,15 @@
- 收集远程测试机上所有的 `/report/json/${timestr}_remote/detail_report_${IP}.json` ；
- 收集远程测试机上所有的 `/report/json/${timestr}_remote/summarize_${IP}.json`；
- `parallel` （分布式）模式下自动汇总 `summarize` 数据：`/report/json/${timestr}_remote/summarize.json`；
- 负载均衡驱动模式下自动汇总 `summarize` 数据：`/report/json/${timestr}_remote/summarize.json`；
- `remote` 模式新增参数：`git_url`、`git_user`、`git_password`、`branch`、`depth` 用以控制拉取 git 仓库代码；
- `remote` 模式新增参数：`git_url`、`git_user`、`git_password`、`branch`、`depth` 用以控制拉取 `git` 仓库代码；
- 由于 [letmego](https://linuxdeepin.github.io/letmego/) 方案使用场景较少，默认环境中移除此模块，子项目需要时通过新增依赖机制进行安装；
**Fix 🐛**
- 修复测试单驱动执行报错的问题；
## 2.6.1（2024/05/21）

17
src/cmdctl.py
View File

@ -7,6 +7,7 @@ import os
# SPDX-License-Identifier: GPL-2.0-only
# pylint: disable=C0114
import subprocess
import sys
from setting import conf
from src import logger
@ -47,14 +48,16 @@ class CmdCtl:
@classmethod
def _getstatusoutput(cls, command, timeout):
"""getstatusoutput"""
kwargs = {
"shell": True,
"stderr": subprocess.STDOUT,
"stdout": subprocess.PIPE,
"timeout": timeout,
}
try:
result = cls._run(
command,
shell=True,
text=True,
stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
timeout=timeout,
if sys.version_info >= (3, 7):
kwargs["text"] = True
result = cls._run(command,**kwargs
)
data = result.stdout
exitcode = result.returncode

24
src/plugins/allure_report_extend.py
View File

@ -36,7 +36,7 @@ class AllureReportExtend:
total, failed, passed, skiped, _ = collect_result(execute)
py_case_info = f"{total}/{passed}/{failed}/{skiped}"
w(f"PMS用例维度(总数/通过/失败/跳过)={py_case_info}")
w(f"Py用例维度(总数/通过/失败/跳过)={py_case_info}")
w(f"网络地址={GlobalConfig.USERNAME}@{GlobalConfig.HOST_IP}")
w(f"工作目录={GlobalConfig.ROOT_DIR}")
w(f"镜像版本={GlobalConfig.VERSION}")
@ -46,22 +46,12 @@ class AllureReportExtend:
w(f"显示协议={GlobalConfig.DISPLAY_SERVER.title()}")
try:
cpu_info = (
os.popen(
f"echo '{GlobalConfig.PASSWORD}' | sudo -S dmidecode -s processor-version"
)
.readlines()[0]
.strip("\n")
)
w(f"CPU信息={cpu_info}")
mem_info = os.popen(
f'''echo '{GlobalConfig.PASSWORD}' | sudo -S dmidecode|grep -A16 'Memory Device' | '''
'grep -v "Memory Device Mapped Address" | grep "Range Size"'
).readlines()
MEM_TOTAL = sum([int(i.split(":")[1].rstrip(" GB\n").strip()) for i in mem_info])
w(f"内存信息={MEM_TOTAL}G")
cpu_info = os.popen('cat /proc/cpuinfo | grep "model name"').readlines()
if cpu_info:
w(f"CPU信息={cpu_info[0]}")
mem_info = os.popen('cat /proc/meminfo | grep MemTotal').readlines()
if mem_info:
w(f"内存信息={mem_info[0]}")
except IndexError:
...

31
src/requestx.py
View File

@ -1,22 +1,20 @@
#!/usr/bin/env python3
# _*_ coding:utf-8 _*_
# SPDX-FileCopyrightText: 2023 UnionTech Software Technology Co., Ltd.
# SPDX-License-Identifier: GPL-2.0-only
# pylint: disable=C0114,C0115,W0621,C0103
from urllib import request
from urllib import parse
from urllib.request import build_opener
from http import cookiejar
from urllib import parse
from urllib import request
from urllib.request import build_opener
import json as _json
class RequestX:
def __init__(
self,
login_url=None,
headers=None,
data=None,
self,
login_url=None,
headers=None,
data=None,
):
self.login_url = login_url
self.headers = headers
@ -54,6 +52,19 @@ class RequestX:
response = self.session.open(url, data=data, timeout=timeout).read().decode()
return response
@classmethod
def post(cls, url: str, headers: dict, data: dict = None, json: dict=None):
if data:
params = parse.urlencode(data).encode("utf-8")
elif json:
params = _json.dumps(json)
params = bytes(params, "utf-8")
else:
raise ValueError
r = request.Request(url=url, data=params, headers=headers, method="POST")
req = request.urlopen(r).read().decode("utf-8")
return req
if __name__ == "__main__":
user = "" # 工号

4
src/rtk/_base.py
View File

@ -68,6 +68,10 @@ class Args(Enum):
execution_mode = "execution_mode"
slaves = "slaves"
pms_case_file_path = "pms_case_file_path"
json_backfill_base_url = "json_backfill_base_url"
json_backfill_task_id = "json_backfill_task_id"
json_backfill_user = "json_backfill_user"
json_backfill_password = "json_backfill_password"
def transform_app_name(app_name):

16
src/rtk/_cargo.py
View File

@ -177,6 +177,18 @@ def remote_runner(parser, sub_parser_remote):
"no:表示测试机分布式执行，服务端会根据收集到的测试用例自动分配给各个测试机执行。"
),
)
sub_parser_remote.add_argument(
"--json_backfill_base_url", default="", help="json报告回填的接口地址"
)
sub_parser_remote.add_argument(
"--json_backfill_task_id", default="", help="json报告回填所属任务id"
)
sub_parser_remote.add_argument(
"--json_backfill_user", default="", help="json报告回填的用户名"
)
sub_parser_remote.add_argument(
"--json_backfill_password", default="", help="json报告回填的密码"
)
local_kwargs, args = local_runner(parser, sub_parser_remote)
from src.rtk._base import Args
@ -192,6 +204,10 @@ def remote_runner(parser, sub_parser_remote):
Args.branch.value: args.branch_or_tag or GlobalConfig.BRANCH,
Args.depth.value: args.depth or GlobalConfig.DEPTH,
Args.parallel.value: args.parallel,
Args.json_backfill_base_url.value: args.json_backfill_base_url,
Args.json_backfill_task_id.value: args.json_backfill_task_id,
Args.json_backfill_user.value: args.json_backfill_user,
Args.json_backfill_password.value: args.json_backfill_password,
}
_remote_kwargs = {
"remote_kwargs": remote_kwargs,

44
src/rtk/api_client.py Normal file
View File

@ -0,0 +1,44 @@
from src.requestx import RequestX
class ApiClient:
def __init__(self, base_url, username, password):
self.base_url = base_url
self.username = username
self.password = password
self.token = self.get_token()
self.header = self.get_header()
def get_token(self):
# 设置身份验证端点URL
auth_endpoint = f"{self.base_url}/api/token/"
# 设置用户凭据
credentials = {
"username": self.username,
"password": self.password
}
headers = {"content-type": "application/x-www-form-urlencoded; charset=UTF-8"}
res = RequestX.post(url=auth_endpoint, data=credentials, headers=headers)
import json
jwt_token = json.loads(res)["data"]["access"]
return jwt_token
def get_header(self):
header = {
"Authorization": f"JWT {self.token}",
"Content-Type": "application/json"
}
return header
def post(self, url, data=None, json=None):
response = RequestX.post(url, headers=self.header, data=data, json=json)
return response
if __name__ == "__main__":
ApiClient(
base_url="http://10.7.55.191:8000",
username="superadmin",
password="ut005492",
)

76
src/rtk/json_backfill.py Normal file
View File

@ -0,0 +1,76 @@
#!/usr/bin/env python3
# _*_ coding:utf-8 _*_
# SPDX-FileCopyrightText: 2023 UnionTech Software Technology Co., Ltd.
# SPDX-License-Identifier: GPL-2.0-only
# pylint: disable=C0114
# pylint: disable=C0301,C0116,W0613,W1514,C0103,C0103
import json
import os
import re
from setting import conf
from src.rtk.api_client import ApiClient
class JsonBackfill:
__author__ = "huangmingqiang@uniontech.com"
def __init__(self, base_url, username, password):
self.base_url = base_url
self.username = username
self.password = password
self.api = ApiClient(
base_url=self.base_url,
username=self.username,
password=self.password,
)
def get_remote_json_data(self, json_path):
json_res = {}
for file in os.listdir(json_path):
if file.startswith("detail_report_") and file.endswith(".json"):
client_ip = re.findall(r"detail_report_(\d+\.\d+\.\d+\.\d+)\.json", file)
file_path = os.path.join(json_path, file)
with open(file_path, "r", encoding="utf-8") as f:
json_data = json.load(f)
if client_ip:
json_res[client_ip[0]] = json_data
return json_res
def remote_backfill(self, json_path, json_backfill_task_id):
json_res = self.get_remote_json_data(json_path)
tpl = {
"case": "",
"task": json_backfill_task_id,
"result": "",
"module": "",
"longrepr": "",
"pm_ip": "",
"owner": "",
}
for _ip, res in json_res.items():
for case_py_path, value in res.items():
_, module, *_, case_py = case_py_path.split("/")
tpl["case"] = case_py
case_id = re.findall(r"test_.*?_(\d+)", case_py)
if case_id:
tpl["case"] = case_id[0]
tpl["module"] = module
tpl["result"] = value.get("result")
tpl["longrepr"] = value.get("longrepr")
tpl["pm_ip"] = _ip
res = self.api.post(
url=f"{self.base_url}/api/youqu/yqresult/",
json=tpl
)
print(res)
if __name__ == '__main__':
JsonBackfill(
base_url="http://10.7.55.191:8000",
username="superadmin",
password="ut005492",
).remote_backfill(f"{conf.REPORT_PATH}/json/0524上午113458_remote/", "d3439082af374a94b95e1d3c0613f513")

23
src/rtk/remote_runner.py
View File

@ -95,6 +95,7 @@ class RemoteRunner:
Args.send_code.value: remote_kwargs.get("send_code") or self.send_code,
Args.build_env.value: remote_kwargs.get("build_env") or self.client_env,
Args.parallel.value: remote_kwargs.get("parallel") or self.parallel,
Args.json_backfill_task_id.value: remote_kwargs.get("json_backfill_task_id"),
}
# 客户端地址
if "/home/" not in GlobalConfig.ROOT_DIR:
@ -110,9 +111,11 @@ class RemoteRunner:
self.client_pms_json_report_path = (
lambda x, y: f"/home/{x}/{self.server_project_path}/report/pms_{y}"
)
self.click_json_report_path = (
self.client_json_report_path = (
lambda x: f"/home/{x}/{self.server_project_path}/report/json"
)
self.strf_time = strftime("%m%d%p%I%M%S")
self.server_detail_json_path = f"{GlobalConfig.REPORT_PATH}/json/{self.strf_time}_remote"
self.client_xml_report_path = (
lambda
x: f"/home/{x}/{self.server_project_path}/{GlobalConfig.report_cfg.get('XML_REPORT_PATH', default='report')}/xml".replace(
@ -127,7 +130,6 @@ class RemoteRunner:
self.scp = "sshpass -p '%s' scp -r"
self.rsync = "sshpass -p '%s' rsync -av -e ssh"
self.empty = "> /dev/null 2>&1"
self.strf_time = strftime("%m%d%p%I%M%S")
self.collection_json = False
self.server_json_dir_id = None
@ -450,13 +452,12 @@ class RemoteRunner:
system(
f"{self.scp % password} {user}@{_ip}:{self.client_pms_json_report_path(user, self.server_json_dir_id)}/* {server_json_path}/ {self.empty}"
)
self.server_detail_json_path = f"{GlobalConfig.REPORT_PATH}/json/{self.strf_time}_remote"
self.make_dir(self.server_detail_json_path)
system(
f"{self.scp % password} {user}@{_ip}:{self.click_json_report_path(user)}/detail_report.json {self.server_detail_json_path}/detail_report_{_ip}.json"
f"{self.scp % password} {user}@{_ip}:{self.client_json_report_path(user)}/detail_report.json {self.server_detail_json_path}/detail_report_{_ip}.json"
)
system(
f"{self.scp % password} {user}@{_ip}:{self.click_json_report_path(user)}/summarize.json {self.server_detail_json_path}/summarize_{_ip}.json"
f"{self.scp % password} {user}@{_ip}:{self.client_json_report_path(user)}/summarize.json {self.server_detail_json_path}/summarize_{_ip}.json"
)
def remote_finish_send_to_pms(self):
@ -507,6 +508,18 @@ class RemoteRunner:
if self.collection_json:
self.remote_finish_send_to_pms()
if all([
self.default.get(Args.json_backfill_base_url.value),
self.default.get(Args.json_backfill_task_id.value),
self.default.get(Args.json_backfill_user.value),
self.default.get(Args.json_backfill_password.value)
]):
from src.rtk.json_backfill import JsonBackfill
JsonBackfill(
base_url=self.default.get(Args.json_backfill_base_url.value),
username=self.default.get(Args.json_backfill_user.value),
password=self.default.get(Args.json_backfill_password.value),
).remote_backfill(self.server_detail_json_path, self.default.get(Args.json_backfill_task_id.value))
# 分布式执行的情况下需要汇总结果
if not self.default.get(Args.parallel.value):
summarize = {