feat: 巻加测试优化功能

- Session ID 自动管理
- 智能等待策略
- 重试机制
- 测试数据管理

Author: kaisongli

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():

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,
+                )

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)