From 01313dde7e6f7ed508f16016c7807fbe70b62ed1 Mon Sep 17 00:00:00 2001 From: YGN-NDWD-Official Date: Fri, 7 Mar 2025 15:38:29 +0800 Subject: [PATCH] A new test platform call module is added, which can execute compatibility tests of OS\NIC\RAID, etc. by sending test parameters to the platform. --- hwcompatible/constants.py | 3 +- tests/compatible/traverssal/Makefile | 24 +++ tests/compatible/traverssal/traversal_api.py | 143 ++++++++++++++++++ .../traverssal/traversal_parameters.py | 31 ++++ tests/compatible/traverssal/traversal_test.py | 44 ++++++ 5 files changed, 244 insertions(+), 1 deletion(-) create mode 100644 tests/compatible/traverssal/Makefile create mode 100644 tests/compatible/traverssal/traversal_api.py create mode 100644 tests/compatible/traverssal/traversal_parameters.py create mode 100644 tests/compatible/traverssal/traversal_test.py diff --git a/hwcompatible/constants.py b/hwcompatible/constants.py index d952ce6..0ccb1c7 100644 --- a/hwcompatible/constants.py +++ b/hwcompatible/constants.py @@ -25,7 +25,8 @@ CHIP = "chipModel" DRIVER = "driverName" VERSION = "driverVersion" NODEVICE = ("intel", "cpufreq", "memory", "clock", "profiler", "system", - "stress", "kdump", "perf", "acpi", "watchdog", "kabi", "kabiwhitelist", "longtermstability", "srpm") + "stress", "kdump", "perf", "acpi", "watchdog", "kabi", + "kabiwhitelist", "longtermstability", "srpm", "traversal") TEST_KABI_ARCH = ("aarch64", "x86_64") CDTYPES = ("DVD_RW", "DVD_PLUS_RW", "DVD_R", "DVD_PLUS_R", "DVD", "BD_RE", "BD_R", "BD", "CD_RW", "CD_R", "CD") diff --git a/tests/compatible/traverssal/Makefile b/tests/compatible/traverssal/Makefile new file mode 100644 index 0000000..4105f87 --- /dev/null +++ b/tests/compatible/traverssal/Makefile @@ -0,0 +1,24 @@ +# Copyright (c) 2024 Huawei Technologies Co., Ltd. +# oec-hardware is licensed under the Mulan PSL v2. +# You can use this software according to the terms and conditions of the Mulan PSL v2. +# You may obtain a copy of Mulan PSL v2 at: +# http://license.coscl.org.cn/MulanPSL2 +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR +# PURPOSE. +# See the Mulan PSL v2 for more details. +# Author: ygn-ndwd-official +# Create: 2025-02-18 +# Desc: traversal test + +.PHONY: install clean + +all: ; + +install: + mkdir -p $(DEST) + cp -a *.py $(DEST) + chmod a+x $(DEST)/*.py + +clean: + rm -rf $(DEST) diff --git a/tests/compatible/traverssal/traversal_api.py b/tests/compatible/traverssal/traversal_api.py new file mode 100644 index 0000000..2dd51b6 --- /dev/null +++ b/tests/compatible/traverssal/traversal_api.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +# coding: utf-8 + +# Copyright (c) 2020-2024 Huawei Technologies Co., Ltd. +# oec-hardware is licensed under the Mulan PSL v2. +# You can use this software according to the terms and conditions of the Mulan PSL v2. +# You may obtain a copy of Mulan PSL v2 at: +# http://license.coscl.org.cn/MulanPSL2 +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR +# PURPOSE. +# See the Mulan PSL v2 for more details. +# Create: 2025-02-10 +# Desc: invoke the api that traverses the test platform. + +import requests +import json +import time +from datetime import datetime, timedelta +import traversal_parameters + + +def send_test_request(create_parameters): + """ + 新建测试请求 + 后端返回任务ID + """ + api_url = traversal_parameters.create_test_url + try: + response = requests.post(api_url, json=create_parameters) + response.raise_for_status() # 检查HTTP错误 + response_json = response.json() + print(response_json) + if response_json['result'] == 'fail': + print(f"操作失败: {response_json['msg']}") + else: + demand_id = response_json['data'] + if demand_id is not None: + data_length = len(demand_id) + print(f"数据长度：{data_length}") + return demand_id + else: + print("警告：即使操作结果不是 'fail'，'data' 字段仍然是 None。") + except requests.exceptions.RequestException as e: + print(f"请求失败: {e}") + return None + + +def get_test_result(demand_id): + """ + 轮询/等待测试结果 + 接口逻辑：每60秒调用一次后端接口进行任务查询 + 任务执行大体分为5个阶段，预设每个阶段最长执行时间为2天，2天仍未完成当前阶段任务，停止轮询 + 任务执行的所有阶段完成后，保存新一次log，由后续向对log内容进行解析 + """ + result_url = traversal_parameters.get_test_result_url + {demand_id} + previous_status_values = None + last_status_change_time = datetime.now() + polling_interval_seconds = 60 # polling_interval_seconds(int): 轮询间隔时间，单位秒，默认为60秒 + + print(f"开始轮询监控 API: {result_url}") + + while True: + try: + response = requests.get(result_url) + response.raise_for_status() # 检查请求是否成功 (HTTP 状态码 200) + response_json = response.json() + + current_status_values = {} + status_changed = False # 状态变动标记 + task_final_status_reached = False # 是否为最终状态标记 + status_fields_to_monitor = { + "status" + } + stability_time_days = 2 # 最大超时时间 + output_file_path = "api_response.json" # 测试log保存位置 + task_status_field = "status" + finish_pending_status_values = { + "pending", + "finish", + "stop" + } + + for field in status_fields_to_monitor: # status_fields_to_monitor 需要监控的状态字段名列表 + if field in response_json: + current_status_values[field] = response_json[field] + else: + print(f"警告: 监控的状态字段 '{field}' 未在 API 响应中找到。") + current_status_values[field] = None # 字段不存在时设置为 None + + # 检查任务状态是否达到 finish_pending_status_values 中的任何一个值 + if task_status_field in response_json and response_json[task_status_field] in finish_pending_status_values: + task_final_status_reached = True + print(f"任务状态变为最终状态: {response_json[task_status_field]}，停止轮询。") + + if previous_status_values is None: # 首次轮询 + previous_status_values = current_status_values + last_status_change_time = datetime.now() + status_changed = True # 首次轮询也视为状态变化 + else: + if current_status_values != previous_status_values: + status_changed = True + last_status_change_time = datetime.now() + previous_status_values = current_status_values + print( + f"状态字段发生变化: {current_status_values}，上次变化时间: {last_status_change_time.strftime('%Y-%m-%d %H:%M:%S')}") + + if task_final_status_reached or (not status_changed and datetime.now() - last_status_change_time >= timedelta(days=stability_time_days)): + print("满足轮询停止条件，进行最后一次 API 调用并保存完整响应...") + final_response = requests.get(result_url) + final_response.raise_for_status() + final_response_json = final_response.json() + + with open(output_file_path, 'w', encoding='utf-8') as f: + json.dump(final_response_json, f, ensure_ascii=False, indent=4) + print(f"完整响应已保存到: {output_file_path}") + break # 退出轮询循环 + elif not task_final_status_reached and not status_changed: # 只有当最终状态未达到 且 状态字段无变化时 才打印持续轮询信息 + time_since_last_change = datetime.now() - last_status_change_time + print(f"状态字段无变化，已持续: {time_since_last_change.days} 天 {time_since_last_change.seconds // 3600} 小时，任务状态仍未达到最终状态，继续轮询...") + elif not task_final_status_reached and status_changed: # 只有当最终状态未达到 且 状态字段发生变化时 才打印继续轮询信息 + print(f"状态字段发生变化，任务状态仍未达到最终状态，继续轮询...") + + time.sleep(polling_interval_seconds) # 等待一段时间后进行下一次轮询 + + except requests.exceptions.RequestException as e: + print(f"API 请求发生错误: {e}") + print("等待后重试...") + time.sleep(polling_interval_seconds) # 发生错误也等待一段时间再重试 + except json.JSONDecodeError: + print("API 响应内容不是有效的 JSON 格式，请检查接口返回。") + print("等待后重试...") + time.sleep(polling_interval_seconds) + except Exception as e: # 捕获其他未知异常，例如文件写入错误等 + print(f"发生未知错误: {e}") + break # 发生未知错误，直接退出循环 (可以根据需要修改处理方式) + +def create_test_report(data): + """ + 自动化生成测试报告 + 等待平台实现接口 + """ + # to_do \ No newline at end of file diff --git a/tests/compatible/traverssal/traversal_parameters.py b/tests/compatible/traverssal/traversal_parameters.py new file mode 100644 index 0000000..6745664 --- /dev/null +++ b/tests/compatible/traverssal/traversal_parameters.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +# coding: utf-8 + +# Copyright (c) 2020-2024 Huawei Technologies Co., Ltd. +# oec-hardware is licensed under the Mulan PSL v2. +# You can use this software according to the terms and conditions of the Mulan PSL v2. +# You may obtain a copy of Mulan PSL v2 at: +# http://license.coscl.org.cn/MulanPSL2 +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR +# PURPOSE. +# See the Mulan PSL v2 for more details. +# Create: 2025-02-10 +# Desc: invoke the api that traverses the test platform. + +create_test_url = "" +get_test_result_url = "" +test_parameters = { + "creator": "", + "arCode": "", + "cVersionName": "", + "bVersionName": "", + "testType": "", + "testStrategy": "", + "testServer": "", + "osName": "", + "cardName": "", + "driverVersion": "", + "fwVersion": "", + "caseNumbers": "" + } \ No newline at end of file diff --git a/tests/compatible/traverssal/traversal_test.py b/tests/compatible/traverssal/traversal_test.py new file mode 100644 index 0000000..923b80b --- /dev/null +++ b/tests/compatible/traverssal/traversal_test.py @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 +# coding: utf-8 + +# Copyright (c) 2020-2024 Huawei Technologies Co., Ltd. +# oec-hardware is licensed under the Mulan PSL v2. +# You can use this software according to the terms and conditions of the Mulan PSL v2. +# You may obtain a copy of Mulan PSL v2 at: +# http://license.coscl.org.cn/MulanPSL2 +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR +# PURPOSE. +# See the Mulan PSL v2 for more details. +# Create: 2025-02-10 +# Desc: main test + +import traversal_api +import traversal_parameters + + +def main(): + # 1. 准备测试配置（根据实际需求修改） + test_parameters = traversal_parameters.test_parameters + + # 2. 发送请求并获取任务ID + task_id = traversal_api.send_test_request(test_parameters) + if not task_id: + print("获取任务ID失败") + else: + print(task_id) + + # 3. 轮询结果（示例：简单轮询） + # task_id = "674924a0a81300007d005076" + traversal_api.get_test_result(task_id) + """ + # 4. 测试报告生成 等待后端平台实现此功能 + if result: + traversal_api.save_results(result) + else: + print("测试未完成或结果获取失败") + """ + + +if __name__ == "__main__": + main() -- Gitee