update

scenarios/tc_generator.py

@@ -0,0 +1,390 @@
+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)
+