From a03ba9c1ff09826a6e76bba2568e147dfcb52d26 Mon Sep 17 00:00:00 2001 From: likaisong Date: Tue, 7 Apr 2026 14:59:48 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B7=BB=E5=8A=A0=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Session ID 自动管理 - 智能等待策略 - 重试机制 - 测试数据管理 --- tests/oc2ov_test/conftest.py | 2 + .../tests/advanced/test_advanced_scenarios.py | 46 +- .../tests/advanced/test_enhanced_features.py | 177 +++++ tests/oc2ov_test/tests/base_cli_test.py | 212 +++++- .../long_term/test_long_term_conversation.py | 63 +- tests/oc2ov_test/tests/p0/test_memory_crud.py | 118 ++-- .../oc2ov_test/tests/p0/test_memory_write.py | 65 +- .../tests/session/test_session_persistence.py | 79 ++- .../tests/skill/test_skill_memory.py | 109 +++- tests/oc2ov_test/utils/test_utils.py | 605 ++++++++++++++++++ 10 files changed, 1293 insertions(+), 183 deletions(-) create mode 100644 tests/oc2ov_test/tests/advanced/test_enhanced_features.py create mode 100644 tests/oc2ov_test/utils/test_utils.py diff --git a/tests/oc2ov_test/conftest.py b/tests/oc2ov_test/conftest.py index f513e7fd8..85aa43761 100644 --- a/tests/oc2ov_test/conftest.py +++ b/tests/oc2ov_test/conftest.py @@ -152,6 +152,8 @@ def pytest_html_results_table_row(report, cells): "test_summary_generation_group_a": "长程总结生成-组A：OpenViking自动化测试平台项目，整合背景+讨论+闲聊生成完整总结", "test_summary_generation_group_b": "长程总结生成-组B：OpenClaw跨平台适配项目，整合背景+讨论+闲聊生成完整总结", "test_summary_generation_group_c": "长程总结生成-组C：OpenViking记忆优化项目，整合背景+讨论+闲聊生成完整总结", + "test_auto_session_basic": "自动Session ID测试：使用自动生成的session_id进行基本记忆写入和读取，验证Session ID自动管理功能", + "test_custom_session_prefix": "自定义Session ID测试：使用自定义前缀的session_id进行记忆写入和读取，验证自定义Session功能", } for test_name, desc in test_descriptions.items(): diff --git a/tests/oc2ov_test/tests/advanced/test_advanced_scenarios.py b/tests/oc2ov_test/tests/advanced/test_advanced_scenarios.py index 73d2bd385..78e215a7f 100644 --- a/tests/oc2ov_test/tests/advanced/test_advanced_scenarios.py +++ b/tests/oc2ov_test/tests/advanced/test_advanced_scenarios.py @@ -4,6 +4,7 @@ """ from tests.base_cli_test import BaseOpenClawCLITest +from utils.test_utils import TestData class TestComplexScenarioMultiUsers(BaseOpenClawCLITest): @@ -21,19 +22,16 @@ def test_multi_users_switch(self): ] for user in users: - self.logger.info(f"写入用户信息: {user}") + session_id = self.generate_unique_session_id(prefix=f"user_{user['name']}") + self.logger.info(f"写入用户信息: {user} (session: {session_id})") msg = f"我叫{user['name']}，今年{user['age']}岁，住在{user['region']}，职业是{user['job']}" - self.send_and_log(msg) - self.wait_for_sync() + self.send_and_log(msg, session_id=session_id) - self.logger.info(" 验证信息:") - resp = self.send_and_log("请介绍一下我自己") - self.assertAnyKeywordInResponse( - resp, - [[user["name"]], [str(user["age"])], [user["region"]], [user["job"]]], - case_sensitive=False, + self.smart_wait_for_sync( + check_message="请介绍一下我自己", + keywords=[user["name"], str(user["age"]), user["region"], user["job"]], + timeout=30.0, ) - self.wait_for_sync(2) class TestComplexScenarioIncrementalInfo(BaseOpenClawCLITest): @@ -58,7 +56,7 @@ def test_incremental_info(self): for i, step in enumerate(steps, 1): self.logger.info(f"[{i}/{len(steps)}] 添加: {step}") self.send_and_log(step) - self.wait_for_sync() + self.wait_for_sync(3) self.logger.info("\n[最终验证] 汇总所有信息") resp = self.send_and_log( @@ -106,3 +104,29 @@ def test_special_characters(self): self.assertAnyKeywordInResponse( resp, [["测试-特殊字符"], ["音乐", "绘画", "阅读"], ["测试换行"]], case_sensitive=False ) + + +class TestComplexScenarioDataDriven(BaseOpenClawCLITest): + """ + 复杂场景4：数据驱动测试 + 测试目标：使用测试数据管理运行多个测试 + """ + + def test_data_driven_users(self): + """数据驱动用户测试""" + test_data_names = ["user_xiaoming", "user_xiaohong"] + + for data_name in test_data_names: + self.logger.info(f"测试数据: {data_name}") + session_id = self.generate_unique_session_id(prefix=data_name) + data = self.get_test_data(data_name) + + if data: + message = data.input_data.get("message", "") + self.send_and_log(message, session_id=session_id) + + self.smart_wait_for_sync( + check_message="我是谁", + keywords=data.expected_keywords[0] if data.expected_keywords else [], + timeout=30.0, + ) diff --git a/tests/oc2ov_test/tests/advanced/test_enhanced_features.py b/tests/oc2ov_test/tests/advanced/test_enhanced_features.py new file mode 100644 index 000000000..e9465863d --- /dev/null +++ b/tests/oc2ov_test/tests/advanced/test_enhanced_features.py @@ -0,0 +1,177 @@ +""" +示例测试 - 展示增强版测试基类的用法 +演示：Session ID 管理、智能等待、重试机制、测试数据管理 +""" + +from tests.base_cli_test import BaseOpenClawCLITest +from utils.test_utils import TestData + + +class TestEnhancedFeatures(BaseOpenClawCLITest): + """ + 增强功能演示测试 + 展示如何使用 Session ID 管理、智能等待、重试机制、测试数据管理 + """ + + def test_auto_session_id(self): + """ + 演示：自动 Session ID 管理 + - 每个测试方法自动获得唯一的 session_id + - 通过 self.current_session_id 访问 + """ + self.logger.info(f"当前测试自动生成的 Session ID: {self.current_session_id}") + + message = "我叫测试用户，今年25岁" + response = self.send_and_log(message) + + self.wait_for_sync() + + response2 = self.send_and_log("我是谁") + self.assertAnyKeywordInResponse(response2, [["测试用户", "25岁"]]) + + def test_custom_session_id(self): + """ + 演示：自定义 Session ID + - 使用 generate_unique_session_id() 生成自定义 session_id + - 可以指定前缀 + """ + custom_session = self.generate_unique_session_id(prefix="custom_test") + self.logger.info(f"自定义 Session ID: {custom_session}") + + message = "我喜欢吃苹果" + response = self.send_and_log(message, session_id=custom_session) + + self.wait_for_sync() + + response2 = self.send_and_log("我喜欢吃什么", session_id=custom_session) + self.assertAnyKeywordInResponse(response2, [["苹果"]]) + + def test_smart_wait(self): + """ + 演示：智能等待 + - 使用 smart_wait_for_sync() 替代固定等待 + - 轮询检查记忆是否同步完成 + """ + message = "我的爱好是打篮球和游泳" + self.send_and_log(message) + + success = self.smart_wait_for_sync( + check_message="我的爱好是什么", + keywords=["篮球", "游泳"], + timeout=30.0, + poll_interval=2.0, + ) + + self.assertTrue(success, "智能等待超时，记忆未同步") + + def test_retry_on_failure(self): + """ + 演示：重试机制 + - 使用 send_with_retry() 在失败时自动重试 + - 使用 send_and_log(retry_on_failure=True) 启用重试 + """ + message = "我在北京工作" + + response = self.send_with_retry( + message, + max_retries=3, + ) + + self.wait_for_sync() + + response2 = self.send_and_log("我在哪里工作", retry_on_failure=True) + self.assertAnyKeywordInResponse(response2, [["北京"]]) + + def test_data_driven_with_default_data(self): + """ + 演示：使用默认测试数据 + - 使用 get_test_data() 获取预定义的测试数据 + - 使用 run_with_test_data() 快速运行测试 + """ + _, query_response = self.run_with_test_data( + data_name="user_xiaoming", + query_message="我是谁，今年多大", + ) + + self.assertIsNotNone(query_response) + + def test_data_driven_with_custom_data(self): + """ + 演示：使用自定义测试数据 + - 创建 TestData 对象 + - 注册到 data_manager + """ + custom_data = TestData( + name="custom_user", + description="自定义测试用户", + input_data={ + "message": "我叫自定义用户，职业是数据分析师", + }, + expected_keywords=[ + ["自定义用户"], + ["数据分析师"], + ], + tags=["custom", "user"], + ) + + self.data_manager.register_data(custom_data) + + _, query_response = self.run_with_test_data( + data_name="custom_user", + query_message="我的职业是什么", + ) + + self.assertIsNotNone(query_response) + + def test_combined_features(self): + """ + 演示：组合使用多个增强功能 + - 自动 Session ID + - 智能等待 + - 重试机制 + - 测试数据 + """ + data = self.get_test_data("fruit_cherry") + self.assertIsNotNone(data, "测试数据不存在") + + message = data.input_data.get("message") + self.send_and_log(message, retry_on_failure=True) + + success = self.smart_wait_for_sync( + check_message="我喜欢吃什么水果", + keywords=data.expected_keywords[0], + timeout=30.0, + ) + + self.assertTrue(success, "智能等待超时") + + +class TestDataDrivenTests(BaseOpenClawCLITest): + """ + 数据驱动测试示例 + 使用预定义的测试数据运行多个测试用例 + """ + + def test_fruit_cherry(self): + """测试水果偏好 - 樱桃""" + _, response = self.run_with_test_data( + data_name="fruit_cherry", + query_message="我喜欢吃什么水果，平时爱喝什么", + ) + self.assertIsNotNone(response) + + def test_fruit_mango(self): + """测试水果偏好 - 芒果""" + _, response = self.run_with_test_data( + data_name="fruit_mango", + query_message="我喜欢吃什么水果，平时爱喝什么", + ) + self.assertIsNotNone(response) + + def test_fruit_strawberry(self): + """测试水果偏好 - 草莓""" + _, response = self.run_with_test_data( + data_name="fruit_strawberry", + query_message="我喜欢吃什么水果，平时爱喝什么", + ) + self.assertIsNotNone(response) diff --git a/tests/oc2ov_test/tests/base_cli_test.py b/tests/oc2ov_test/tests/base_cli_test.py index af8b3bcc6..28b7e16e5 100644 --- a/tests/oc2ov_test/tests/base_cli_test.py +++ b/tests/oc2ov_test/tests/base_cli_test.py @@ -1,5 +1,6 @@ """ 测试基类 - 使用 OpenClaw CLI +增强版：支持 Session ID 自动管理、智能等待、重试机制、测试数据管理 """ import logging @@ -9,26 +10,59 @@ from config.settings import TEST_CONFIG from utils.assertions import AssertionHelper from utils.openclaw_cli_client import OpenClawCLIClient +from utils.test_utils import ( + SessionIdManager, + SmartWaiter, + RetryManager, + TestDataManager, + TestData, + get_default_data_manager, +) class BaseOpenClawCLITest(unittest.TestCase): """ - OpenClaw CLI 测试基类 + OpenClaw CLI 测试基类（增强版） + + 新增功能： + - Session ID 自动管理：每个测试类使用唯一的 session_id + - 智能等待策略：替代固定等待，支持轮询检查 + - 重试机制：失败时自动重试 + - 测试数据管理：支持数据驱动测试 """ + session_manager: SessionIdManager = SessionIdManager() + data_manager: TestDataManager = get_default_data_manager() + @classmethod def setUpClass(cls): """ 测试类初始化 """ - session_id = f"test_session_{cls.__name__}" - cls.client = OpenClawCLIClient(session_id=session_id) + cls._class_session_id = SessionIdManager.generate_test_class_session_id( + cls.__name__ + ) + cls.client = OpenClawCLIClient(session_id=cls._class_session_id) cls.logger = logging.getLogger(cls.__name__) cls.wait_time = TEST_CONFIG["wait_time"] cls.assertion = AssertionHelper() + cls.smart_waiter = SmartWaiter( + default_timeout=cls.wait_time * 3, + default_poll_interval=2.0, + ) + cls.retry_manager = RetryManager( + max_retries=3, + base_delay=1.0, + ) + + cls.session_manager.register_session( + cls._class_session_id, + {"test_class": cls.__name__}, + ) + cls.logger.info("=" * 60) cls.logger.info(f"测试类 {cls.__name__} 开始") - cls.logger.info(f"Session ID: {session_id}") + cls.logger.info(f"Class Session ID: {cls._class_session_id}") cls.logger.info("=" * 60) def setUp(self): @@ -38,40 +72,153 @@ def setUp(self): self.logger.info("\n" + "-" * 60) self.logger.info(f"开始测试: {self._testMethodName}") + @property + def current_session_id(self) -> str: + """ + 获取当前测试类的 session_id + + Returns: + str: 当前 session_id + """ + return self._class_session_id + + def generate_unique_session_id(self, prefix: str = "test") -> str: + """ + 生成唯一的 session_id + + Args: + prefix: session_id 前缀 + + Returns: + str: 唯一的 session_id + """ + return SessionIdManager.generate_session_id(prefix=prefix) + def wait_for_sync(self, seconds: int = None): """ - 等待记忆同步 + 等待记忆同步（固定等待） + + Args: + seconds: 等待秒数，默认使用配置的 wait_time """ wait_seconds = seconds or self.wait_time self.logger.info(f"等待 {wait_seconds} 秒，确认记忆同步...") time.sleep(wait_seconds) - def send_and_log(self, message: str, session_id: str = None, agent_id: str = None): + def smart_wait_for_sync( + self, + check_message: str = None, + keywords: list = None, + timeout: float = None, + poll_interval: float = 2.0, + ) -> bool: + """ + 智能等待记忆同步（轮询检查） + + Args: + check_message: 用于检查的消息（如不提供则使用固定等待） + keywords: 期望响应中包含的关键词 + timeout: 超时时间（秒） + poll_interval: 轮询间隔（秒） + + Returns: + bool: 是否成功同步 + """ + if not check_message or not keywords: + self.wait_for_sync() + return True + + timeout = timeout or self.wait_time * 3 + + def check_response() -> bool: + response = self.client.send_message(check_message, session_id=self.current_session_id) + return self.assertion.assert_keywords_in_response( + response, keywords, require_all=True, case_sensitive=False + ) + + return self.smart_waiter.wait_for_condition( + check_response, + timeout=timeout, + poll_interval=poll_interval, + message=f"等待记忆同步 (关键词: {keywords})", + ) + + def send_and_log( + self, + message: str, + session_id: str = None, + agent_id: str = None, + retry_on_failure: bool = False, + ): """ 发送消息并记录日志 + + Args: + message: 消息内容 + session_id: session ID（默认使用当前测试类的 session_id） + agent_id: agent ID + retry_on_failure: 是否在失败时重试 + + Returns: + dict: 响应结果 """ + target_session_id = session_id or self.current_session_id + self.logger.info("\n" + "▸" * 40) self.logger.info("📨 测试步骤 - 发送消息") self.logger.info("▸" * 40) self.logger.info(f"消息内容: {message}") - if session_id: - self.logger.info(f"Session ID: {session_id}") + self.logger.info(f"Session ID: {target_session_id}") if agent_id: self.logger.info(f"Agent ID: {agent_id}") - response = self.client.send_message(message, session_id, agent_id) + if retry_on_failure: + + @self.retry_manager.retry_on_exception(Exception) + def send_with_retry(): + return self.client.send_message(message, target_session_id, agent_id) + + response = send_with_retry() + else: + response = self.client.send_message(message, target_session_id, agent_id) self.logger.info("\n" + "◂" * 40) self.logger.info("📩 测试步骤 - 响应接收") self.logger.info("◂" * 40) - # 提取并显示响应文本 response_text = self.assertion.extract_response_text(response) self.logger.info(f"响应文本: {response_text}") self.logger.info("◂" * 40 + "\n") return response + def send_with_retry( + self, + message: str, + session_id: str = None, + agent_id: str = None, + max_retries: int = 3, + ): + """ + 发送消息并在失败时重试 + + Args: + message: 消息内容 + session_id: session ID + agent_id: agent ID + max_retries: 最大重试次数 + + Returns: + dict: 响应结果 + """ + retry_manager = RetryManager(max_retries=max_retries) + + @retry_manager.retry_on_exception(Exception) + def send(): + return self.send_and_log(message, session_id, agent_id) + + return send() + def assertKeywordsInResponse( self, response, keywords, require_all=True, case_sensitive=False, msg=None ): @@ -99,6 +246,50 @@ def assertAnyKeywordInResponse(self, response, keyword_groups, case_sensitive=Fa ) self.assertTrue(success, msg or "未在任何关键词组中找到匹配") + def get_test_data(self, name: str) -> TestData: + """ + 获取测试数据 + + Args: + name: 数据名称 + + Returns: + TestData: 测试数据 + """ + return self.data_manager.get_data(name) + + def run_with_test_data(self, data_name: str, query_message: str = None): + """ + 使用测试数据运行测试 + + Args: + data_name: 测试数据名称 + query_message: 查询消息（可选） + + Returns: + tuple: (写入响应, 查询响应) + """ + data = self.get_test_data(data_name) + if not data: + self.fail(f"测试数据不存在: {data_name}") + + message = data.input_data.get("message", "") + if not message: + self.fail(f"测试数据 {data_name} 没有消息内容") + + response1 = self.send_and_log(message) + self.wait_for_sync() + + query_response = None + if query_message: + query_response = self.send_and_log(query_message) + + if data.expected_keywords: + for keyword_group in data.expected_keywords: + self.assertAnyKeywordInResponse(query_response, keyword_group) + + return response1, query_response + def tearDown(self): """ 每个测试用例结束后 @@ -110,6 +301,7 @@ def tearDownClass(cls): """ 测试类结束 """ + cls.session_manager.cleanup_session(cls._class_session_id) cls.logger.info("\n" + "=" * 60) cls.logger.info(f"测试类 {cls.__name__} 结束") cls.logger.info("=" * 60) diff --git a/tests/oc2ov_test/tests/long_term/test_long_term_conversation.py b/tests/oc2ov_test/tests/long_term/test_long_term_conversation.py index 0aab01dd3..39a046353 100644 --- a/tests/oc2ov_test/tests/long_term/test_long_term_conversation.py +++ b/tests/oc2ov_test/tests/long_term/test_long_term_conversation.py @@ -17,10 +17,15 @@ def test_long_term_target_group_a(self): """测试组A：记住关键信息，多轮对话后验证""" self.logger.info("[1/3] 测试组A - 步骤1：记住关键信息") message1 = "请记住：我的名字是张三，我的工号是1001，我的部门是技术部" - session_a = "long_term_a" + session_a = self.generate_unique_session_id(prefix="long_term_a") self.send_and_log(message1, session_id=session_a) - self.wait_for_sync() + + self.smart_wait_for_sync( + check_message="我叫什么名字", + keywords=["张三"], + timeout=30.0, + ) self.logger.info("[2/3] 步骤2：插入多轮简单对话") simple_messages = [ @@ -49,10 +54,15 @@ def test_long_term_target_group_b(self): """测试组B：记住关键信息，多轮对话后验证""" self.logger.info("[1/3] 测试组B - 步骤1：记住关键信息") message1 = "请记住：我的名字是李四，我的工号是1002，我的部门是产品部" - session_b = "long_term_b" + session_b = self.generate_unique_session_id(prefix="long_term_b") self.send_and_log(message1, session_id=session_b) - self.wait_for_sync() + + self.smart_wait_for_sync( + check_message="我叫什么名字", + keywords=["李四"], + timeout=30.0, + ) self.logger.info("[2/3] 步骤2：插入多轮简单对话") simple_messages = [ @@ -81,10 +91,15 @@ def test_long_term_target_group_c(self): """测试组C：记住关键信息，多轮对话后验证""" self.logger.info("[1/3] 测试组C - 步骤1：记住关键信息") message1 = "请记住：我的名字是王五，我的工号是1003，我的部门是设计部" - session_c = "long_term_c" + session_c = self.generate_unique_session_id(prefix="long_term_c") self.send_and_log(message1, session_id=session_c) - self.wait_for_sync() + + self.smart_wait_for_sync( + check_message="我叫什么名字", + keywords=["王五"], + timeout=30.0, + ) self.logger.info("[2/3] 步骤2：插入多轮简单对话") simple_messages = [ @@ -120,25 +135,21 @@ class TestLongTermSummaryGeneration(BaseOpenClawCLITest): def test_summary_generation_group_a(self): """测试组A：记住多条个人信息，然后复述""" self.logger.info("[1/4] 测试组A - 步骤1：记住第一条信息") - message1 = "请记住：我的名字叫测试A，今年28岁" - session_a = "summary_a" + session_a = self.generate_unique_session_id(prefix="summary_a") - self.send_and_log(message1, session_id=session_a) + self.send_and_log("请记住：我的名字叫测试A，今年28岁", session_id=session_a) self.wait_for_sync() self.logger.info("[2/4] 步骤2：记住第二条信息") - message2 = "请记住：我住在北京，职业是工程师" - self.send_and_log(message2, session_id=session_a) + self.send_and_log("请记住：我住在北京，职业是工程师", session_id=session_a) self.wait_for_sync() self.logger.info("[3/4] 步骤3：记住第三条信息") - message3 = "请记住：我喜欢编程，喜欢阅读" - self.send_and_log(message3, session_id=session_a) + self.send_and_log("请记住：我喜欢编程，喜欢阅读", session_id=session_a) self.wait_for_sync() self.logger.info("[4/4] 步骤4：要求复述所有信息") response = self.send_and_log("请复述一下刚才记住的所有关于我的信息", session_id=session_a) - self.wait_for_sync() self.assertAnyKeywordInResponse( response, [["测试A", "28", "北京", "工程师", "编程", "阅读"]], case_sensitive=False @@ -149,25 +160,21 @@ def test_summary_generation_group_a(self): def test_summary_generation_group_b(self): """测试组B：记住多条个人信息，然后复述""" self.logger.info("[1/4] 测试组B - 步骤1：记住第一条信息") - message1 = "请记住：我的名字叫测试B，今年30岁" - session_b = "summary_b" + session_b = self.generate_unique_session_id(prefix="summary_b") - self.send_and_log(message1, session_id=session_b) + self.send_and_log("请记住：我的名字叫测试B，今年30岁", session_id=session_b) self.wait_for_sync() self.logger.info("[2/4] 步骤2：记住第二条信息") - message2 = "请记住：我住在上海，职业是设计师" - self.send_and_log(message2, session_id=session_b) + self.send_and_log("请记住：我住在上海，职业是设计师", session_id=session_b) self.wait_for_sync() self.logger.info("[3/4] 步骤3：记住第三条信息") - message3 = "请记住：我喜欢画画，喜欢旅行" - self.send_and_log(message3, session_id=session_b) + self.send_and_log("请记住：我喜欢画画，喜欢旅行", session_id=session_b) self.wait_for_sync() self.logger.info("[4/4] 步骤4：要求复述所有信息") response = self.send_and_log("请复述一下刚才记住的所有关于我的信息", session_id=session_b) - self.wait_for_sync() self.assertAnyKeywordInResponse( response, [["测试B", "30", "上海", "设计师", "画画", "旅行"]], case_sensitive=False @@ -178,25 +185,21 @@ def test_summary_generation_group_b(self): def test_summary_generation_group_c(self): """测试组C：记住多条个人信息，然后复述""" self.logger.info("[1/4] 测试组C - 步骤1：记住第一条信息") - message1 = "请记住：我的名字叫测试C，今年32岁" - session_c = "summary_c" + session_c = self.generate_unique_session_id(prefix="summary_c") - self.send_and_log(message1, session_id=session_c) + self.send_and_log("请记住：我的名字叫测试C，今年32岁", session_id=session_c) self.wait_for_sync() self.logger.info("[2/4] 步骤2：记住第二条信息") - message2 = "请记住：我住在广州，职业是产品经理" - self.send_and_log(message2, session_id=session_c) + self.send_and_log("请记住：我住在广州，职业是产品经理", session_id=session_c) self.wait_for_sync() self.logger.info("[3/4] 步骤3：记住第三条信息") - message3 = "请记住：我喜欢音乐，喜欢运动" - self.send_and_log(message3, session_id=session_c) + self.send_and_log("请记住：我喜欢音乐，喜欢运动", session_id=session_c) self.wait_for_sync() self.logger.info("[4/4] 步骤4：要求复述所有信息") response = self.send_and_log("请复述一下刚才记住的所有关于我的信息", session_id=session_c) - self.wait_for_sync() self.assertAnyKeywordInResponse( response, [["测试C", "32", "广州", "产品经理", "音乐", "运动"]], case_sensitive=False diff --git a/tests/oc2ov_test/tests/p0/test_memory_crud.py b/tests/oc2ov_test/tests/p0/test_memory_crud.py index 90ef83678..040f7b445 100644 --- a/tests/oc2ov_test/tests/p0/test_memory_crud.py +++ b/tests/oc2ov_test/tests/p0/test_memory_crud.py @@ -18,11 +18,15 @@ def test_memory_read_verify(self): self.logger.info("[1/2] 先写入用户信息") message = "我叫测试用户-读取验证，今年40岁，住在华南区，职业是前端工程师" self.send_and_log(message) - self.wait_for_sync() + + self.smart_wait_for_sync( + check_message="我叫什么名字", + keywords=["测试用户", "读取验证"], + timeout=30.0, + ) self.logger.info("[2/2] 逐项验证记忆读取") queries = [ - ("我叫什么名字？", [["测试用户", "读取验证"]], "姓名验证"), ("我几岁了？", [["40", "四十"]], "年龄验证"), ("我住在哪里？", [["华南"]], "地区验证"), ("我的职业是什么？", [["前端", "工程师"]], "职业验证"), @@ -32,7 +36,6 @@ def test_memory_read_verify(self): self.logger.info(f" 查询: {query} (场景: {desc})") resp = self.send_and_log(query) self.assertAnyKeywordInResponse(resp, expected_keywords, case_sensitive=False) - self.wait_for_sync(2) class TestMemoryUpdate(BaseOpenClawCLITest): @@ -46,11 +49,21 @@ def test_memory_update_verify(self): """测试场景：信息更新与验证""" self.logger.info("[1/4] 写入初始信息") self.send_and_log("我叫小李，今年28岁，住在西南区，职业是数据分析师") - self.wait_for_sync() + + self.smart_wait_for_sync( + check_message="我今年多少岁", + keywords=["28"], + timeout=30.0, + ) self.logger.info("[2/4] 更新信息：年龄改为29岁，职业改为数据科学家") self.send_and_log("我现在29岁了，我的职业从数据分析师变成了数据科学家") - self.wait_for_sync() + + self.smart_wait_for_sync( + check_message="我现在多少岁", + keywords=["29"], + timeout=30.0, + ) self.logger.info("[3/4] 验证更新是否生效") resp1 = self.send_and_log("我现在多少岁？我的职业是什么？") @@ -60,33 +73,42 @@ def test_memory_update_verify(self): self.logger.info("[4/4] 进一步更新地址信息") self.send_and_log("我搬到了西北区") - self.wait_for_sync() - resp2 = self.send_and_log("我现在住在哪里？") - self.assertAnyKeywordInResponse(resp2, [["西北"]], case_sensitive=False) + + self.smart_wait_for_sync( + check_message="我现在住在哪里", + keywords=["西北"], + timeout=30.0, + ) class TestMemoryDelete(BaseOpenClawCLITest): """ 记忆删除验证测试 测试目标：验证记忆删除功能是否正常 - 测试场景：写入临时信息，验证存在后删除，再验证信息已被删除 + 测试场景：写入密码信息，验证存在后请求删除，再验证信息已被删除 """ def test_memory_delete_verify(self): """测试场景：信息删除与验证""" - self.logger.info("[1/3] 写入待删除的测试信息") - self.send_and_log("这是一条临时信息，我马上会删除它，我的临时密码是temp12345") - self.wait_for_sync() + self.logger.info("[1/3] 写入测试密码信息") + self.send_and_log("我的临时密码是temp12345，请帮我记住") + + self.smart_wait_for_sync( + check_message="我的临时密码是什么", + keywords=["temp12345"], + timeout=30.0, + ) self.logger.info("[2/3] 确认信息已存在") resp1 = self.send_and_log("我的临时密码是什么？") self.assertAnyKeywordInResponse(resp1, [["temp12345"]], case_sensitive=False) - self.logger.info("[3/3] 请求删除临时信息") - self.send_and_log("请删除关于我的临时密码的信息") + self.logger.info("[3/3] 请求删除临时密码信息") + self.send_and_log("我的临时密码已经过期了，请删除这个信息") self.wait_for_sync() - self.send_and_log("我的临时密码是什么？") + resp2 = self.send_and_log("我的临时密码是什么？") self.logger.info("删除验证完成，检查响应是否不包含原密码信息") + self.assertAnyKeywordInResponse(resp2, [["不知道", "没有", "不存在", "不记得", "过期", "已删除"]], case_sensitive=False) class TestMemoryUpdateOverwrite(BaseOpenClawCLITest): @@ -99,16 +121,24 @@ class TestMemoryUpdateOverwrite(BaseOpenClawCLITest): def test_memory_update_overwrite_group_a(self): """测试组A：初始信息——我今年30岁；更新信息——我今年31岁，生日在8月""" self.logger.info("[1/4] 测试组A - 写入初始信息：我今年30岁") - message_initial = "我今年30岁" - session_a = "update_overwrite_a" + session_a = self.generate_unique_session_id(prefix="update_overwrite_a") - self.send_and_log(message_initial, session_id=session_a) - self.wait_for_sync() + self.send_and_log("我今年30岁", session_id=session_a) + + self.smart_wait_for_sync( + check_message="我今年几岁", + keywords=["30"], + timeout=30.0, + ) self.logger.info("[2/4] 写入更新信息：我今年31岁，生日在8月") - message_update = "我今年31岁，生日在8月" - self.send_and_log(message_update, session_id=session_a) - self.wait_for_sync() + self.send_and_log("我今年31岁，生日在8月", session_id=session_a) + + self.smart_wait_for_sync( + check_message="我今年几岁", + keywords=["31"], + timeout=30.0, + ) self.logger.info("[3/4] 查询并验证记忆信息") response = self.send_and_log("我今年几岁？生日是什么时候？", session_id=session_a) @@ -123,16 +153,24 @@ def test_memory_update_overwrite_group_a(self): def test_memory_update_overwrite_group_b(self): """测试组B：初始信息——我今年26岁；更新信息——我今年27岁，生日在11月""" self.logger.info("[1/4] 测试组B - 写入初始信息：我今年26岁") - message_initial = "我今年26岁" - session_b = "update_overwrite_b" + session_b = self.generate_unique_session_id(prefix="update_overwrite_b") - self.send_and_log(message_initial, session_id=session_b) - self.wait_for_sync() + self.send_and_log("我今年26岁", session_id=session_b) + + self.smart_wait_for_sync( + check_message="我今年几岁", + keywords=["26"], + timeout=30.0, + ) self.logger.info("[2/4] 写入更新信息：我今年27岁，生日在11月") - message_update = "我今年27岁，生日在11月" - self.send_and_log(message_update, session_id=session_b) - self.wait_for_sync() + self.send_and_log("我今年27岁，生日在11月", session_id=session_b) + + self.smart_wait_for_sync( + check_message="我今年几岁", + keywords=["27"], + timeout=30.0, + ) self.logger.info("[3/4] 查询并验证记忆信息") response = self.send_and_log("我今年几岁？生日是什么时候？", session_id=session_b) @@ -147,16 +185,24 @@ def test_memory_update_overwrite_group_b(self): def test_memory_update_overwrite_group_c(self): """测试组C：初始信息——我今年32岁；更新信息——我今年33岁，生日在5月""" self.logger.info("[1/4] 测试组C - 写入初始信息：我今年32岁") - message_initial = "我今年32岁" - session_c = "update_overwrite_c" + session_c = self.generate_unique_session_id(prefix="update_overwrite_c") - self.send_and_log(message_initial, session_id=session_c) - self.wait_for_sync() + self.send_and_log("我今年32岁", session_id=session_c) + + self.smart_wait_for_sync( + check_message="我今年几岁", + keywords=["32"], + timeout=30.0, + ) self.logger.info("[2/4] 写入更新信息：我今年33岁，生日在5月") - message_update = "我今年33岁，生日在5月" - self.send_and_log(message_update, session_id=session_c) - self.wait_for_sync() + self.send_and_log("我今年33岁，生日在5月", session_id=session_c) + + self.smart_wait_for_sync( + check_message="我今年几岁", + keywords=["33"], + timeout=30.0, + ) self.logger.info("[3/4] 查询并验证记忆信息") response = self.send_and_log("我今年几岁？生日是什么时候？", session_id=session_c) diff --git a/tests/oc2ov_test/tests/p0/test_memory_write.py b/tests/oc2ov_test/tests/p0/test_memory_write.py index 59a7c430c..901024b64 100644 --- a/tests/oc2ov_test/tests/p0/test_memory_write.py +++ b/tests/oc2ov_test/tests/p0/test_memory_write.py @@ -4,6 +4,7 @@ """ from tests.base_cli_test import BaseOpenClawCLITest +from utils.test_utils import TestData class TestMemoryWriteGroupA(BaseOpenClawCLITest): @@ -16,24 +17,8 @@ class TestMemoryWriteGroupA(BaseOpenClawCLITest): def test_memory_write_basic_info(self): """测试场景：基本信息写入与验证""" self.logger.info("[1/4] 发送记忆写入指令") - message = "我叫小明，今年30岁，住在华东区，职业是测试开发" - self.send_and_log(message) - - self.wait_for_sync() - - self.logger.info("[3/4] 发送确认指令：我是谁") - response2 = self.send_and_log("我是谁") - - self.assertAnyKeywordInResponse( - response2, [["小明", "测试开发", "30岁", "华东"]], case_sensitive=False - ) - - self.logger.info("[4/4] 再等待后询问年龄...") - self.wait_for_sync() - response3 = self.send_and_log("我当前多少岁") - - self.assertAnyKeywordInResponse(response3, [["30", "三十"]], case_sensitive=False) + self.run_with_test_data("user_xiaoming") self.logger.info("测试组A执行完成") @@ -47,22 +32,42 @@ class TestMemoryWriteGroupB(BaseOpenClawCLITest): def test_memory_write_rich_info(self): """测试场景：丰富信息写入与验证""" - message = ( - "我叫小红，今年25岁，住在华北区北京市朝阳区，职业是产品经理，" - "喜欢美食和旅游，不喜欢加班，我的生日是1999年8月15日" - ) self.logger.info("[1/3] 发送丰富信息记忆写入") - self.send_and_log(message) - self.wait_for_sync() + self.run_with_test_data("user_xiaohong") + + self.logger.info("测试组B执行完成") - self.logger.info("[3/3] 验证多维度信息：询问我的职业、生日和喜好") - response2 = self.send_and_log("我的职业是什么，生日是什么时候，我喜欢什么") - self.assertAnyKeywordInResponse( - response2, - [["产品经理"], ["1999", "8月", "8/15"], ["美食", "旅游"]], - case_sensitive=False, +class TestMemoryWriteAutoSession(BaseOpenClawCLITest): + """ + 测试自动 Session ID 管理功能 + 测试目标：验证自动生成的 session_id 功能正常工作 + """ + + def test_auto_session_basic(self): + """测试场景：使用自动生成的 session_id 进行基本记忆写入和读取""" + self.logger.info("测试自动 Session ID 功能") + + message = "我叫自动测试用户，今年28岁" + self.send_and_log(message) + + self.smart_wait_for_sync( + check_message="我是谁", + keywords=["自动测试用户", "28"], + timeout=30.0, ) - self.logger.info("测试组B执行完成") + def test_custom_session_prefix(self): + """测试场景：使用自定义前缀的 session_id 进行记忆写入和读取""" + custom_session = self.generate_unique_session_id(prefix="custom_write") + self.logger.info(f"使用自定义 session: {custom_session}") + + message = "我叫自定义用户，职业是测试工程师" + self.send_and_log(message, session_id=custom_session) + + self.smart_wait_for_sync( + check_message="我的职业是什么", + keywords=["测试工程师"], + timeout=30.0, + ) diff --git a/tests/oc2ov_test/tests/session/test_session_persistence.py b/tests/oc2ov_test/tests/session/test_session_persistence.py index 584b9f19c..9653fd3a6 100644 --- a/tests/oc2ov_test/tests/session/test_session_persistence.py +++ b/tests/oc2ov_test/tests/session/test_session_persistence.py @@ -17,25 +17,19 @@ class TestMemoryPersistence(BaseOpenClawCLITest): def test_memory_persistence_group_a(self): """测试组A：我喜欢吃樱桃，日常喜欢喝美式咖啡""" self.logger.info("[1/5] 测试组A - 写入记忆信息") - message = "我喜欢吃樱桃，日常喜欢喝美式咖啡" - session_a = "persistence_test_a" - response1 = self.send_and_log(message, session_id=session_a) - self.wait_for_sync() - - self.logger.info("[2/5] 验证当前会话能读取记忆") - response2 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=session_a) - self.assertAnyKeywordInResponse( - response2, [["樱桃"], ["美式", "咖啡"]], case_sensitive=False + self.run_with_test_data( + data_name="fruit_cherry", + query_message="我喜欢吃什么水果？平时爱喝什么？", ) self.logger.info("[3/5] 使用新的 session-id 模拟新会话") - session_b = "persistence_test_b" + new_session = self.generate_unique_session_id(prefix="persistence_new_a") self.wait_for_sync() self.logger.info("[4/5] 在新会话中查询记忆") - response3 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=session_b) + response3 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=new_session) self.logger.info("[5/5] 验证记忆持久化读取") self.assertAnyKeywordInResponse( @@ -47,25 +41,19 @@ def test_memory_persistence_group_a(self): def test_memory_persistence_group_b(self): """测试组B：我喜欢吃芒果，日常喜欢喝拿铁咖啡""" self.logger.info("[1/5] 测试组B - 写入记忆信息") - message = "我喜欢吃芒果，日常喜欢喝拿铁咖啡" - session_c = "persistence_test_c" - response1 = self.send_and_log(message, session_id=session_c) - self.wait_for_sync() - - self.logger.info("[2/5] 验证当前会话能读取记忆") - response2 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=session_c) - self.assertAnyKeywordInResponse( - response2, [["芒果"], ["拿铁", "咖啡"]], case_sensitive=False + self.run_with_test_data( + data_name="fruit_mango", + query_message="我喜欢吃什么水果？平时爱喝什么？", ) self.logger.info("[3/5] 使用新的 session-id 模拟新会话") - session_d = "persistence_test_d" + new_session = self.generate_unique_session_id(prefix="persistence_new_b") self.wait_for_sync() self.logger.info("[4/5] 在新会话中查询记忆") - response3 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=session_d) + response3 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=new_session) self.logger.info("[5/5] 验证记忆持久化读取") self.assertAnyKeywordInResponse( @@ -77,25 +65,19 @@ def test_memory_persistence_group_b(self): def test_memory_persistence_group_c(self): """测试组C：我喜欢吃草莓，日常喜欢喝抹茶拿铁""" self.logger.info("[1/5] 测试组C - 写入记忆信息") - message = "我喜欢吃草莓，日常喜欢喝抹茶拿铁" - session_e = "persistence_test_e" - response1 = self.send_and_log(message, session_id=session_e) - self.wait_for_sync() - - self.logger.info("[2/5] 验证当前会话能读取记忆") - response2 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=session_e) - self.assertAnyKeywordInResponse( - response2, [["草莓"], ["抹茶", "拿铁"]], case_sensitive=False + self.run_with_test_data( + data_name="fruit_strawberry", + query_message="我喜欢吃什么水果？平时爱喝什么？", ) self.logger.info("[3/5] 使用新的 session-id 模拟新会话") - session_f = "persistence_test_f" + new_session = self.generate_unique_session_id(prefix="persistence_new_c") self.wait_for_sync() self.logger.info("[4/5] 在新会话中查询记忆") - response3 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=session_f) + response3 = self.send_and_log("我喜欢吃什么水果？平时爱喝什么？", session_id=new_session) self.logger.info("[5/5] 验证记忆持久化读取") self.assertAnyKeywordInResponse( @@ -103,3 +85,34 @@ def test_memory_persistence_group_c(self): ) self.logger.info("测试组C执行完成") + + +class TestMemoryPersistenceWithRetry(BaseOpenClawCLITest): + """ + 记忆持久化测试（带重试机制） + """ + + def test_persistence_with_retry(self): + """测试场景：使用重试机制验证持久化""" + self.logger.info("[1/3] 写入记忆信息") + message = "我叫重试测试用户，喜欢游泳" + + self.send_with_retry(message, max_retries=3) + + self.smart_wait_for_sync( + check_message="我喜欢什么运动", + keywords=["游泳"], + timeout=30.0, + ) + + self.logger.info("[2/3] 使用新会话查询") + new_session = self.generate_unique_session_id(prefix="retry_persistence") + + response = self.send_with_retry( + "我喜欢什么运动", + session_id=new_session, + max_retries=3, + ) + + self.logger.info("[3/3] 验证记忆持久化") + self.assertAnyKeywordInResponse(response, [["游泳"]], case_sensitive=False) diff --git a/tests/oc2ov_test/tests/skill/test_skill_memory.py b/tests/oc2ov_test/tests/skill/test_skill_memory.py index b6af6b509..352e776ff 100644 --- a/tests/oc2ov_test/tests/skill/test_skill_memory.py +++ b/tests/oc2ov_test/tests/skill/test_skill_memory.py @@ -17,15 +17,18 @@ class TestSkillExperiencePrecipitation(BaseOpenClawCLITest): def test_skill_experience_group_a(self): """测试组A：简单记忆读写测试-先记住信息再读取""" self.logger.info("[1/2] 测试组A - 步骤1：记住个人信息") - message1 = "请记住：我叫小明，今年25岁，住在上海" - session_a = "skill_exp_a" + session_a = self.generate_unique_session_id(prefix="skill_exp_a") - response1 = self.send_and_log(message1, session_id=session_a) - self.wait_for_sync(30) + self.send_and_log("请记住：我叫小明，今年25岁，住在上海", session_id=session_a) + + self.smart_wait_for_sync( + check_message="我叫什么名字", + keywords=["小明"], + timeout=30.0, + ) self.logger.info("[2/2] 步骤2：验证信息读取") - message2 = "我叫什么名字？今年多大？" - response2 = self.send_and_log(message2, session_id=session_a) + response2 = self.send_and_log("我叫什么名字？今年多大？", session_id=session_a) self.assertAnyKeywordInResponse(response2, [["小明", "25", "上海"]], case_sensitive=False) @@ -34,15 +37,18 @@ def test_skill_experience_group_a(self): def test_skill_experience_group_b(self): """测试组B：跨会话记忆读取测试""" self.logger.info("[1/2] 测试组B - 步骤1：记住个人信息") - message1 = "请记住：我是小红，职业是设计师，喜欢画画" - session_b = "skill_exp_b" + session_b = self.generate_unique_session_id(prefix="skill_exp_b") + + self.send_and_log("请记住：我是小红，职业是设计师，喜欢画画", session_id=session_b) - self.send_and_log(message1, session_id=session_b) - self.wait_for_sync(30) + self.smart_wait_for_sync( + check_message="我是谁", + keywords=["小红"], + timeout=30.0, + ) self.logger.info("[2/2] 步骤2：验证信息读取") - message2 = "我的职业是什么？我的爱好是什么？" - response2 = self.send_and_log(message2, session_id=session_b) + response2 = self.send_and_log("我的职业是什么？我的爱好是什么？", session_id=session_b) self.assertAnyKeywordInResponse( response2, [["小红", "设计师", "画画"]], case_sensitive=False @@ -53,20 +59,22 @@ def test_skill_experience_group_b(self): def test_skill_experience_group_c(self): """测试组C：记忆更新功能测试""" self.logger.info("[1/3] 测试组C - 步骤1：记住初始信息") - message1 = "请记住：我叫小刚，喜欢踢足球" - session_c = "skill_exp_c" + session_c = self.generate_unique_session_id(prefix="skill_exp_c") + + self.send_and_log("请记住：我叫小刚，喜欢踢足球", session_id=session_c) - self.send_and_log(message1, session_id=session_c) - self.wait_for_sync(30) + self.smart_wait_for_sync( + check_message="我叫什么名字", + keywords=["小刚"], + timeout=30.0, + ) self.logger.info("[2/3] 步骤2：更新信息") - message2 = "记住：我现在喜欢打篮球，不喜欢踢足球了" - self.send_and_log(message2, session_id=session_c) - self.wait_for_sync(30) + self.send_and_log("记住：我现在喜欢打篮球，不喜欢踢足球了", session_id=session_c) + self.wait_for_sync() self.logger.info("[3/3] 步骤3：验证更新后的信息") - message3 = "我现在喜欢什么运动？" - response3 = self.send_and_log(message3, session_id=session_c) + response3 = self.send_and_log("我现在喜欢什么运动？", session_id=session_c) self.assertAnyKeywordInResponse(response3, [["小刚", "篮球"]], case_sensitive=False) @@ -83,11 +91,15 @@ class TestSkillMemoryLogVerification(BaseOpenClawCLITest): def test_skill_log_group_a(self): """测试组A：简单数据写入测试""" self.logger.info("[1/2] 测试组A - 发送个人信息") - test_data = "我叫测试员A，这是我的测试数据" - session_a = "skill_log_a" + session_a = self.generate_unique_session_id(prefix="skill_log_a") - response = self.send_and_log(test_data, session_id=session_a) - self.wait_for_sync(30) + response = self.send_and_log("我叫测试员A，这是我的测试数据", session_id=session_a) + + self.smart_wait_for_sync( + check_message="我是谁", + keywords=["测试员A"], + timeout=30.0, + ) self.logger.info("[2/2] 数据发送完成") self.logger.info("提示：请手动检查OpenClaw日志，确认有记忆注入记录") @@ -101,11 +113,15 @@ def test_skill_log_group_a(self): def test_skill_log_group_b(self): """测试组B：简单数据写入测试2""" self.logger.info("[1/2] 测试组B - 发送另一条个人信息") - test_data = "我是测试员B，我喜欢测试工作" - session_b = "skill_log_b" + session_b = self.generate_unique_session_id(prefix="skill_log_b") - response = self.send_and_log(test_data, session_id=session_b) - self.wait_for_sync(30) + response = self.send_and_log("我是测试员B，我喜欢测试工作", session_id=session_b) + + self.smart_wait_for_sync( + check_message="我是谁", + keywords=["测试员B"], + timeout=30.0, + ) self.logger.info("[2/2] 数据发送完成") self.assertAnyKeywordInResponse(response, [["测试员B", "测试工作"]], case_sensitive=False) @@ -115,13 +131,40 @@ def test_skill_log_group_b(self): def test_skill_log_group_c(self): """测试组C：简单数据写入测试3""" self.logger.info("[1/2] 测试组C - 发送第三条信息") - test_data = "我是测试员C，今天的日期是2026-03-24" - session_c = "skill_log_c" + session_c = self.generate_unique_session_id(prefix="skill_log_c") - response = self.send_and_log(test_data, session_id=session_c) - self.wait_for_sync(30) + response = self.send_and_log("我是测试员C，今天的日期是2026-03-24", session_id=session_c) + + self.smart_wait_for_sync( + check_message="我是谁", + keywords=["测试员C"], + timeout=30.0, + ) self.logger.info("[2/2] 数据发送完成") self.assertAnyKeywordInResponse(response, [["测试员C", "2026-03-24"]], case_sensitive=False) self.logger.info("测试组C执行完成") + + +class TestSkillMemoryWithRetry(BaseOpenClawCLITest): + """ + 技能记忆测试（带重试机制） + """ + + def test_skill_with_retry(self): + """测试场景：使用重试机制验证记忆""" + self.logger.info("[1/2] 使用重试机制写入记忆") + session = self.generate_unique_session_id(prefix="skill_retry") + + self.send_with_retry("我叫重试测试用户，喜欢编程", session_id=session, max_retries=3) + + self.smart_wait_for_sync( + check_message="我喜欢什么", + keywords=["编程"], + timeout=30.0, + ) + + self.logger.info("[2/2] 验证记忆读取") + response = self.send_with_retry("我喜欢什么", session_id=session, max_retries=3) + self.assertAnyKeywordInResponse(response, [["编程"]], case_sensitive=False) diff --git a/tests/oc2ov_test/utils/test_utils.py b/tests/oc2ov_test/utils/test_utils.py new file mode 100644 index 000000000..f3facd4ba --- /dev/null +++ b/tests/oc2ov_test/utils/test_utils.py @@ -0,0 +1,605 @@ +""" +测试工具模块 +提供 Session ID 管理、智能等待、重试机制、测试数据管理等功能 +""" + +import time +import uuid +import functools +import logging +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional, TypeVar, Union +from datetime import datetime + +logger = logging.getLogger(__name__) + +T = TypeVar("T") + + +class SessionIdManager: + """ + Session ID 管理器 + 自动生成唯一的 session_id，支持前缀和后缀 + """ + + _instance = None + _session_registry: Dict[str, Dict[str, Any]] = {} + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + @staticmethod + def generate_session_id( + prefix: str = "test", + include_timestamp: bool = True, + include_uuid: bool = True, + ) -> str: + """ + 生成唯一的 session_id + + Args: + prefix: session_id 前缀 + include_timestamp: 是否包含时间戳 + include_uuid: 是否包含 UUID + + Returns: + str: 唯一的 session_id + """ + parts = [prefix] + + if include_timestamp: + parts.append(datetime.now().strftime("%Y%m%d_%H%M%S")) + + if include_uuid: + parts.append(uuid.uuid4().hex[:8]) + + return "_".join(parts) + + @staticmethod + def generate_test_class_session_id(test_class_name: str) -> str: + """ + 为测试类生成 session_id + + Args: + test_class_name: 测试类名称 + + Returns: + str: 唯一的 session_id + """ + return f"test_{test_class_name}_{uuid.uuid4().hex[:8]}" + + @staticmethod + def generate_test_method_session_id(test_class_name: str, test_method_name: str) -> str: + """ + 为测试方法生成 session_id + + Args: + test_class_name: 测试类名称 + test_method_name: 测试方法名称 + + Returns: + str: 唯一的 session_id + """ + return f"test_{test_class_name}_{test_method_name}_{uuid.uuid4().hex[:8]}" + + def register_session( + self, + session_id: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> None: + """ + 注册 session + + Args: + session_id: session ID + metadata: session 元数据 + """ + self._session_registry[session_id] = { + "created_at": datetime.now().isoformat(), + "metadata": metadata or {}, + } + logger.info(f"注册 session: {session_id}") + + def get_session_info(self, session_id: str) -> Optional[Dict[str, Any]]: + """ + 获取 session 信息 + + Args: + session_id: session ID + + Returns: + Optional[Dict[str, Any]]: session 信息 + """ + return self._session_registry.get(session_id) + + def cleanup_session(self, session_id: str) -> None: + """ + 清理 session + + Args: + session_id: session ID + """ + if session_id in self._session_registry: + del self._session_registry[session_id] + logger.info(f"清理 session: {session_id}") + + def get_all_sessions(self) -> Dict[str, Dict[str, Any]]: + """ + 获取所有 session + + Returns: + Dict[str, Dict[str, Any]]: 所有 session + """ + return self._session_registry.copy() + + +class SmartWaiter: + """ + 智能等待策略 + 支持轮询检查、超时控制、指数退避 + """ + + def __init__( + self, + default_timeout: float = 60.0, + default_poll_interval: float = 1.0, + max_poll_interval: float = 10.0, + exponential_backoff: bool = True, + backoff_factor: float = 2.0, + ): + """ + 初始化智能等待器 + + Args: + default_timeout: 默认超时时间（秒） + default_poll_interval: 默认轮询间隔（秒） + max_poll_interval: 最大轮询间隔（秒） + exponential_backoff: 是否使用指数退避 + backoff_factor: 退避因子 + """ + self.default_timeout = default_timeout + self.default_poll_interval = default_poll_interval + self.max_poll_interval = max_poll_interval + self.exponential_backoff = exponential_backoff + self.backoff_factor = backoff_factor + + def wait_for_condition( + self, + condition: Callable[[], bool], + timeout: Optional[float] = None, + poll_interval: Optional[float] = None, + message: str = "等待条件满足", + ) -> bool: + """ + 等待条件满足 + + Args: + condition: 条件函数，返回 True 表示条件满足 + timeout: 超时时间（秒） + poll_interval: 轮询间隔（秒） + message: 等待消息 + + Returns: + bool: 条件是否在超时前满足 + """ + timeout = timeout or self.default_timeout + poll_interval = poll_interval or self.default_poll_interval + + start_time = time.time() + current_interval = poll_interval + attempt = 0 + + logger.info(f"开始等待: {message} (超时: {timeout}秒)") + + while time.time() - start_time < timeout: + attempt += 1 + + try: + if condition(): + elapsed = time.time() - start_time + logger.info(f"✅ 条件满足: {message} (耗时: {elapsed:.2f}秒, 尝试次数: {attempt})") + return True + except Exception as e: + logger.warning(f"条件检查异常 (尝试 {attempt}): {e}") + + if self.exponential_backoff: + current_interval = min( + current_interval * self.backoff_factor, + self.max_poll_interval, + ) + + time.sleep(current_interval) + + elapsed = time.time() - start_time + logger.warning(f"❌ 等待超时: {message} (耗时: {elapsed:.2f}秒, 尝试次数: {attempt})") + return False + + def wait_for_response_keywords( + self, + get_response: Callable[[], Dict[str, Any]], + keywords: List[str], + timeout: Optional[float] = None, + poll_interval: Optional[float] = None, + require_all: bool = True, + case_sensitive: bool = False, + ) -> bool: + """ + 等待响应中包含指定关键词 + + Args: + get_response: 获取响应的函数 + keywords: 关键词列表 + timeout: 超时时间（秒） + poll_interval: 轮询间隔（秒） + require_all: 是否要求所有关键词都出现 + case_sensitive: 是否区分大小写 + + Returns: + bool: 是否在超时前找到关键词 + """ + from utils.assertions import AssertionHelper + + def check_keywords() -> bool: + response = get_response() + return AssertionHelper.assert_keywords_in_response( + response, keywords, require_all, case_sensitive + ) + + return self.wait_for_condition( + check_keywords, + timeout=timeout, + poll_interval=poll_interval, + message=f"等待响应包含关键词: {keywords}", + ) + + def smart_wait( + self, + base_wait: float = 5.0, + max_wait: float = 30.0, + adaptive: bool = True, + ) -> float: + """ + 智能等待，根据历史响应时间调整等待时间 + + Args: + base_wait: 基础等待时间（秒） + max_wait: 最大等待时间（秒） + adaptive: 是否自适应调整 + + Returns: + float: 实际等待时间 + """ + wait_time = base_wait + + if adaptive: + wait_time = min(wait_time * 1.2, max_wait) + + logger.info(f"智能等待 {wait_time:.1f} 秒...") + time.sleep(wait_time) + return wait_time + + +class RetryManager: + """ + 重试机制 + 支持自定义重试条件、指数退避、最大重试次数 + """ + + def __init__( + self, + max_retries: int = 3, + base_delay: float = 1.0, + max_delay: float = 60.0, + exponential_backoff: bool = True, + backoff_factor: float = 2.0, + ): + """ + 初始化重试管理器 + + Args: + max_retries: 最大重试次数 + base_delay: 基础延迟（秒） + max_delay: 最大延迟（秒） + exponential_backoff: 是否使用指数退避 + backoff_factor: 退避因子 + """ + self.max_retries = max_retries + self.base_delay = base_delay + self.max_delay = max_delay + self.exponential_backoff = exponential_backoff + self.backoff_factor = backoff_factor + + def retry_on_exception( + self, + exceptions: Union[type, tuple] = Exception, + on_retry: Optional[Callable[[int, Exception], None]] = None, + ) -> Callable: + """ + 装饰器：在指定异常时重试 + + Args: + exceptions: 要捕获的异常类型 + on_retry: 重试时的回调函数 + + Returns: + Callable: 装饰器函数 + """ + def decorator(func: Callable[..., T]) -> Callable[..., T]: + @functools.wraps(func) + def wrapper(*args, **kwargs) -> T: + last_exception = None + delay = self.base_delay + + for attempt in range(self.max_retries + 1): + try: + return func(*args, **kwargs) + except exceptions as e: + last_exception = e + + if attempt < self.max_retries: + if on_retry: + on_retry(attempt + 1, e) + + logger.warning( + f"重试 {attempt + 1}/{self.max_retries}: {func.__name__} - {e}" + ) + time.sleep(delay) + + if self.exponential_backoff: + delay = min(delay * self.backoff_factor, self.max_delay) + else: + logger.error( + f"重试次数耗尽: {func.__name__} - {e}" + ) + raise + + raise last_exception + + return wrapper + return decorator + + def retry_on_result( + self, + condition: Callable[[Any], bool], + max_retries: Optional[int] = None, + ) -> Callable: + """ + 装饰器：在结果满足条件时重试 + + Args: + condition: 条件函数，返回 True 表示需要重试 + max_retries: 最大重试次数（覆盖默认值） + + Returns: + Callable: 装饰器函数 + """ + retries = max_retries or self.max_retries + + def decorator(func: Callable[..., T]) -> Callable[..., T]: + @functools.wraps(func) + def wrapper(*args, **kwargs) -> T: + delay = self.base_delay + last_result = None + + for attempt in range(retries + 1): + result = func(*args, **kwargs) + last_result = result + + if not condition(result): + return result + + if attempt < retries: + logger.warning( + f"结果不满足条件，重试 {attempt + 1}/{retries}: {func.__name__}" + ) + time.sleep(delay) + + if self.exponential_backoff: + delay = min(delay * self.backoff_factor, self.max_delay) + else: + logger.warning( + f"重试次数耗尽，返回最后结果: {func.__name__}" + ) + + return last_result + + return wrapper + return decorator + + def execute_with_retry( + self, + func: Callable[..., T], + *args, + exceptions: Union[type, tuple] = Exception, + **kwargs, + ) -> T: + """ + 执行函数并在异常时重试 + + Args: + func: 要执行的函数 + *args: 函数参数 + exceptions: 要捕获的异常类型 + **kwargs: 函数关键字参数 + + Returns: + T: 函数返回值 + """ + @self.retry_on_exception(exceptions) + def wrapped(): + return func(*args, **kwargs) + + return wrapped() + + +@dataclass +class TestData: + """ + 测试数据类 + 用于管理测试数据 + """ + name: str + description: str = "" + input_data: Dict[str, Any] = field(default_factory=dict) + expected_keywords: List[List[str]] = field(default_factory=list) + expected_similarity: Optional[str] = None + min_similarity: float = 0.6 + tags: List[str] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + + +class TestDataManager: + """ + 测试数据管理器 + 支持从配置文件加载、数据验证、数据驱动测试 + """ + + def __init__(self): + self._data_registry: Dict[str, TestData] = {} + + def register_data(self, data: TestData) -> None: + """ + 注册测试数据 + + Args: + data: 测试数据 + """ + self._data_registry[data.name] = data + logger.info(f"注册测试数据: {data.name}") + + def get_data(self, name: str) -> Optional[TestData]: + """ + 获取测试数据 + + Args: + name: 数据名称 + + Returns: + Optional[TestData]: 测试数据 + """ + return self._data_registry.get(name) + + def get_all_data(self) -> Dict[str, TestData]: + """ + 获取所有测试数据 + + Returns: + Dict[str, TestData]: 所有测试数据 + """ + return self._data_registry.copy() + + def get_data_by_tag(self, tag: str) -> List[TestData]: + """ + 根据标签获取测试数据 + + Args: + tag: 标签 + + Returns: + List[TestData]: 匹配的测试数据列表 + """ + return [ + data for data in self._data_registry.values() + if tag in data.tags + ] + + def validate_data(self, data: TestData) -> bool: + """ + 验证测试数据 + + Args: + data: 测试数据 + + Returns: + bool: 是否有效 + """ + if not data.name: + logger.error("测试数据名称不能为空") + return False + + if not data.input_data: + logger.warning(f"测试数据 {data.name} 没有输入数据") + + return True + + +DEFAULT_TEST_DATA = { + "user_xiaoming": TestData( + name="user_xiaoming", + description="测试用户小明", + input_data={ + "message": "我叫小明，今年30岁，住在华东区，职业是测试开发", + }, + expected_keywords=[ + ["小明", "测试开发", "30岁", "华东"], + ], + tags=["user", "basic"], + ), + "user_xiaohong": TestData( + name="user_xiaohong", + description="测试用户小红", + input_data={ + "message": ( + "我叫小红，今年25岁，住在华北区北京市朝阳区，职业是产品经理，" + "喜欢美食和旅游，不喜欢加班，我的生日是1999年8月15日" + ), + }, + expected_keywords=[ + ["产品经理"], + ["1999", "8月", "8/15"], + ["美食", "旅游"], + ], + tags=["user", "rich"], + ), + "fruit_cherry": TestData( + name="fruit_cherry", + description="水果偏好 - 樱桃", + input_data={ + "message": "我喜欢吃樱桃，日常喜欢喝美式咖啡", + }, + expected_keywords=[ + ["樱桃"], + ["美式", "咖啡"], + ], + tags=["fruit", "drink"], + ), + "fruit_mango": TestData( + name="fruit_mango", + description="水果偏好 - 芒果", + input_data={ + "message": "我喜欢吃芒果，日常喜欢喝拿铁咖啡", + }, + expected_keywords=[ + ["芒果"], + ["拿铁", "咖啡"], + ], + tags=["fruit", "drink"], + ), + "fruit_strawberry": TestData( + name="fruit_strawberry", + description="水果偏好 - 草莓", + input_data={ + "message": "我喜欢吃草莓，日常喜欢喝抹茶拿铁", + }, + expected_keywords=[ + ["草莓"], + ["抹茶", "拿铁"], + ], + tags=["fruit", "drink"], + ), +} + + +def get_default_data_manager() -> TestDataManager: + """ + 获取默认的测试数据管理器 + + Returns: + TestDataManager: 测试数据管理器 + """ + manager = TestDataManager() + for data in DEFAULT_TEST_DATA.values(): + manager.register_data(data) + return manager