agent-Specialization/test/test_server_refactor_smoke.py

57 lines
2.2 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))
if __name__ == "__main__":
unittest.main()