scripts/oppo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
scripts@glrocky.com
2026-07-03 14:35:42 +08:00
parent fa51b703a7
commit 73fab6ea76
3 changed files with 321 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
glrocky_sdk-0.4.5-py3-none-any.whl Normal file
View File

Binary file not shown.

437
scenarios/tc_generator.py
View File

@@ -1,10 +1,6 @@
import os
import re
import sys
import json
import time
import concurrent.futures
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Any
@@ -16,78 +12,20 @@ from glrocky.framework.marks import Marks as M
from glrocky.framework.schemas import Device
from glrocky.services.dify.dify import run_workflow
from loguru._logger import Logger
from pydantic import BaseModel, ConfigDict, Field, field_serializer, create_model
from pydantic import BaseModel, ConfigDict, Field, field_serializer
import socket
PRODUCTION_ENV_NAME = "GLROCKY_PRODUCTION_MODE"
PRODUCTION_EXECUTOR = "192.168.0.139"
YAML_FILE = Path(__file__).parent / "test_cases.yaml"
def _get_local_ip() -> str:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(("8.8.8.8", 80))
ip: str = s.getsockname()[0]
return str(ip)
finally:
s.close()
s.connect(("8.8.8.8", 80))
ip: str = s.getsockname()[0]
s.close()
return str(ip)
def _is_production_mode() -> bool:
env_val = os.getenv(PRODUCTION_ENV_NAME)
if env_val is not None:
return env_val.lower() in ("true", "yes", "1")
return _get_local_ip() == PRODUCTION_EXECUTOR
def pre_process_material(input: list[str]) -> str:
assert isinstance(input, list)
if not input:
return json.dumps([], ensure_ascii=False)
result = []
for text in input:
splited = _Q_PATTERN.split(text, maxsplit=50)
for part in splited:
if not part.strip():
continue
result.append(part.strip())
return json.dumps(result, ensure_ascii=False)
IS_PRODUCTION = _is_production_mode()
METRIC_FIELD_MAPPING: dict[str, str] = {
"result": "result|结果",
"_dify_result": "_dify_result|业务完成",
"_dify_message": "_dify_message|业务错误信息",
"resultText": "resultText|文本结果",
"recordVideo": "recordVideo|录像",
"recordAudio": "recordAudio|录音",
"screenshotList": "screenshotList|截图",
"firstToken": "firstToken|首字符时长",
"timeSeries": "timeSeries|时间序列",
"fileList": "fileList|文件",
}
IMAGE_EXTENSIONS = {".jpg", ".png", ".gif", ".bmp"}
VIDEO_EXTENSIONS = {".mp4", ".mkv"}
_Q_PATTERN = re.compile(r"Q\d+\s*[:：\.]+\s*", flags=re.MULTILINE)
@dataclass
class _TimeoutCtx:
client: Any = None
task_id: str | None = None
def stop(self):
if self.client and self.task_id:
try:
logger.info(f"Stopping workflow: {self.task_id}")
self.client.stop_workflow(self.task_id)
logger.info(f"Stopped workflow: {self.task_id}")
except Exception as e:
logger.debug(f"Stop workflow failed: {e}")
PRODUCTION_EXECUTOR = "192.168.0.139"
IS_PRODUCTION = _get_local_ip() == PRODUCTION_EXECUTOR
class MetricManager:
@@ -99,32 +37,54 @@ class MetricManager:
def _smart_name_label(self, name: str, label: str | None) -> tuple[str, str]:
"""retruns: (name,label)"""
if not name or not name.strip():
if not name:
raise ValueError()
if not name.strip():
raise ValueError("Empty name")
if label:
return name, label
sep = next((s for s in ("|", ":") if s in name), None)
if sep:
_n, _l = name.split(sep, maxsplit=1)
return (_n, _l)
for sep in ("|", ":"):
if sep in name:
_n, _l = name.split(sep, maxsplit=1)
return (_n, _l)
return name, name
def _add_metric(self, name: str, value: Any, m_type: str, label: str | None = None):
_n, _l = self._smart_name_label(name, label)
self.metric.add(name=_n, label=_l, value=value, type=m_type)
def add_text_metric(self, name: str, value: str, label: str | None = None):
self._add_metric(name, value, "text", label)
_n, _l = self._smart_name_label(name, label)
self.metric.add(
name=_n,
label=_l,
value=value,
type="text",
)
def add_number_metric(self, name: str, value: float, label: str | None = None):
self._add_metric(name, value, "number", label)
_n, _l = self._smart_name_label(name, label)
self.metric.add(
name=_n,
label=_l,
value=value,
type="number",
)
def add_image_metric(self, name: str, value: Path | str, label: str | None = None):
self._add_metric(name, str(value), "image", label)
_n, _l = self._smart_name_label(name, label)
self.metric.add(
name=_n,
label=_l,
value=str(value),
type="image",
)
def add_video_metric(self, name: str, value: Path | str, label: str | None = None):
self._add_metric(name, value, "video", label)
_n, _l = self._smart_name_label(name, label)
self.metric.add(
name=_n,
label=_l,
value=value,
type="video",
)
def send(self):
self.metric.send_all()
@@ -133,26 +93,35 @@ class MetricManager:
if not the_dict:
raise ValueError("dict is null")
for _k, v in the_dict.items():
k = METRIC_FIELD_MAPPING.get(_k, _k)
k = {
"result": "result|结果", # True ,False
"_dify_result": "_dify_result|业务完成", # True ,False
"_dify_message": "_dify_message|业务错误信息", # string,None
"resultText": "resultText|文本结果",
"recordVideo": "recordVideo|录像",
"recordAudio": "recordAudio|录音",
"screenshotList": "screenshotList|截图",
"firstToken": "firstToken|首字符时长",
"timeSeries":"timeSeries|时间序列",
"fileList":"fileList|文件"
}.get(_k, _k)
logger.info(k)
match v:
case None:
self.add_text_metric(name=k, value="")
case int() | float():
self.add_number_metric(name=k, value=v)
case str():
self.add_text_metric(name=k, value=v)
case Path():
match v.suffix.lower():
case s if s in IMAGE_EXTENSIONS:
self.add_image_metric(name=k, value=str(v.resolve()))
case s if s in VIDEO_EXTENSIONS:
self.add_video_metric(name=k, value=str(v.resolve()))
case _:
self.add_text_metric(name=k, value=str(v.resolve()))
case _:
raise TypeError(f"{type(v)} not supported")
if v is None:
self.add_text_metric(name=k, value="")
elif isinstance(v, (int, float)):
self.add_number_metric(name=k, value=v)
elif isinstance(v, str):
self.add_text_metric(name=k, value=v)
elif isinstance(v, Path):
if v.suffix.lower() in (".jpg", ".png", ".gif", ".bmp"):
self.add_image_metric(name=k, value=str(v.resolve()))
elif v.suffix.lower() in (".mp4", ".mkv"):
self.add_video_metric(name=k, value=str(v.resolve()))
else:
self.add_text_metric(name=k, value=str(v.resolve()))
else:
raise NotImplementedError(f"{type(v)} not supported")
class DifySettings(BaseModel):
@@ -168,31 +137,40 @@ class DifySettings(BaseModel):
class MaterialForDify(BaseModel):
paramGroupUuid: list[str] = Field(
default_factory=list, title="UUID", description="UUID"
)
paramGroupUuid: list[str] = Field(default=[], title="UUID", description="UUID")
inputTextList: list[str] = Field(
default_factory=list,
title="对话列表，支持多轮对话",
description="对话列表，支持多轮对话",
)
prompt: list[str] = Field(
default_factory=list, title="提示词", description="提示词"
default=[], title="对话列表，支持多轮对话", description="对话列表，支持多轮对话"
)
prompt: list[str] = Field(default=[], title="提示词", description="提示词")
model_config = ConfigDict(arbitrary_types_allowed=True) # allows uuid extra...
def pre_process_material(input:list[str])->str:
assert isinstance(input,list)
if not input:
return json.dumps([],ensure_ascii=False)
result=[]
p=re.compile(r"Q\d+\s*[:：\.]+\s*",flags=re.MULTILINE)
for text in input:
splited=re.split(p,text,maxsplit=50)
for part in splited:
if not part.strip():
continue
result.append(part.strip())
return json.dumps(result, ensure_ascii=False)
class DifyPayload(BaseModel):
inputTextList: list[str] | str
prompt: list[str] | str
deviceId: str
address: str
caseId: str
parameters: dict[str, Any] |str| None = None
caseId:str # 用例名称，当前运行的用例编号
appName: str | None = None
@field_serializer("inputTextList")
def serialize_input_text(self, value: list[str] | str) -> str:
if isinstance(value, list):
if isinstance(value, list):
return pre_process_material(value)
return value
@@ -203,97 +181,36 @@ class DifyPayload(BaseModel):
return value
def create_dynamic_model(model_name: str, parameters: list[dict]) -> type[BaseModel]:
field_definitions = {}
for idx, param in enumerate(parameters):
name: str = (
param.get("name", "").strip() if isinstance(param.get("name"), str) else ""
)
if not name:
raise ValueError(f"参数定义错误: 第 {idx + 1} 个参数缺少 'name' 字段或为空")
label = (
param.get("label", "").strip()
if isinstance(param.get("label"), str)
else name
)
description = (
param.get("description", label).strip()
if isinstance(param.get("description"), str)
else label
)
param_type = (
param.get("type", "string").strip()
if isinstance(param.get("type"), str)
else "string"
)
if param_type not in ("string",):
raise ValueError(
f"参数 '{name}' 不支持的类型 '{param_type}'，目前仅支持 'string'"
)
default = param.get("defaultValue", "")
if default and isinstance(default, str):
default = default.strip()
field_definitions[name] = (
str,
Field(default=default, title=label, description=description),
)
try:
model = create_model(model_name, __base__=BaseModel, **field_definitions)
model.model_config["extra"] = "ignore"
model()
return model
except Exception as e:
raise ValueError(f"创建参数模型 '{model_name}' 失败: {e}")
def call_dify(
case_meta_info: dict[str, str],
case_meta_info:dict[str,str],
logger: Logger,
device_info: Device,
material: list[MaterialForDify],
dify_cfg: DifySettings,
metric,
material_reporter,
parameters: dict[str, Any] | None = None,
timeout_sec: int | None = None,
app_name: str | None = None,
):
# logger.info(device_info.device_serial)
logger.info(dify_cfg)
event_callbacks: dict[str, Callable[..., None]] = {}
def on_node_started(_, __, d: dict[str, Any]):
def on_node_started(c, n, d: dict[str, Any]):
logger.info(f"开始执行:{d.get('title', '')}")
logger.info(f"输入节点参数：{d.get('inputs')}")
def on_node_finished(_, __, d):
def on_node_finished(c, n, d):
logger.info(f"结束执行:{d.get('title', '')}")
logger.info(f"节点输出：{d.get('outputs')}")
event_callbacks["on_node_started"] = on_node_started
event_callbacks["on_node_finished"] = on_node_finished
if timeout_sec:
ctx = _TimeoutCtx()
def on_workflow_started(client, event_name, event_data):
logger.info("开始执行dify任务")
ctx.client = client
ctx.task_id = event_data.get("task_id")
event_callbacks["on_workflow_started"] = on_workflow_started
if not material:
if not material or len(material) < 1:
raise RuntimeError("缺少素材")
else:
logger.info(f"下发素材：{material}")
metric_manager = MetricManager(metric)
local_ip = _get_local_ip()
mm = MetricManager(metric)
for material_index, item in enumerate(material, 1):
paramGroupUuid = item.paramGroupUuid[0]
@@ -303,48 +220,28 @@ def call_dify(
dify_final_status = False
try:
# IMPORTANT: the group uuid for params
payload_data = {
payload = DifyPayload(
**item.model_dump(),
"deviceId": device_info.device_serial,
"address": local_ip,
"caseId": case_meta_info.get("id", "Unknown"),
}
if parameters:
payload_data["parameters"] = json.dumps(parameters)
payload = DifyPayload(**payload_data)
logger.info(f"payload send to dify:\n{payload.model_dump_json(indent=2)}\n")
metric_manager.new_span(
"default", f"default-{material_index}", material_index
deviceId=device_info.device_serial,
address=_get_local_ip(),
caseId=case_meta_info.get('id','Unkown'),
appName=app_name,
)
logger.info(f"payload send to dify:\n{payload.model_dump_json()}\n")
mm.new_span("default", f"default-{material_index}", material_index)
result = run_workflow(
api_key=dify_cfg.difyApiKey,
base_url=dify_cfg.difyUrl,
workflow_id=dify_cfg.difyWorkflowId,
inputs=payload.model_dump(),
event_callbacks=event_callbacks,
)
def _run_workflow():
return run_workflow(
api_key=dify_cfg.difyApiKey,
base_url=dify_cfg.difyUrl,
workflow_id=dify_cfg.difyWorkflowId,
inputs=payload.model_dump(exclude_none=True),
event_callbacks=event_callbacks,
)
if timeout_sec:
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(_run_workflow)
try:
result = future.result(timeout=timeout_sec)
except concurrent.futures.TimeoutError:
logger.error(f"Workflow timeout after {timeout_sec}s,(yaml config)")
time.sleep(0.1)
logger.info("尝试停止工作流...")
ctx.stop()
raise TimeoutError(f"Workflow timeout after {timeout_sec}s")
else:
result = _run_workflow()
logger.info(f"工作流返回结果:{result.outputs}")
logger.info(f"工作流最终状态:{result.success}")
if result.outputs:
if result.outputs and len(result.outputs):
logger.info("提交结果到执行器")
metric_manager.from_dict(result.outputs)
mm.from_dict(result.outputs)
metric.send_all()
logger.info("提交结果到执行器完成")
else:
@@ -361,39 +258,31 @@ def call_dify(
def make_dify_test(
meta: dict[str, str],
dify_settings: DifySettings | None = None,
param_model: type[BaseModel] | None = None,
timeout: int | None = None,
case_meta: dict[str, str],
apps_config: dict[str, DifySettings] | None = None,
) -> Callable[..., None]:
if param_model is None:
param_model = create_model("EmptyParams", __base__=BaseModel)
@pytest.mark.usefixtures("cfg")
def _func_impl(
@M.meta(**case_meta)
def _func(
logger: Logger,
device_info: Device,
material: list[MaterialForDify],
metric,
material_reporter,
cfg: param_model,
meta,
) -> None:
param_dict = cfg.model_dump() if cfg else {}
dify_settings = apps_config.get(meta.app_name) if apps_config else None
call_dify(
meta,
case_meta,
logger,
device_info,
material,
dify_cfg=dify_settings,
metric=metric,
material_reporter=material_reporter,
parameters=param_dict,
timeout_sec=timeout,
app_name=meta.app_name,
)
# always use cfg fixture
# _func_impl = pytest.mark.usefixtures("cfg")(_func_impl)
return M.meta(**meta)(_func_impl)
return _func
def make_skip_test(meta: dict[str, str]) -> Callable[..., None]:
@@ -403,7 +292,7 @@ def make_skip_test(meta: dict[str, str]) -> Callable[..., None]:
logger: Logger,
device_info: Device,
material: list[MaterialForDify],
metric,
metric,
) -> None:
logger.error(f"此用例在手机{device_info.device_serial}上暂未实现")
assert False, "not implemented"
@@ -416,56 +305,55 @@ def generate_cases_from_yaml(module_name: str, yaml_path: Path):
logger.warning(f"Test case definition file not found: {yaml_path}")
return
try:
with open(file=yaml_path, mode="r", encoding="utf-8") as f:
all_cases: list[dict[str, Any]] = yaml.safe_load( # pyright: ignore[reportAny]
f
)
except yaml.YAMLError as e:
logger.error(f"Failed to parse YAML file {yaml_path}: {e}")
pytest.fail(f"YAML解析失败: {yaml_path}")
return
with open(file=yaml_path, mode="r", encoding="utf-8") as f:
all_cases: list[dict[str, Any]] = yaml.safe_load( # pyright: ignore[reportAny]
f
)
if not all_cases:
return
for case_info in all_cases.get("cases", []):
for case_info in all_cases["cases"]:
if not isinstance(case_info, dict):
logger.debug(f"Skipping non-dictionary item in YAML file: {case_info}")
continue
case_id: str = case_info.get("id", "")
case_id: str = case_info.get("id",'')
if not case_id:
logger.warning(f"用例 缺少 case_id 字段。{case_info=}")
pytest.fail(f"用例 缺少 case_id 字段。{case_info=}")
description: str = case_info.get("description", "")
description: str = case_info.get("description",'')
if not description:
logger.warning(f"用例{case_id} 缺少 description 字段。")
pytest.fail(f"用例{case_id} 缺少 description 字段。")
action: str = case_info.get("action", "skipped")
meta = {"id": case_id, "description": description}
meta = {"id": case_id, "description": description, "appNames": ["agent"]}
fn_name = f"test_{case_id.lower().replace('-', '_')}"
if action == "dify":
dify_settings: DifySettings | None = None
param_model: type[BaseModel] | None = None
timeout = case_info.get("timeout")
if parameters_def := case_info.get("parameters"):
model_name = f"{case_id.replace('-', '_')}_Params"
param_model = create_dynamic_model(model_name, parameters_def)
if dify_config_block := case_info.get("dify_config"):
apps_config: dict[str, DifySettings] = {}
if apps_list := case_info.get("apps"):
app_names = []
for app_entry in apps_list:
name = app_entry.get("name", "")
app_names.append(name)
if cfg_block := app_entry.get("dify_config"):
env_key = "production" if IS_PRODUCTION else "testing"
if env_cfg := cfg_block.get(env_key):
apps_config[name] = DifySettings(
difyUrl=env_cfg.get("url"),
difyWorkflowId=env_cfg.get("workflow_id"),
difyApiKey=env_cfg.get("api_key"),
)
meta["appNames"] = app_names
elif cfg_block := case_info.get("dify_config"):
env_key = "production" if IS_PRODUCTION else "testing"
if env_config := dify_config_block.get(env_key):
dify_settings = DifySettings(
difyUrl=env_config.get("url"),
difyWorkflowId=env_config.get("workflow_id"),
difyApiKey=env_config.get("api_key"),
if env_cfg := cfg_block.get(env_key):
apps_config["agent"] = DifySettings(
difyUrl=env_cfg.get("url"),
difyWorkflowId=env_cfg.get("workflow_id"),
difyApiKey=env_cfg.get("api_key"),
)
else:
logger.warning(
f"No '{env_key}' config found for {case_id}, will use default."
)
fn = make_dify_test(meta, dify_settings, param_model, timeout)
meta["appNames"] = ["agent"]
fn = make_dify_test(meta, apps_config or None)
elif action == "skipped":
fn = make_skip_test(meta)
elif action == "custom":
@@ -479,5 +367,24 @@ def generate_cases_from_yaml(module_name: str, yaml_path: Path):
logger.info(f"Generated {fn_name} for {case_id} with action '{action}'")
# class TC_CUSTOME_CONFIG(BaseModel):
# phoneNumber: str = Field(
# default="15338070617", title="拨号电话", description="待测机的手机号码"
# )
# @M.meta(id="TC-9991")
# def tc_9991(cfg:TC_CUSTOME_CONFIG):
# ...
# class TC_CUSTOME_CONFIG2(BaseModel):
# phoneNumber: str = Field(
# default="15338070617", title="拨号电话", description="待测机的手机号码"
# )
# phoneNumber2: str = Field(
# default="15338070617", title="拨号电话", description="待测机的手机号码"
# )
# def tc_9992(cfg:TC_CUSTOME_CONFIG2):
# ...
YAML_FILE = Path(__file__).parent / "test_cases.yaml"
generate_cases_from_yaml(__name__, YAML_FILE)

159
scenarios/test_cases.yaml
View File

@@ -82,6 +82,11 @@ cases:
category: AI社交.社交应用
action: skipped
- id: TC-0113-txt
description: 社交应用内翻译
category: AI社交.跨语言沟通
action: skipped
- id: TC-0201-img
description: 日程管理
category: AI办公.日程管理
@@ -301,6 +306,31 @@ cases:
production: *production_config
testing: *testing_config
- id: TC-0217-txt
description: 实时翻译
category: AI办公.翻译
action: skipped
- id: TC-0218-txt
description: AI帮写邮件
category: AI办公.文稿处理
action: skipped
- id: TC-0219-txt
description: 邮件翻译
category: AI办公.文稿处理
action: skipped
- id: TC-0220-txt
description: 邮件总结摘要
category: AI办公.文稿处理
action: skipped
- id: TC-0221-txt
description: 网页摘要
category: AI办公.文稿处理
action: skipped
- id: TC-0301-txt
description: 用机问答
category: AI学习.学习问答
@@ -419,16 +449,7 @@ cases:
- id: TC-0312-img
description: 拍摄解题
category: AI学习.AI教学
action: dify
dify_config:
production:
url: "http://192.168.0.140:8090/v1"
workflow_id: "fa6651c7-4820-477c-a17d-950113dbd9d4"
api_key: "app-CCWde8N5Y7t8pGFHUyWca6Nv"
testing:
url: "http://192.168.0.163:8090/v1"
workflow_id: "30ecb958-b56d-4b65-8713-f23945a8db91"
api_key: "app-w2NVHlHZzlIGnmxmkRUO4FPJ"
action: skipped
- id: TC-0313-img
description: 试卷还原
@@ -691,6 +712,14 @@ cases:
category: AI影像.视频处理
action: skipped
- id: TC-0422-txt
description: 音乐生成
category: AI影像.音乐生成
action: dify
dify_config:
production: *production_config
testing: *testing_config
- id: TC-0501-txt
description: 公共服务查询（天气/交通路线/节假日等)
category: AI生活.生活类问答
@@ -833,6 +862,116 @@ cases:
category: AI生活.无障碍辅助
action: skipped
- id: TC-0516-txt
description: AI播客生成
category: AI生活.咨询传播
action: skipped
- id: TC-0516-audio
description: AI播客生成
category: AI生活.咨询传播
action: skipped
- id: TC-0517-txt
description: 个人信息查询
category: AI生活.生活服务
action: skipped
- id: TC-0518-img
description: AI壁纸
category: AI生活.个性化推荐
action: skipped
- id: TC-0519-txt
description: 应用级翻译
category: AI生活.跨语言服务
action: skipped
- id: TC-0520-txt
description: 智能打开付款码
category: AI生活.购物/支付
action: skipped
- id: TC-0521-txt
description: 话费充值
category: AI生活.生活服务
action: skipped
- id: TC-0522-txt
description: 点外卖
category: AI生活.生活服务
action: skipped
- id: TC-0523-txt
description: 生活缴费
category: AI生活.生活服务
action: skipped
- id: TC-0524-txt
description: 餐厅预定
category: AI生活.生活服务
action: skipped
- id: TC-0525-txt
description: 电影预定
category: AI生活.文娱消费服务
action: skipped
- id: TC-0526-txt
description: 演出预定
category: AI生活.文娱消费服务
action: skipped
- id: TC-0527-txt
description: 快递查询
category: AI生活.生活服务
action: skipped
- id: TC-0528-txt
description: 快递提醒
category: AI生活.生活服务
action: skipped
- id: TC-0529-txt
description: 机票/酒店/火车票预定
category: AI生活.智能出行
action: skipped
- id: TC-0530-txt
description: 出行推荐
category: AI生活.智能出行
action: skipped
- id: TC-0531-txt
description: 出行提醒
category: AI生活.日程提醒
action: skipped
- id: TC-0532-txt
description: 健康检测
category: AI生活.运动健康
action: skipped
- id: TC-0533-txt
description: 健康风险预警(心律/睡眠)
category: AI生活.运动健康
action: skipped
- id: TC-0534-txt
description: 健身计划生成
category: AI生活.运动健康
action: skipped
- id: TC-0535-txt
description: 声音修复
category: AI生活.音频智能处理
action: skipped
- id: TC-0536-txt
description: 手语翻译
category: AI生活.无障碍服务
action: skipped
- id: TC-0601-audio
description: 语音唤醒
category: 智能体基础测试.语音唤醒