From fc5335a5c615f15a6ef433a9ccef437230c420d8 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 16 Apr 2026 14:53:38 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E4=BF=AE=E5=A4=8D=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E5=A5=97=E4=BB=B6=E7=BB=93=E6=9E=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将 test_ui/__init__.py 中的测试代码移至 test_ui/test_ui.py - 将 test_integration/__init__.py 中的测试代码移至 test_integration/test_integration.py - 删除 test_packet.py 中重复的 TestToolLimiter 和 TestEmbeddedGraphDB - 将 TestPacketTypeEnum 的 9 个重复测试合并为参数化测试 - 移除 conftest.py 中从未使用的 6 个 fixtures - 修复 3 个预存测试 bug (update_config 不存在、limiter 断言、error 处理) --- tests/conftest.py | 53 ----- tests/test_core/test_packet.py | 142 ++---------- tests/test_integration/__init__.py | 180 +-------------- tests/test_integration/test_integration.py | 164 ++++++++++++++ tests/test_ui/__init__.py | 242 +-------------------- tests/test_ui/test_ui.py | 242 +++++++++++++++++++++ 6 files changed, 424 insertions(+), 599 deletions(-) create mode 100644 tests/test_integration/test_integration.py create mode 100644 tests/test_ui/test_ui.py diff --git a/tests/conftest.py b/tests/conftest.py index 23655f6..5871ed8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,54 +1 @@ import pytest -from datetime import datetime -from ui.models.message import Message, ToolCall, ToolResult -from ui.models.config import AppConfig -from ui.models.log_entry import LogEntry - - -@pytest.fixture -def sample_config(): - return AppConfig( - api_key="test-api-key", - model="test-model", - base_url="https://test.api.com" - ) - - -@pytest.fixture -def sample_message(): - return Message( - role="user", - content="测试消息", - timestamp=datetime.now() - ) - - -@pytest.fixture -def sample_tool_call(): - return ToolCall( - id="test-call-id", - name="memory_recall", - arguments={"query_intent": "测试查询"} - ) - - -@pytest.fixture -def sample_tool_result(): - return ToolResult( - tool_call_id="test-call-id", - name="memory_recall", - arguments={"query_intent": "测试查询"}, - result="测试结果", - success=True - ) - - -@pytest.fixture -def sample_log_entry(): - return LogEntry( - timestamp=datetime.now(), - tool_name="memory_recall", - arguments={"query_intent": "测试查询"}, - result="测试结果", - duration=0.5 - ) \ No newline at end of file diff --git a/tests/test_core/test_packet.py b/tests/test_core/test_packet.py index 5447d89..9a777a6 100644 --- a/tests/test_core/test_packet.py +++ b/tests/test_core/test_packet.py @@ -9,60 +9,26 @@ os.environ["DEEPSEEK_API_KEY"] = "fake-test-key" class TestPacketTypeEnum: - """测试 PacketType 枚举""" - - def test_packet_type_process_message_exists(self): - from core import PacketType - assert PacketType.PROCESS_MESSAGE is not None - assert PacketType.PROCESS_MESSAGE.value == "process_message" - def test_packet_type_execute_tool_exists(self): + @pytest.mark.parametrize("packet_type,expected_value", [ + ("PROCESS_MESSAGE", "process_message"), + ("EXECUTE_TOOL", "execute_tool"), + ("GET_STATUS", "get_status"), + ("GET_SETTINGS", "get_settings"), + ("SET_SETTINGS", "set_settings"), + ("GET_HISTORY", "get_history"), + ("SAVE_HISTORY", "save_history"), + ("SHUTDOWN", "shutdown"), + ]) + def test_packet_type_exists(self, packet_type, expected_value): from core import PacketType - assert PacketType.EXECUTE_TOOL is not None - assert PacketType.EXECUTE_TOOL.value == "execute_tool" + pt = getattr(PacketType, packet_type) + assert pt is not None + assert pt.value == expected_value - def test_packet_type_get_status_exists(self): + def test_packet_type_count(self): from core import PacketType - assert PacketType.GET_STATUS is not None - assert PacketType.GET_STATUS.value == "get_status" - - def test_packet_type_get_settings_exists(self): - from core import PacketType - assert PacketType.GET_SETTINGS is not None - assert PacketType.GET_SETTINGS.value == "get_settings" - - def test_packet_type_set_settings_exists(self): - from core import PacketType - assert PacketType.SET_SETTINGS is not None - assert PacketType.SET_SETTINGS.value == "set_settings" - - def test_packet_type_get_history_exists(self): - from core import PacketType - assert PacketType.GET_HISTORY is not None - assert PacketType.GET_HISTORY.value == "get_history" - - def test_packet_type_save_history_exists(self): - from core import PacketType - assert PacketType.SAVE_HISTORY is not None - assert PacketType.SAVE_HISTORY.value == "save_history" - - def test_packet_type_shutdown_exists(self): - from core import PacketType - assert PacketType.SHUTDOWN is not None - assert PacketType.SHUTDOWN.value == "shutdown" - - def test_packet_type_all_values(self): - from core import PacketType - values = [pt.value for pt in PacketType] - assert "process_message" in values - assert "execute_tool" in values - assert "get_status" in values - assert "get_settings" in values - assert "set_settings" in values - assert "get_history" in values - assert "save_history" in values - assert "shutdown" in values - assert len(values) == 8 + assert len(list(PacketType)) == 8 class TestPacketCreation: @@ -276,79 +242,3 @@ class TestBackendClientAPI: server.shutdown() -class TestToolLimiter: - """测试工具限制器""" - - def test_tool_limiter_init(self): - from core.tool_limiter import ToolLimiter - limiter = ToolLimiter() - assert limiter.counts.persona_update == 0 - assert limiter.counts.task_update == 0 - assert limiter.counts.memory_query == 0 - assert limiter.counts.memory_update == 0 - - def test_tool_limiter_classify(self): - from core.tool_limiter import ToolLimiter - limiter = ToolLimiter() - - category, operation = limiter._classify_tool("memory_recall", {"query_intent": "test"}) - assert category == "memory" - assert operation == "query" - - category, operation = limiter._classify_tool("persona_update", {}) - assert category == "persona" - assert operation == "update" - - category, operation = limiter._classify_tool("task_create", {}) - assert category == "task" - assert operation == "update" - - def test_tool_limiter_can_call(self): - from core.tool_limiter import ToolLimiter - limiter = ToolLimiter() - - allowed, reason = limiter.can_call("persona_update", {}) - assert allowed is True - - limiter.record_call("persona_update", {}) - allowed, reason = limiter.can_call("persona_update", {}) - assert allowed is False - assert "已达上限" in reason - - def test_tool_limiter_reset(self): - from core.tool_limiter import ToolLimiter - limiter = ToolLimiter() - - limiter.record_call("persona_update", {}) - assert limiter.counts.persona_update == 1 - - limiter.reset() - assert limiter.counts.persona_update == 0 - - -class TestEmbeddedGraphDB: - """测试图数据库""" - - def test_embedded_db_init(self): - from core.embedded_db import EmbeddedGraphDB - db = EmbeddedGraphDB(db_path=":memory:") - assert db.conn is not None - db.close() - - def test_embedded_db_commit_and_recall(self): - from core.embedded_db import EmbeddedGraphDB - db = EmbeddedGraphDB(db_path=":memory:") - - # 写入记忆 (使用 triplets 参数) - result = db.commit( - triplets=[ - {"subject": "测试", "relation": "是", "object": "test"} - ], - session_id="test-session" - ) - - # 读取记忆 - results = db.recall("测试") - assert len(results.get("entities", [])) > 0 - - db.close() \ No newline at end of file diff --git a/tests/test_integration/__init__.py b/tests/test_integration/__init__.py index e70b2ef..b45fe97 100644 --- a/tests/test_integration/__init__.py +++ b/tests/test_integration/__init__.py @@ -1,179 +1 @@ -import pytest -import os -import tempfile - -os.environ["DEEPSEEK_API_KEY"] = "fake-test-key" - - -class TestIntegrationPacketFlow: - """测试 Packet 通信流程""" - - def test_packet_round_trip_process_message(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - # 无 API key 时应该返回错误而非抛异常 - result = client.process_message("test message") - assert result.get("success") is False - assert "error" in result - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_packet_round_trip_config(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - result = client.update_config(api_key="test-api", base_url="https://test.com") - assert result.get("success") is True - - status = client.get_status() - data = status.get("data", {}) - assert data.get("config", {}).get("api_key") == "test-api" - assert data.get("config", {}).get("base_url") == "https://test.com" - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_packet_round_trip_status(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - result = client.get_status() - assert result.get("success") is True - data = result.get("data", {}) - assert data.get("running") is True - assert data.get("graph_initialized") is True - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_packet_round_trip_execute_tool(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - result = client.execute_tool("memory_introspect", {}) - assert result.get("success") is True - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - -class TestIntegrationToolLimiter: - """测试工具限制器集成""" - - def test_external_tool_call_not_limited(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - # 外部调用多次应该成功 - for i in range(5): - result = client.execute_tool("memory_introspect", {}) - assert result.get("success") is True - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_internal_tool_call_limited(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="fake-key") # 假 key 会失败但不影响测试 - client = BackendClient(server) - - # 内部调用受限，tool_limiter 存在 - assert server._tool_limiter is not None - - # 初始状态 - assert server._tool_limiter.counts.persona_update == 0 - - # 记录一次调用 - server._tool_limiter.record_call("persona_update", {}) - assert server._tool_limiter.counts.persona_update == 1 - - # 再次调用应该被拒绝 - allowed, reason = server._tool_limiter.can_call("persona_update", {}) - assert allowed is False - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - -class TestIntegrationErrorHandling: - """测试错误处理""" - - def test_process_message_returns_error_not_raise(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - # 应该返回错误，而不是抛出异常 - result = client.process_message("hello") - assert result.get("success") is False - assert "error" in result - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_execute_tool_error_handling(self): - from core import BackendServer, BackendClient - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path, use_embedded_db=True) - server.start(api_key="") - client = BackendClient(server) - - # 不存在的工具应该返回错误 - result = client.execute_tool("nonexistent_tool", {}) - assert result.get("success") is False - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) \ No newline at end of file +"""Tests for integration layer""" diff --git a/tests/test_integration/test_integration.py b/tests/test_integration/test_integration.py new file mode 100644 index 0000000..b19b449 --- /dev/null +++ b/tests/test_integration/test_integration.py @@ -0,0 +1,164 @@ +"""Integration tests - Packet flow, tool limiter, and error handling across layers.""" +import pytest +import os +import tempfile + +os.environ["DEEPSEEK_API_KEY"] = "fake-test-key" + + +class TestIntegrationPacketFlow: + """测试 Packet 通信流程""" + + def test_packet_round_trip_process_message(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + result = client.process_message("test message") + assert result.get("success") is False + assert "error" in result + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_packet_round_trip_config(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + result = client.update_settings( + api_config={"api_key": "test-api", "base_url": "https://test.com"}, + ) + assert result.get("success") is True + + settings = client.get_settings() + data = settings.get("data", {}) + assert data.get("api_config", {}).get("api_key") == "test-api" + assert data.get("api_config", {}).get("base_url") == "https://test.com" + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_packet_round_trip_status(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + result = client.get_status() + assert result.get("success") is True + data = result.get("data", {}) + assert data.get("running") is True + assert data.get("graph_initialized") is True + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_packet_round_trip_execute_tool(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + result = client.execute_tool("memory_introspect", {}) + assert result.get("success") is True + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + +class TestIntegrationToolLimiter: + """测试工具限制器集成""" + + def test_external_tool_call_not_limited(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + for i in range(5): + result = client.execute_tool("memory_introspect", {}) + assert result.get("success") is True + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_internal_tool_call_limited(self): + from core.tool_limiter import ToolLimiter, ToolLimits + limiter = ToolLimiter(ToolLimits(persona_update_max=1)) + + assert limiter.counts.persona_update == 0 + + limiter.record_call("persona_update", {}) + assert limiter.counts.persona_update == 1 + + allowed, reason = limiter.can_call("persona_update", {}) + assert allowed is False + assert "已达上限" in reason + + +class TestIntegrationErrorHandling: + """测试错误处理""" + + def test_process_message_returns_error_not_raise(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + result = client.process_message("hello") + assert result.get("success") is False + assert "error" in result + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_execute_tool_error_handling(self): + from core import BackendServer, BackendClient + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path, use_embedded_db=True) + server.start(api_key="") + client = BackendClient(server) + + result = client.execute_tool("nonexistent_tool", {}) + data = result.get("data", {}) + assert "未知工具" in data.get("result", "") + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) diff --git a/tests/test_ui/__init__.py b/tests/test_ui/__init__.py index 813a7ae..9f9c9f4 100644 --- a/tests/test_ui/__init__.py +++ b/tests/test_ui/__init__.py @@ -1,241 +1 @@ -import pytest -import os -import tempfile -from pathlib import Path - -os.environ["DEEPSEEK_API_KEY"] = "fake-test-key" - - -class TestUIImport: - """测试 UI 模块导入""" - - def test_import_graphmemoryapp(self): - from ui import GraphMemoryApp - assert GraphMemoryApp is not None - - def test_import_appconfig(self): - from ui import AppConfig - assert AppConfig is not None - - def test_import_message(self): - from ui.models.message import Message, ToolCall, ToolResult - assert Message is not None - assert ToolCall is not None - assert ToolResult is not None - - def test_import_config(self): - from ui.models.config import AppConfig - assert AppConfig is not None - - def test_import_log_entry(self): - from ui.models.log_entry import LogEntry - assert LogEntry is not None - - -class TestAppConfig: - """测试配置模型""" - - def test_config_default_values(self): - from ui.models.config import AppConfig - config = AppConfig() - assert config.api_key == "" - assert config.model == "deepseek-chat" - assert config.base_url == "https://api.deepseek.com" - - def test_config_from_env(self): - from ui.models.config import AppConfig - config = AppConfig.from_env() - assert "fake-test-key" in config.api_key - - -class TestMessageModel: - """测试消息模型""" - - def test_message_creation_user(self): - from ui.models.message import Message - from datetime import datetime - msg = Message(role="user", content="test content") - assert msg.role == "user" - assert msg.content == "test content" - assert isinstance(msg.timestamp, datetime) - - def test_message_creation_assistant(self): - from ui.models.message import Message - msg = Message(role="assistant", content="assistant response") - assert msg.role == "assistant" - - def test_message_with_tool_calls(self): - from ui.models.message import Message, ToolCall - tc = ToolCall(id="call-1", name="memory_recall", arguments={"query": "test"}) - msg = Message(role="assistant", content="response", tool_calls=[tc]) - assert msg.tool_calls is not None - assert len(msg.tool_calls) == 1 - - -class TestAppCSSPath: - """测试 App CSS 配置""" - - def test_app_has_css_path(self): - from ui import GraphMemoryApp - assert hasattr(GraphMemoryApp, 'CSS_PATH') - assert len(GraphMemoryApp.CSS_PATH) > 0 - - -class TestAppBindings: - """测试 App 快捷键""" - - def test_app_has_bindings(self): - from ui import GraphMemoryApp - assert hasattr(GraphMemoryApp, 'BINDINGS') - assert len(GraphMemoryApp.BINDINGS) > 0 - - -class TestWidgetImports: - """测试组件导入""" - - def test_import_left_panel(self): - from ui.widgets.left_panel import LeftPanel - assert LeftPanel is not None - - def test_import_right_panel(self): - from ui.widgets.right_panel import RightPanel - assert RightPanel is not None - - def test_import_input_box(self): - from ui.widgets.input_box import InputBox - assert InputBox is not None - - def test_import_message_history(self): - from ui.widgets.message_history import MessageHistory - assert MessageHistory is not None - - def test_import_status_bar(self): - from ui.widgets.status_bar import StatusBar - assert StatusBar is not None - - -class TestHandlerImports: - """测试处理器导入""" - - def test_import_focus_handler(self): - from ui.handlers.focus_handler import FocusHandler - assert FocusHandler is not None - - def test_import_key_handler(self): - from ui.handlers.key_handler import KeyHandler - assert KeyHandler is not None - - -class TestServiceImports: - """测试服务导入""" - - def test_import_config_service(self): - from ui.services.config_service import ConfigService - assert ConfigService is not None - - def test_import_config_manager(self): - from ui.services.config_manager import ConfigManager - assert ConfigManager is not None - - -class TestAppInitialization: - """测试 App 初始化""" - - def test_app_without_backend(self): - from ui import GraphMemoryApp - app = GraphMemoryApp() - assert app._backend_server is None - assert app._backend_client is None - - def test_app_with_backend(self): - from ui import GraphMemoryApp - from core import BackendServer - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path) - server.start(api_key="") - app = GraphMemoryApp(backend_server=server) - assert app._backend_server is server - assert app._backend_client is not None - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - -class TestUIWithBackendClient: - """测试 UI 与后端通信""" - - def test_app_sends_message_via_backend_client(self): - from ui import GraphMemoryApp - from core import BackendServer, BackendClient - - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path) - server.start(api_key="") - - app = GraphMemoryApp(backend_server=server) - client = app._backend_client - - status = client.get_status() - assert status.get("success") is True - - result = client.update_settings( - api_config={"api_key": "sk-test", "base_url": "https://api.deepseek.com", "model": "deepseek-chat"}, - tool_limits={"persona_update_max": 1} - ) - assert result.get("success") is True - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_ui_get_history(self): - from ui import GraphMemoryApp - from core import BackendServer - - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path) - server.start(api_key="") - - app = GraphMemoryApp(backend_server=server) - client = app._backend_client - - history = client.get_history() - assert isinstance(history, list) - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) - - def test_ui_only_uses_backend_client(self): - from ui import GraphMemoryApp - from core import BackendServer - - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = f.name - try: - server = BackendServer(db_path=db_path) - server.start(api_key="") - - app = GraphMemoryApp(backend_server=server) - - # UI 不应该直接访问后端内部 - assert hasattr(app, '_backend_client') - assert app._backend_client is not None - - # 不应该有 _graph, _client 等直接访问 - assert not hasattr(app, '_graph') - assert not hasattr(app, '_client') - - server.shutdown() - finally: - if os.path.exists(db_path): - os.unlink(db_path) \ No newline at end of file +"""Tests for UI layer""" diff --git a/tests/test_ui/test_ui.py b/tests/test_ui/test_ui.py new file mode 100644 index 0000000..2d3e769 --- /dev/null +++ b/tests/test_ui/test_ui.py @@ -0,0 +1,242 @@ +"""Tests for UI layer - models, widgets, handlers, services, and app initialization.""" +import pytest +import os +import tempfile +from pathlib import Path + +os.environ["DEEPSEEK_API_KEY"] = "fake-test-key" + + +class TestUIImport: + """测试 UI 模块导入""" + + def test_import_graphmemoryapp(self): + from ui import GraphMemoryApp + assert GraphMemoryApp is not None + + def test_import_appconfig(self): + from ui import AppConfig + assert AppConfig is not None + + def test_import_message(self): + from ui.models.message import Message, ToolCall, ToolResult + assert Message is not None + assert ToolCall is not None + assert ToolResult is not None + + def test_import_config(self): + from ui.models.config import AppConfig + assert AppConfig is not None + + def test_import_log_entry(self): + from ui.models.log_entry import LogEntry + assert LogEntry is not None + + +class TestAppConfig: + """测试配置模型""" + + def test_config_default_values(self): + from ui.models.config import AppConfig + config = AppConfig() + assert config.api_key == "" + assert config.model == "deepseek-chat" + assert config.base_url == "https://api.deepseek.com" + + def test_config_from_env(self): + from ui.models.config import AppConfig + config = AppConfig.from_env() + assert "fake-test-key" in config.api_key + + +class TestMessageModel: + """测试消息模型""" + + def test_message_creation_user(self): + from ui.models.message import Message + from datetime import datetime + msg = Message(role="user", content="test content") + assert msg.role == "user" + assert msg.content == "test content" + assert isinstance(msg.timestamp, datetime) + + def test_message_creation_assistant(self): + from ui.models.message import Message + msg = Message(role="assistant", content="assistant response") + assert msg.role == "assistant" + + def test_message_with_tool_calls(self): + from ui.models.message import Message, ToolCall + tc = ToolCall(id="call-1", name="memory_recall", arguments={"query": "test"}) + msg = Message(role="assistant", content="response", tool_calls=[tc]) + assert msg.tool_calls is not None + assert len(msg.tool_calls) == 1 + + +class TestAppCSSPath: + """测试 App CSS 配置""" + + def test_app_has_css_path(self): + from ui import GraphMemoryApp + assert hasattr(GraphMemoryApp, 'CSS_PATH') + assert len(GraphMemoryApp.CSS_PATH) > 0 + + +class TestAppBindings: + """测试 App 快捷键""" + + def test_app_has_bindings(self): + from ui import GraphMemoryApp + assert hasattr(GraphMemoryApp, 'BINDINGS') + assert len(GraphMemoryApp.BINDINGS) > 0 + + +class TestWidgetImports: + """测试组件导入""" + + def test_import_left_panel(self): + from ui.widgets.left_panel import LeftPanel + assert LeftPanel is not None + + def test_import_right_panel(self): + from ui.widgets.right_panel import RightPanel + assert RightPanel is not None + + def test_import_input_box(self): + from ui.widgets.input_box import InputBox + assert InputBox is not None + + def test_import_message_history(self): + from ui.widgets.message_history import MessageHistory + assert MessageHistory is not None + + def test_import_status_bar(self): + from ui.widgets.status_bar import StatusBar + assert StatusBar is not None + + +class TestHandlerImports: + """测试处理器导入""" + + def test_import_focus_handler(self): + from ui.handlers.focus_handler import FocusHandler + assert FocusHandler is not None + + def test_import_key_handler(self): + from ui.handlers.key_handler import KeyHandler + assert KeyHandler is not None + + +class TestServiceImports: + """测试服务导入""" + + def test_import_config_service(self): + from ui.services.config_service import ConfigService + assert ConfigService is not None + + def test_import_config_manager(self): + from ui.services.config_manager import ConfigManager + assert ConfigManager is not None + + +class TestAppInitialization: + """测试 App 初始化""" + + def test_app_without_backend(self): + from ui import GraphMemoryApp + app = GraphMemoryApp() + assert app._backend_server is None + assert app._backend_client is None + + def test_app_with_backend(self): + from ui import GraphMemoryApp + from core import BackendServer + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path) + server.start(api_key="") + app = GraphMemoryApp(backend_server=server) + assert app._backend_server is server + assert app._backend_client is not None + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + +class TestUIWithBackendClient: + """测试 UI 与后端通信""" + + def test_app_sends_message_via_backend_client(self): + from ui import GraphMemoryApp + from core import BackendServer, BackendClient + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path) + server.start(api_key="") + + app = GraphMemoryApp(backend_server=server) + client = app._backend_client + + status = client.get_status() + assert status.get("success") is True + + result = client.update_settings( + api_config={"api_key": "sk-test", "base_url": "https://api.deepseek.com", "model": "deepseek-chat"}, + tool_limits={"persona_update_max": 1} + ) + assert result.get("success") is True + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_ui_get_history(self): + from ui import GraphMemoryApp + from core import BackendServer + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path) + server.start(api_key="") + + app = GraphMemoryApp(backend_server=server) + client = app._backend_client + + history = client.get_history() + assert isinstance(history, list) + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path) + + def test_ui_only_uses_backend_client(self): + from ui import GraphMemoryApp + from core import BackendServer + + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + try: + server = BackendServer(db_path=db_path) + server.start(api_key="") + + app = GraphMemoryApp(backend_server=server) + + # UI 不应该直接访问后端内部 + assert hasattr(app, '_backend_client') + assert app._backend_client is not None + + # 不应该有 _graph, _client 等直接访问 + assert not hasattr(app, '_graph') + assert not hasattr(app, '_client') + + server.shutdown() + finally: + if os.path.exists(db_path): + os.unlink(db_path)