import re import sys import json from pathlib import Path from typing import Callable, Any import pytest import yaml from glrocky.core.logger import logger 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 import socket def _get_local_ip() -> str: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip: str = s.getsockname()[0] s.close() return str(ip) PRODUCTION_EXECUTOR = "192.168.0.139" IS_PRODUCTION = _get_local_ip() == PRODUCTION_EXECUTOR class MetricManager: def __init__(self, metric): self.metric = metric def new_span(self, m_type: str = "default", m_id: str = "default", m_iter: int = 1): self.metric.span(m_type, m_id, m_iter) def _smart_name_label(self, name: str, label: str | None) -> tuple[str, str]: """retruns: (name,label)""" if not name: raise ValueError() if not name.strip(): raise ValueError("Empty name") if label: return name, label for sep in ("|", ":"): if sep in name: _n, _l = name.split(sep, maxsplit=1) return (_n, _l) return name, name def add_text_metric(self, name: str, value: str, label: str | None = None): _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): _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): _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): _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() def from_dict(self, the_dict: dict[str, Any]): if not the_dict: raise ValueError("dict is null") for _k, v in the_dict.items(): 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) 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): difyUrl: str = Field( title="Dify服务地址", ) difyWorkflowId: str = Field( title="Dify工作流ID", ) difyApiKey: str = Field( title="Dify API密钥", ) class MaterialForDify(BaseModel): paramGroupUuid: list[str] = Field(default=[], title="UUID", description="UUID") inputTextList: list[str] = Field( 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 # 用例名称，当前运行的用例编号 appName: str | None = None @field_serializer("inputTextList") def serialize_input_text(self, value: list[str] | str) -> str: if isinstance(value, list): return pre_process_material(value) return value @field_serializer("prompt") def serialize_prompt_as_single_str(self, value: list[str] | str) -> str: if isinstance(value, list): return value[0] if len(value) > 0 else "" return value def call_dify( case_meta_info:dict[str,str], logger: Logger, device_info: Device, material: list[MaterialForDify], dify_cfg: DifySettings, metric, material_reporter, 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(c, n, d: dict[str, Any]): logger.info(f"开始执行:{d.get('title', '')}") logger.info(f"输入节点参数：{d.get('inputs')}") 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 not material or len(material) < 1: raise RuntimeError("缺少素材") else: logger.info(f"下发素材：{material}") mm = MetricManager(metric) for material_index, item in enumerate(material, 1): paramGroupUuid = item.paramGroupUuid[0] assert paramGroupUuid, "The param Group Uuid Value must be set." material_reporter.begin(paramGroupUuid) dify_final_status = False try: # IMPORTANT: the group uuid for params payload = DifyPayload( **item.model_dump(), 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, ) logger.info(f"工作流返回结果:{result.outputs}") logger.info(f"工作流最终状态:{result.success}") if result.outputs and len(result.outputs): logger.info("提交结果到执行器") mm.from_dict(result.outputs) metric.send_all() logger.info("提交结果到执行器完成") else: logger.warning("dify 工作流无返回") if not result.success: logger.error(f"工作流执行失败:{result.error}") dify_final_status = result.success except Exception as e: logger.error(e) dify_final_status = False continue # do not break next round finally: material_reporter.end_with(paramGroupUuid, dify_final_status) def make_dify_test( case_meta: dict[str, str], apps_config: dict[str, DifySettings] | None = None, ) -> Callable[..., None]: @M.meta(**case_meta) def _func( logger: Logger, device_info: Device, material: list[MaterialForDify], metric, material_reporter, meta, ) -> None: dify_settings = apps_config.get(meta.app_name) if apps_config else None call_dify( case_meta, logger, device_info, material, dify_cfg=dify_settings, metric=metric, material_reporter=material_reporter, app_name=meta.app_name, ) return _func def make_skip_test(meta: dict[str, str]) -> Callable[..., None]: @M.skip(reason=f"{meta['id']} 未实现") @M.meta(**meta) def _func( logger: Logger, device_info: Device, material: list[MaterialForDify], metric, ) -> None: logger.error(f"此用例在手机{device_info.device_serial}上暂未实现") assert False, "not implemented" return _func def generate_cases_from_yaml(module_name: str, yaml_path: Path): if not yaml_path.exists(): logger.warning(f"Test case definition file not found: {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["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",'') if not case_id: logger.warning(f"用例 缺少 case_id 字段。{case_info=}") pytest.fail(f"用例 缺少 case_id 字段。{case_info=}") 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, "appNames": ["agent"]} fn_name = f"test_{case_id.lower().replace('-', '_')}" if action == "dify": 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_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"), ) meta["appNames"] = ["agent"] fn = make_dify_test(meta, apps_config or None) elif action == "skipped": fn = make_skip_test(meta) elif action == "custom": logger.info(f"Case {case_id} is a custom test.") continue else: logger.warning(f"Unknown action '{action}' for case {case_id}. Skipping.") continue setattr(sys.modules[module_name], fn_name, fn) 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)