test: 修复测试套件结构问题
- 将 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 处理)
This commit is contained in:
root
@ -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
|
||||
)
|
||||
@ -9,60 +9,26 @@ os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
|
||||
|
||||
|
||||
class TestPacketTypeEnum:
|
||||
"""测试 PacketType 枚举"""
|
||||
|
||||
def test_packet_type_process_message_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.PROCESS_MESSAGE is not None
|
||||
assert PacketType.PROCESS_MESSAGE.value == "process_message"
|
||||
pt = getattr(PacketType, packet_type)
|
||||
assert pt is not None
|
||||
assert pt.value == expected_value
|
||||
|
||||
def test_packet_type_execute_tool_exists(self):
|
||||
def test_packet_type_count(self):
|
||||
from core import PacketType
|
||||
assert PacketType.EXECUTE_TOOL is not None
|
||||
assert PacketType.EXECUTE_TOOL.value == "execute_tool"
|
||||
|
||||
def test_packet_type_get_status_exists(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()
|
||||
@ -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)
|
||||
"""Tests for integration layer"""
|
||||
|
||||
164
tests/test_integration/test_integration.py
Normal file
164
tests/test_integration/test_integration.py
Normal file
@ -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)
|
||||
@ -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)
|
||||
"""Tests for UI layer"""
|
||||
|
||||
242
tests/test_ui/test_ui.py
Normal file
242
tests/test_ui/test_ui.py
Normal file
@ -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)
|
||||
Reference in New Issue
Block a user