JOJO
f75e2f07a3
- 删除 compact 文件和 inject guide 中的最近工具记录 - 新结构：用户所有输入（按压缩轮次分段）+ 历次压缩总结 + 最近一次输入 - 用户输入分段标记改为 <第N次压缩> 和 <当前触发的第X次压缩> - 压缩总结提示词迁移到 prompts/deep_compression_summary.txt - deep_compression_records 增加 user_inputs_before 和 summary 字段 - 压缩后清空全部 frozen prompt 缓存 - 修复 wait 模式 in-place 压缩前端不刷新问题 - 更新手动压缩确认弹窗文案 - 修复 _apply_workspace_personalization_preferences 测试 mock 签名 - 清理 context.py 中误导性的主提示词构建参数
126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
|
|
class ServerRefactorSmokeTest(unittest.TestCase):
|
|
def test_chat_flow_imports_and_runner_binding(self):
|
|
import server.chat_flow as chat_flow
|
|
import server.chat_flow_runner as runner
|
|
|
|
self.assertIs(chat_flow.handle_task_with_sender, runner.handle_task_with_sender)
|
|
self.assertTrue(callable(chat_flow.start_chat_task))
|
|
self.assertTrue(callable(chat_flow.run_chat_task_sync))
|
|
|
|
def test_chat_flow_helper_functions(self):
|
|
import server.chat_flow_helpers as helpers
|
|
|
|
terminal = SimpleNamespace(
|
|
deep_thinking_mode=False,
|
|
thinking_mode=True,
|
|
api_client=SimpleNamespace(
|
|
force_thinking_next_call=False,
|
|
skip_thinking_next_call=False,
|
|
last_call_used_thinking=False,
|
|
),
|
|
thinking_fast_interval=2,
|
|
)
|
|
|
|
helpers.apply_thinking_schedule(terminal, default_interval=2, debug_logger=lambda _: None)
|
|
helpers.update_thinking_after_call(terminal, debug_logger=lambda _: None)
|
|
state = helpers.get_thinking_state(terminal)
|
|
|
|
self.assertIsInstance(state, dict)
|
|
self.assertIn("fast_streak", state)
|
|
self.assertTrue(helpers.detect_tool_failure({"success": False}))
|
|
self.assertFalse(helpers.detect_tool_failure({"success": True}))
|
|
|
|
def test_runner_helpers_pure_functions(self):
|
|
import server.chat_flow_runner_helpers as helpers
|
|
|
|
self.assertEqual(helpers.extract_intent_from_partial('{"intent": "fix"}'), "fix")
|
|
self.assertEqual(helpers.resolve_monitor_path({"file_path": " a.txt "}), "a.txt")
|
|
self.assertEqual(helpers.resolve_monitor_memory([1, 2, 3], entry_limit=2), ["1", "2"])
|
|
|
|
def test_conversation_stats_exports(self):
|
|
import server.conversation as conversation
|
|
import server.conversation_stats as stats
|
|
|
|
self.assertTrue(callable(stats.build_admin_dashboard_snapshot))
|
|
self.assertTrue(callable(conversation.compute_workspace_storage))
|
|
self.assertTrue(callable(conversation.collect_upload_events))
|
|
|
|
def test_custom_tools_refresh_disabled_after_dynamic_load(self):
|
|
from core.main_terminal_parts.tools_definition import MainTerminalToolsDefinitionMixin
|
|
|
|
class Registry:
|
|
def reload(self):
|
|
return [
|
|
{
|
|
"id": "demo_tool",
|
|
"description": "Demo tool",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
}
|
|
]
|
|
|
|
class Terminal(MainTerminalToolsDefinitionMixin):
|
|
custom_tools_enabled = True
|
|
user_role = "admin"
|
|
custom_tool_registry = Registry()
|
|
|
|
def __init__(self):
|
|
self.tool_categories_map = {
|
|
"custom": SimpleNamespace(
|
|
tools=[],
|
|
default_enabled=True,
|
|
silent_when_disabled=False,
|
|
)
|
|
}
|
|
self.tool_category_states = {"custom": False}
|
|
self.admin_forced_category_states = {}
|
|
self.disabled_tools = set()
|
|
|
|
def _refresh_disabled_tools(self):
|
|
disabled = set()
|
|
for key, category in self.tool_categories_map.items():
|
|
if not self.tool_category_states.get(key, category.default_enabled):
|
|
disabled.update(category.tools)
|
|
self.disabled_tools = disabled
|
|
|
|
terminal = Terminal()
|
|
custom_tools = terminal._build_custom_tools()
|
|
|
|
self.assertEqual(custom_tools[0]["function"]["name"], "demo_tool")
|
|
self.assertEqual(terminal.tool_categories_map["custom"].tools, ["demo_tool"])
|
|
self.assertIn("demo_tool", terminal.disabled_tools)
|
|
|
|
def test_context_applies_workspace_personalization_preferences(self):
|
|
import server.context as context
|
|
|
|
calls = []
|
|
|
|
class Terminal:
|
|
run_mode = "fast"
|
|
thinking_mode = False
|
|
model_key = None
|
|
|
|
def apply_personalization_preferences(self, config, **kwargs):
|
|
calls.append(config)
|
|
|
|
workspace = SimpleNamespace(data_dir="/tmp/workspace-data")
|
|
original_loader = context.load_personalization_config
|
|
try:
|
|
context.load_personalization_config = lambda data_dir: {
|
|
"disabled_tool_categories": ["custom"]
|
|
}
|
|
context._apply_workspace_personalization_preferences(Terminal(), workspace)
|
|
finally:
|
|
context.load_personalization_config = original_loader
|
|
|
|
self.assertEqual(calls, [{"disabled_tool_categories": ["custom"]}])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|