test(python): 全面测试 (#1050)

Author: MistEO

.github/workflows/test.yml

@@ -144,6 +144,7 @@ jobs:
               run: |
                   python3 -m pip install ./source/binding/Python
                   python3 ./test/python/binding_test.py ./source/binding/Python ./install
+                  python3 ./test/python/pipeline_test.py ./source/binding/Python ./install
                   python3 ./test/agent/agent_main_test.py ./source/binding/Python ./install
 
             - name: Run NodeJS testing
@@ -271,6 +272,7 @@ jobs:
                   source .venv/bin/activate
                   python3 -m pip install ./source/binding/Python
                   python3 ./test/python/binding_test.py ./source/binding/Python ./install
+                  python3 ./test/python/pipeline_test.py ./source/binding/Python ./install
                   python3 ./test/agent/agent_main_test.py ./source/binding/Python ./install
 
             - name: Run NodeJS testing
@@ -398,6 +400,7 @@ jobs:
                   source .venv/bin/activate
                   python3 -m pip install ./source/binding/Python
                   python3 ./test/python/binding_test.py ./source/binding/Python ./install
+                  python3 ./test/python/pipeline_test.py ./source/binding/Python ./install
                   python3 ./test/agent/agent_main_test.py ./source/binding/Python ./install
 
             - name: Run NodeJS testing

AGENTS.md

@@ -183,8 +183,8 @@ MaaFramework/
 - [ ] 更新 Pipeline Schema（如涉及）
 - [ ] 添加/更新中英文文档
 - [ ] 更新 MaaMsg.h 及绑定层消息解析代码（如有新消息）
-
-> **注意**：如没有特别说明，不需要更新测试用例。
+- [ ] 更新 `test/python/` 中的单元测试
+- [ ] 更新 `test/agent/` 中的 Agent 单元测试（如涉及 Agent 接口）
 
 ## 常见开发场景
 
@@ -194,8 +194,9 @@ MaaFramework/
 2. 在 `source/MaaFramework/Resource/PipelineParser.cpp` 添加解析逻辑
 3. 在 `source/MaaFramework/Resource/PipelineDumper.cpp` 添加序列化逻辑
 4. 更新 `source/binding` 中的解析逻辑（注意：binding 中的数据来源于 `PipelineDumper` 序列化的 JSON，而非原始 Pipeline JSON。）
-5. 更新 `tools/pipeline.schema.json`
-6. 更新中英文文档 `docs/*/3.1-*`
+5. 在 `test/python/pipeline_test.py` 添加相应的单元测试，验证字段的解析、override 和 get 功能
+6. 更新 `tools/pipeline.schema.json`
+7. 更新中英文文档 `docs/*/3.1-*`
 
 ### 修改回调消息（MaaMsg）
 
@@ -214,12 +215,26 @@ MaaFramework/
 2. 在 `source/MaaAgentClient/Client/AgentClient.cpp` 实现该方法
 3. 在 `source/MaaAgentServer/RemoteInstance/Remote<Module>.h` 添加对应方法声明
 4. 在 `source/MaaAgentServer/RemoteInstance/Remote<Module>.cpp` 实现远程调用逻辑
+5. 在 `test/python/binding_test.py` 添加相应的 API 调用测试
+6. 在 `test/agent/` 中添加 Agent 单元测试，验证远程调用功能
+
+**Agent 单元测试说明**：
+
+- `test/agent/agent_main_test.py`：AgentClient 端测试（主进程），负责创建 Resource/Controller/Tasker，并测试 AgentClient 的连接管理、custom_*_list 等 API
+- `test/agent/agent_child_test.py`：AgentServer 端测试（子进程），在自定义识别器/动作中测试 Remote* 类提供的 Context/Tasker/Resource/Controller API
 
 **Agent 架构说明**：
 
 - **AgentClient**：运行在主程序中，将 Custom 请求转发到 Server，并处理 Server 的远程调用
 - **AgentServer**：运行在用户进程中，注册自定义识别器/动作，通过 Remote* 类代理访问主程序实例
 
+**AgentServer 不支持的 API**（参考 `source/MaaAgentServer/API/MaaAgentServerNotImpl.cpp`）：
+
+- 创建/销毁实例：`MaaResourceCreate`、`MaaTaskerCreate`、`MaaControllerDestroy` 等
+- 全局选项：`MaaGlobalSetOption`、`MaaSetGlobalOption`
+- 插件加载：`MaaGlobalLoadPlugin`
+- Sink 管理：Remote* 实例不支持 `add_sink`/`remove_sink`，需使用 `AgentServer.add_*_sink` 代替
+
 ### 修改 workflows 和 actions
 
 修改 workflows 和 actions 后如果不会自动触发 workflow ，请提醒我 dry run 相关的所有 workflows。

test/agent/agent_child_test.py

@@ -1,13 +1,25 @@
+"""
+AgentServer 端测试
+
+测试范围:
+1. AgentServer: 注册自定义识别器/动作、事件监听器、生命周期管理
+2. Context: run_task/run_recognition/run_action、override_*、clone、anchor、hit_count
+3. Tasker: get_*_detail、running/stopping/post_stop、clear_cache
+4. Resource: get_node_data、node_list、custom_*_list、hash、override_*
+5. Controller: 各种输入操作、post_key_down/up、post_scroll
+"""
+
 import os
 from pathlib import Path
 import sys
 import io
+import numpy
 
 # Fix encoding issues on Windows (cp1252 cannot encode some Unicode characters)
-if sys.stdout.encoding != 'utf-8':
-    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
-if sys.stderr.encoding != 'utf-8':
-    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
+if sys.stdout.encoding != "utf-8":
+    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
+if sys.stderr.encoding != "utf-8":
+    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
 
 if len(sys.argv) < 4:
     print("Call agent_main_test.py instead of this file.")
@@ -36,6 +48,10 @@
 from maa.library import Library
 
 
+analyzed: bool = False
+runned: bool = False
+
+
 def main():
     if len(sys.argv) < 2:
         print("Usage: python agent_main.py <socket_id>")
@@ -56,8 +72,14 @@ def analyze(
         argv: CustomRecognition.AnalyzeArg,
     ) -> CustomRecognition.AnalyzeResult:
         print(
-            f"on MyRecognition.analyze, context: {context}, image: {argv.image.shape}, task_detail: {argv.task_detail}, reco_name: {argv.custom_recognition_name}, reco_param: {argv.custom_recognition_param}"
+            f"on MyRecognition.analyze, context: {context}, image: {argv.image.shape}, "
+            f"task_detail: {argv.task_detail}, reco_name: {argv.custom_recognition_name}, "
+            f"reco_param: {argv.custom_recognition_param}"
         )
+
+        # ============================================================
+        # Context API 测试
+        # ============================================================
         entry = "ColorMatch"
         ppover = {
             "ColorMatch": {
@@ -67,22 +89,95 @@ def analyze(
                 "action": "Click",
             }
         }
+
+        # 测试 run_task
         context.run_task(entry, ppover)
-        action_detail = context.run_action(entry, [114, 514, 191, 810], "RunAction Detail", ppover)
-        print(f"action_detail: {action_detail}")
+
+        # 测试 run_action
+        action_detail = context.run_action(
+            entry, [114, 514, 191, 810], "RunAction Detail", ppover
+        )
+        print(f"  action_detail: {action_detail}")
+
+        # 测试 run_recognition
         reco_detail = context.run_recognition(entry, argv.image, ppover)
-        print(f"reco_detail: {reco_detail}")
+        print(f"  reco_detail: {reco_detail}")
 
+        # 测试 clone 和 override
         new_ctx = context.clone()
         new_ctx.override_pipeline({"TaskA": {}, "TaskB": {}})
         new_ctx.override_next(argv.node_name, ["TaskA", "TaskB"])
 
-        node_detail = new_ctx.tasker.get_latest_node("ColorMatch")
-        print(node_detail)
+        # 测试 get_node_data (Context 级别)
+        node_data = new_ctx.get_node_data(argv.node_name)
+        print(f"  ctx.get_node_data keys: {list(node_data.keys()) if node_data else None}")
+
+        # 测试 anchor API
+        new_ctx.set_anchor("test_anchor", "TaskA")
+        anchor_result = new_ctx.get_anchor("test_anchor")
+        print(f"  anchor_result: {anchor_result}")
+        assert anchor_result == "TaskA", f"anchor should be 'TaskA', got {anchor_result}"
+
+        # 测试 hit count API
+        hit_count = new_ctx.get_hit_count(argv.node_name)
+        print(f"  hit_count: {hit_count}")
+        new_ctx.clear_hit_count(argv.node_name)
 
+        # 测试 override_image (Context 级别)
+        test_image = numpy.zeros((100, 100, 3), dtype=numpy.uint8)
+        new_ctx.override_image("test_image", test_image)
+
+        # 测试 get_task_job
         task_job = new_ctx.get_task_job()
+        print(f"  task_job: {task_job}")
         new_task_detail = task_job.get()
-        print(new_task_detail)
+        print(
+            f"  task_detail entry: {new_task_detail.entry if new_task_detail else None}"
+        )
+
+        # ============================================================
+        # Tasker API 测试 (通过 context.tasker)
+        # ============================================================
+        tasker = new_ctx.tasker
+        print(f"  tasker.inited: {tasker.inited}")
+
+        # 测试 get_latest_node
+        node_detail = tasker.get_latest_node("ColorMatch")
+        print(f"  latest_node ColorMatch: {node_detail}")
+
+        # 测试 running 和 stopping
+        print(f"  tasker.running: {tasker.running}")
+        print(f"  tasker.stopping: {tasker.stopping}")
+
+        # ============================================================
+        # Resource API 测试 (通过 context.tasker.resource)
+        # ============================================================
+        resource = tasker.resource
+
+        # 测试 loaded (valid) 属性
+        print(f"  resource.loaded: {resource.loaded}")
+
+        # 测试 get_node_data (Resource 级别)
+        res_node_data = resource.get_node_data(argv.node_name)
+        print(
+            f"  res.get_node_data keys: {list(res_node_data.keys()) if res_node_data else None}"
+        )
+
+        # 测试 hash 属性
+        res_hash = resource.hash
+        print(f"  resource.hash: {res_hash[:16] if res_hash else None}...")
+
+        # 测试 node_list
+        node_list = resource.node_list
+        print(f"  resource.node_list count: {len(node_list)}")
+
+        # 测试 custom_recognition_list 和 custom_action_list
+        reco_list = resource.custom_recognition_list
+        action_list = resource.custom_action_list
+        print(f"  custom_recognition_list: {reco_list}")
+        print(f"  custom_action_list: {action_list}")
+        assert "MyRec" in reco_list, "MyRec should be in custom_recognition_list"
+        assert "MyAct" in action_list, "MyAct should be in custom_action_list"
 
         global analyzed
         analyzed = True
@@ -100,53 +195,121 @@ def run(
         argv: CustomAction.RunArg,
     ) -> CustomAction.RunResult:
         print(
-            f"on MyAction.run, context: {context}, task_detail: {argv.task_detail}, action_name: {argv.custom_action_name}, action_param: {argv.custom_action_param}, box: {argv.box}, reco_detail: {argv.reco_detail}"
+            f"on MyAction.run, context: {context}, task_detail: {argv.task_detail}, "
+            f"action_name: {argv.custom_action_name}, action_param: {argv.custom_action_param}, "
+            f"box: {argv.box}, reco_detail: {argv.reco_detail}"
         )
+
+        # ============================================================
+        # Controller API 测试 (通过 context.tasker.controller)
+        # ============================================================
         controller = context.tasker.controller
+
+        # 测试 connected 和 uuid
+        connected = controller.connected
+        uuid = controller.uuid
+        print(f"  connected: {connected}, uuid: {uuid}")
+
+        # 测试 post_screencap
         new_image = controller.post_screencap().wait().get()
-        print(f"new_image: {new_image.shape}")
+        print(f"  new_image: {new_image.shape}")
+
+        # 测试 cached_image
+        cached_image = controller.cached_image
+        print(f"  cached_image shape: {cached_image.shape}")
+
+        # 测试基本输入操作
         controller.post_click(191, 98).wait()
         controller.post_swipe(100, 200, 300, 400, 100).wait()
         controller.post_input_text("Hello World!").wait()
         controller.post_click_key(32).wait()
+
+        # 测试触摸操作
         controller.post_touch_down(1, 100, 100, 0).wait()
         controller.post_touch_move(1, 200, 200, 0).wait()
         controller.post_touch_up(1).wait()
+
+        # 测试按键操作
+        controller.post_key_down(65).wait()
+        controller.post_key_up(65).wait()
+
+        # 测试滚动操作
+        controller.post_scroll(0, 120).wait()
+
+        # 测试应用操作
         controller.post_start_app("aaa")
         controller.post_stop_app("bbb")
 
-        cached_image = controller.cached_image
-        connected = controller.connected
-        uuid = controller.uuid
+        # ============================================================
+        # Tasker API 补充测试 (详情获取)
+        # ============================================================
+        tasker = context.tasker
+
+        # 获取当前任务详情用于后续测试
+        task_job = context.get_task_job()
+        task_detail = task_job.get()
+
+        if task_detail:
+            print(f"  task_detail: entry={task_detail.entry}, status={task_detail.status}")
+
+            # 测试 get_task_detail
+            fetched_task_detail = tasker.get_task_detail(task_detail.task_id)
+            print(
+                f"  get_task_detail: {fetched_task_detail.entry if fetched_task_detail else None}"
+            )
+
+            # 测试 get_node_detail
+            if task_detail.nodes:
+                node = task_detail.nodes[0]
+                node_detail = tasker.get_node_detail(node.node_id)
+                print(
+                    f"  get_node_detail: {node_detail.name if node_detail else None}"
+                )
+
+                # 测试 get_recognition_detail
+                if node.recognition:
+                    reco_detail = tasker.get_recognition_detail(node.recognition.reco_id)
+                    print(
+                        f"  get_recognition_detail: {reco_detail.name if reco_detail else None}"
+                    )
+
+        # 测试 clear_cache
+        tasker.clear_cache()
+        print("  tasker.clear_cache() called")
 
         global runned
         runned = True
 
         return CustomAction.RunResult(success=True)
 
 
+# ============================================================================
+# Event Sink 装饰器方式注册
+# ============================================================================
+
+
 @AgentServer.resource_sink()
 class MyResSink(ResourceEventSink):
     def on_raw_notification(self, resource, msg: str, details: dict):
-        print(f"resource: {resource}, msg: {msg}, details: {details} ")
+        print(f"[ResourceSink] msg: {msg}")
 
 
 @AgentServer.controller_sink()
 class MyCtrlSink(ControllerEventSink):
     def on_raw_notification(self, controller, msg: str, details: dict):
-        print(f"controller: {controller}, msg: {msg}, details: {details} ")
+        print(f"[ControllerSink] msg: {msg}")
 
 
 @AgentServer.tasker_sink()
 class MyTaskerSink(TaskerEventSink):
     def on_raw_notification(self, tasker, msg: str, details: dict):
-        print(f"tasker: {tasker}, msg: {msg}, details: {details} ")
+        print(f"[TaskerSink] msg: {msg}")
 
 
 @AgentServer.context_sink()
 class MyCtxSink(ContextEventSink):
     def on_raw_notification(self, context, msg: str, details: dict):
-        print(f"context: {context}, msg: {msg}, details: {details} ")
+        print(f"[ContextSink] msg: {msg}")
 
 
 if __name__ == "__main__":