UFO Galaxy GalaxySession 测试指南DAG 任务编排会话的测试套件设计与实战【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO本技术指南基于 UFO 仓库中 tests/galaxy/session/README.md 所描述的 GalaxySession 测试套件展开围绕GalaxySession这一 DAG 任务编排会话的核心实现系统讲解其测试文件结构、Mock 策略、事件系统验证与运行方式并结合 galaxy_session.py 源码深入剖析被测对象的关键机制。读者读完本指南后能够理解 GalaxySession 的初始化、轮次Round管理、状态机驱动、观察者集成与取消清理机制掌握用真实生产类 仅 Mock 外部依赖的正确测试方法并可直接运行仓库中的测试套件进行验证。一、背景什么是 GalaxySession在 UFO 的 Galaxy 框架中GalaxySession是承接用户请求、驱动星座Constellation这一 DAG 任务图完成编排与执行的核心会话对象。其定位正如 galaxy_session.py 模块注释所描述继承 UFO 的BaseSession支持基于 DAG 的任务编排协调Constellation与TaskConstellationOrchestrator的协同工作。从源码结构看GalaxySession在构造时完成了一系列关键装配galaxy_session.py编排器创建TaskConstellationOrchestrator并传入client.device_manager与日志开关Agent创建真实的ConstellationAgent通过_init_agents方法而非 mock上下文初始化Context将设备注册表信息写入ContextNames.DEVICE_INFO并挂载 MCP 命令调度器事件系统获取全局单例EventBus并通过_setup_observers注册一批观察者会话状态记录_initial_request、_current_constellation、_session_results等。测试套件正是围绕这一装配过程与后续的运行时行为展开的覆盖了初始化、属性、轮次创建、事件系统、状态管理与清理等全部核心面。二、测试套件结构与文件职责2.1 目录组织tests/ └── galaxy/ └── session/ ├── test_galaxy_session.py # 基础功能测试 ├── test_galaxy_session_integration.py # 集成测试 ├── test_galaxy_session_proper_mock.py # 正确的Mock测试 └── test_galaxy_session_final.py # 最终综合测试四个测试文件各自承担不同的验证层次从单元到集成再到端到端演示形成完整的测试金字塔。2.2test_galaxy_session.py—— 基础功能测试该文件重点验证GalaxySession的五个基础能力test_galaxy_session.py测试点验证内容关键断言对象GalaxySession 初始化构造参数task / should_evaluate / id / client / initial_request是否被正确接收session.task、session._id、session.agent、session.orchestrator会话属性验证状态查询 API 是否可用current_constellation、is_finished()、is_error()、next_request()、request_to_evaluate()Round 创建首个轮次能创建、第二个轮次正确返回Nonesession.create_new_round()的返回值与_id事件系统集成事件总线与观察者是否就位session._event_bus、session._observers数量与类型会话控制功能强制结束链路是否生效await session.force_finish(...)后is_finished()与 agent 状态值得注意该文件还包含一个独立的问题检测函数test_galaxy_session_issuestest_galaxy_session.py专门检查三类常见隐患模块导入是否完整GalaxySession、GalaxyRound、ConstellationAgent、TaskConstellationOrchestrator是否存在抽象方法未实现导致的TypeError实例是否缺少_agent、_orchestrator、_event_bus、_observers、_context、_session_results等关键属性。这体现了该套件不仅测功能还测工程质量的取向。2.3test_galaxy_session_integration.py—— 集成测试该文件把会话放进完整工作流中验证test_galaxy_session_integration.py覆盖会话创建与装配、轮次创建与执行、会话状态管理、Agent 集成agent 名称/状态/编排器/星座、事件系统集成与会话清理共六个阶段。其错误场景测试test_galaxy_session_integration.py则验证鲁棒性边界clientNone构造应当被正确拒绝源码中self._orchestrator TaskConstellationOrchestrator(device_managerclient.device_manager, ...)在 client 为None时会抛出AttributeError测试断言应当失败即验证了该前置条件超长任务名构造 200 字符的任务名应被优雅处理——源码通过safe_task_name[:50]截断并以字母/数字/空格/中划线/下划线过滤非法字符空名则回退为galaxy_session_{id}galaxy_session.py保证日志路径合法空请求next_request()与request_to_evaluate()应能正常返回request_to_evaluate回退到self._initial_request or self.task见 galaxy_session.py。2.4test_galaxy_session_proper_mock.py—— 正确的 Mock 测试这是整套测试中方法论价值最高的一份。它明确了正确 Mock的三条原则test_galaxy_session_proper_mock.py使用真实ConstellationAgent生产代码不让 mock 覆盖被测对象本体只 Mock 外部依赖编排器、处理器、MCP 上下文加载、配置单例保持生产代码完整不通过修改源码来适配测试。具体实现上它通过unittest.mock.patch精准注入三层替身# ① Patch 编排器类避免真实设备执行 with patch( ufo.galaxy.session.galaxy_session.TaskConstellationOrchestrator, MockTaskConstellationOrchestrator, ): # ② Patch Agent 的处理器类避免真实 LLM 调用 with patch( ufo.galaxy.agents.constellation_agent.ConstellationAgentProcessor, MockProcessor, ): # ③ Patch 上下文供给方法避免 MCP 工具调用 with patch.object( ConstellationAgent, context_provision, new_callableAsyncMock ): session GalaxySession( taskTest task: analyze data and generate insights, should_evaluateTrue, idtest_session_001, clientmock_client, initial_requestPlease help me analyze the sales data and provide insights, )代码依据 test_galaxy_session_proper_mock.py其中的MockProcessor.process会通过 mocks.py 中的create_simple_test_constellation生成一个真实的TaskConstellation并写入全局上下文从而在无 LLM 的情况下驱动完整的状态机流转。此外它还额外验证了Agent 方法级 Mockcontext_provision、_load_mcp_context单独打桩验证 agent 初始化name constellation_agent、status START、状态切换CONTINUE/FINISH与状态机设置事件系统 Mock自定义MockObserver订阅全局事件总线发布CONSTELLATION_STARTED事件后断言观察者收到且字段正确事件类型、source_id。该文件底部还用asyncio.wait_for(session.run(), timeout10.0)为会话运行设置超时保护避免 Mock 场景下死锁挂起——这是异步测试中非常实用的工程技巧。2.5test_galaxy_session_final.py—— 最终综合测试该文件以完整特性演示的方式test_galaxy_session_final.py将八个功能面串成一条流水线会话创建与配置、观察者系统集成、轮次管理、会话状态管理、Agent 与编排器集成、事件系统、会话控制与清理、请求处理。其中有两个细节对理解实现很有帮助观察者能力探测通过hasattr(observer, enable_visualization)/hasattr(observer, session_id)动态检查观察者的可选能力说明观察者接口是鸭子类型友好、按需扩展的请求处理矩阵依次将initial_request设为普通字符串、空字符串、超长字符串逐一调用next_request()与request_to_evaluate()并打印截断后的结果验证请求分发与评估回退逻辑对边界输入的稳定性。三、测试运行方式3.1 从根目录运行所有测试仓库在 tests/run_galaxy_session_tests.py 提供了统一运行器通过subprocess逐个执行测试文件并汇总结果python run_galaxy_session_tests.py该运行器实际执行的文件清单run_galaxy_session_tests.py除了本套件外还包含观察者、可视化集成、DAG 可视化事件与框架汇总等相邻测试覆盖面更广。3.2 运行单个测试文件# 从根目录运行 python tests/galaxy/session/test_galaxy_session.py python tests/galaxy/session/test_galaxy_session_integration.py python tests/galaxy/session/test_galaxy_session_proper_mock.py python tests/galaxy/session/test_galaxy_session_final.py3.3 从测试目录运行cd tests/galaxy/session python test_galaxy_session.py python test_galaxy_session_integration.py python test_galaxy_session_proper_mock.py python test_galaxy_session_final.py3.4 路径配置说明每个测试文件顶部都有统一的路径注入语句test_galaxy_session.pysys.path.append(os.path.join(os.path.dirname(__file__), .., .., ..))test_galaxy_session_proper_mock.py还额外把上级目录加入sys.path以导入共享 Mocktest_galaxy_session_proper_mock.pysys.path.append(os.path.join(os.path.dirname(__file__), .., ..)) from galaxy.mocks import MockConstellationAgent, MockTaskConstellationOrchestrator这就是 README 注意事项中测试文件已正确配置sys.path以访问 UFO 模块的落地实现它保证了无论从根目录还是测试目录启动都能正确解析galaxy.*与ufo.*的导入。四、测试结果示例与运行器输出套件运行后的典型输出如下摘自 tests/galaxy/session/README.md Galaxy Session Test Suite Runner ✅ Basic GalaxySession Functionality - PASSED ✅ Integration Tests - PASSED ✅ Proper Mocking Tests - PASSED ✅ Final Comprehensive Tests - PASSED Test Results: 4/4 tests passed All tests passed!运行器main会统计通过数并返回退出码全部通过返回 0否则返回 1便于直接挂入 CI 流水线。README 中给出的 CI 片段如下可在.github/workflows中直接使用- name: Run Galaxy Session Tests run: python run_galaxy_session_tests.py五、结合源码GalaxySession 被测对象的机制拆解理解测试为何这样设计需要回到被测对象本身。以下是测试中反复触碰的几个核心机制。5.1 轮次创建与请求分发create_new_roundgalaxy_session.py从next_request()取请求首轮返回_initial_request之后返回空串因此第二个轮次创建返回None是设计预期而非缺陷多个测试文件都对此做了断言。每个轮次是一个GalaxyRound继承 UFO 的BaseRound。5.2 状态机驱动的轮次执行GalaxyRound.rungalaxy_session.py揭示了轮次如何被驱动# Initialize agent in START state self._agent.set_state(StartConstellationAgentState()) # Run agent state machine until completion while not self.is_finished(): await self._agent.handle(self._context) # 执行当前状态 self.state self._agent.state.next_state(self._agent) # 推导下一状态 self._agent.set_state(self.state) await asyncio.sleep(0.01) # 防忙等is_finished同时受三个条件约束galaxy_session.py强制结束标志、状态机is_round_end()、以及ContextNames.SESSION_STEP galaxy_config.constellation.MAX_STEP。MAX_STEP来自 config/galaxy/constellation.yaml默认值为 15并支持LOG_TO_MARKDOWN控制是否将轨迹输出为 Markdown会话结束时经GalaxyTrajectory.to_markdown写入log_path。5.3 观察者系统一次装配、订阅全局事件_setup_observersgalaxy_session.py在会话构造时一次性注册五类观察者观察者类职责源码位置ConstellationProgressObserver任务进度更新base_observer.pySessionMetricsObserver性能指标统计base_observer.pyDAGVisualizationObserver星座 DAG 可视化dag_visualization_observer.pyAgentOutputObserver处理 Agent 响应/动作事件并交给 presenter 展示agent_output_observer.pyConstellationModificationSynchronizer协调星座修改同步constellation_sync_observer.py这些观察者全部订阅到全局单例事件总线get_event_bus()因此测试中通过get_event_bus()发布ConstellationEvent即可验证事件链路而无须构造真实会话——这正是test_galaxy_session_proper_mock.py中事件系统测试的直接依据。5.4 会话级状态与生命周期错误判定is_error()galaxy_session.py优先检查当前星座状态是否为ConstellationState.FAILED否则回退检查当前轮次状态是否等于AgentStatus.ERROR结束判定is_finished()galaxy_session.py由_finish标志、step MAX_STEP、total_rounds MAX_STEP共同决定强制结束force_finish(reason)galaxy_session.py置_finish、将 agent 状态置为FINISH、记录finish_reason到session_results并级联强制结束当前轮次取消request_cancellation()galaxy_session.py在置标志与强结束之外还会调用_orchestrator.cancel_execution(constellation_id)并清理观察者防止跨会话重复接收事件重置reset()galaxy_session.py清空轮次、结果与星座引用但保留DEVICE_INFO使同一会话实例可复用处理新请求。六、测试方法论要点为什么这样测这套测试的设计原则可以总结为三条直接呼应 README 的测试特点生产环境用真实ConstellationAgent测试环境只 Mock 外部依赖。GalaxySession的_init_agents硬编码创建ConstellationAgent测试通过 patch 其依赖编排器、处理器、MCP 加载来隔离外部世界而非替换 Agent 本身从而保证被测路径与生产一致。不修改生产代码。所有替身注入均通过unittest.mock.patch与patch.object在测试进程内完成生产模块零改动回归风险低。全面覆盖与可维护性。基础功能、集成、错误处理、状态管理与事件系统分层覆盖统一运行器提供清晰的通过/失败汇总与退出码测试日志详尽便于定位失败环节。七、延伸阅读会话实现源码galaxy/session/galaxy_session.py会话观察者模块galaxy/session/observers/全局事件总线galaxy/core/events.pyAgent 实现galaxy/agents/constellation_agent.py共享 Mock 与测试星座构造tests/galaxy/mocks.pyGalaxy 配置含MAX_STEP、LOG_TO_MARKDOWNconfig/galaxy/constellation.yaml观察者相关测试tests/test_session_observers.py、tests/test_session_visualization_integration.py会话单元测试tests/unit/galaxy/session/本测试套件已覆盖 GalaxySession 的全部核心能力并持续可用可作为 Galaxy 框架二次开发与回归验证的基线参考。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考