refactor: 统一 Packet 通信协议 + 后端配置管理 + UI 清理
- 合并 server.py 到 core/__init__.py，使用统一 Packet 协议 - 后端管理配置持久化 (~/.trulymem/config.json) - 前端移除 ConfigService，通过 BackendClient 与后端通信 - 删除 UI 中冗余的 AI 推理逻辑 (chat_service, tool_service, message_handler) - 删除 core/tools 重复文件 (tool_executor, tool_limiter) - 提示词管理器支持用户自定义 (~/.trulyemem/system_prompt.md) - 启动入口优化配置路径逻辑 - 更新测试覆盖 (42 tests) - 更新文档
This commit is contained in:
root
@ -9,304 +9,300 @@ os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
|
||||
|
||||
|
||||
class TestPacketTypeEnum:
|
||||
def test_packet_type_message_exists(self):
|
||||
"""测试 PacketType 枚举"""
|
||||
|
||||
def test_packet_type_process_message_exists(self):
|
||||
from core import PacketType
|
||||
assert PacketType.MESSAGE is not None
|
||||
assert PacketType.MESSAGE.value == "message"
|
||||
assert PacketType.PROCESS_MESSAGE is not None
|
||||
assert PacketType.PROCESS_MESSAGE.value == "process_message"
|
||||
|
||||
def test_packet_type_config_exists(self):
|
||||
def test_packet_type_execute_tool_exists(self):
|
||||
from core import PacketType
|
||||
assert PacketType.CONFIG is not None
|
||||
assert PacketType.CONFIG.value == "config"
|
||||
assert PacketType.EXECUTE_TOOL is not None
|
||||
assert PacketType.EXECUTE_TOOL.value == "execute_tool"
|
||||
|
||||
def test_packet_type_tool_exists(self):
|
||||
def test_packet_type_get_status_exists(self):
|
||||
from core import PacketType
|
||||
assert PacketType.TOOL is not None
|
||||
assert PacketType.TOOL.value == "tool"
|
||||
assert PacketType.GET_STATUS is not None
|
||||
assert PacketType.GET_STATUS.value == "get_status"
|
||||
|
||||
def test_packet_type_status_exists(self):
|
||||
def test_packet_type_get_config_exists(self):
|
||||
from core import PacketType
|
||||
assert PacketType.STATUS is not None
|
||||
assert PacketType.STATUS.value == "status"
|
||||
assert PacketType.GET_CONFIG is not None
|
||||
assert PacketType.GET_CONFIG.value == "get_config"
|
||||
|
||||
def test_packet_type_history_exists(self):
|
||||
def test_packet_type_set_config_exists(self):
|
||||
from core import PacketType
|
||||
assert PacketType.HISTORY is not None
|
||||
assert PacketType.HISTORY.value == "history"
|
||||
assert PacketType.SET_CONFIG is not None
|
||||
assert PacketType.SET_CONFIG.value == "set_config"
|
||||
|
||||
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 "message" in values
|
||||
assert "config" in values
|
||||
assert "tool" in values
|
||||
assert "status" in values
|
||||
assert "history" in values
|
||||
assert len(values) == 5
|
||||
assert "process_message" in values
|
||||
assert "execute_tool" in values
|
||||
assert "get_status" in values
|
||||
assert "get_config" in values
|
||||
assert "set_config" in values
|
||||
assert "get_history" in values
|
||||
assert "save_history" in values
|
||||
assert "shutdown" in values
|
||||
assert len(values) == 8
|
||||
|
||||
|
||||
class TestPacketCreation:
|
||||
"""测试 Packet 创建"""
|
||||
|
||||
def test_packet_with_id_and_type(self):
|
||||
from core import Packet, PacketType
|
||||
packet = Packet(id="test-1", type=PacketType.MESSAGE, body={"message": "hello"})
|
||||
packet = Packet(id="test-1", type=PacketType.PROCESS_MESSAGE, body={"user_input": "hello"})
|
||||
assert packet.id == "test-1"
|
||||
assert packet.type == PacketType.MESSAGE
|
||||
assert packet.type == PacketType.PROCESS_MESSAGE
|
||||
|
||||
def test_packet_body(self):
|
||||
from core import Packet, PacketType
|
||||
body = {"message": "test", "extra": "data"}
|
||||
packet = Packet(id="test-2", type=PacketType.CONFIG, body=body)
|
||||
body = {"user_input": "test", "extra": "data"}
|
||||
packet = Packet(id="test-2", type=PacketType.EXECUTE_TOOL, body=body)
|
||||
assert packet.body == body
|
||||
|
||||
def test_packet_with_empty_body(self):
|
||||
from core import Packet, PacketType
|
||||
packet = Packet(id="test-3", type=PacketType.GET_STATUS, body={})
|
||||
assert packet.body == {}
|
||||
|
||||
def test_packet_created_at_default(self):
|
||||
from core import Packet, PacketType
|
||||
before = time.time()
|
||||
packet = Packet(id="test-3", type=PacketType.STATUS, body={})
|
||||
packet = Packet(id="test-4", type=PacketType.GET_CONFIG, body={})
|
||||
after = time.time()
|
||||
assert before <= packet.created_at <= after
|
||||
|
||||
def test_packet_with_custom_created_at(self):
|
||||
from core import Packet, PacketType
|
||||
custom_time = 1234567890.0
|
||||
packet = Packet(id="test-4", type=PacketType.HISTORY, body={}, created_at=custom_time)
|
||||
assert packet.created_at == custom_time
|
||||
|
||||
class TestPacketResponse:
|
||||
"""测试 PacketResponse"""
|
||||
|
||||
def test_packet_response_success(self):
|
||||
from core import PacketResponse
|
||||
response = PacketResponse(id="resp-1", success=True, data={"result": "ok"})
|
||||
assert response.id == "resp-1"
|
||||
assert response.success is True
|
||||
assert response.data == {"result": "ok"}
|
||||
|
||||
def test_packet_response_error(self):
|
||||
from core import PacketResponse
|
||||
response = PacketResponse(id="resp-2", success=False, error="error msg")
|
||||
assert response.id == "resp-2"
|
||||
assert response.success is False
|
||||
assert response.error == "error msg"
|
||||
|
||||
|
||||
class TestBackendServerInit:
|
||||
def test_create_server_defaults(self):
|
||||
class TestBackendServerCreation:
|
||||
"""测试 BackendServer 创建"""
|
||||
|
||||
def test_backend_server_init(self):
|
||||
from core import BackendServer
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
assert server._db_path == ":memory:"
|
||||
assert server._use_embedded_db is True
|
||||
assert server._graph is None
|
||||
assert server._client is None
|
||||
assert server._tool_limiter is None
|
||||
|
||||
def test_backend_server_default_params(self):
|
||||
from core import BackendServer
|
||||
server = BackendServer()
|
||||
assert server._db_path == "graph_memory.db"
|
||||
assert server._use_embedded_db is True
|
||||
assert server._graph is None
|
||||
assert server._client is None
|
||||
|
||||
|
||||
class TestBackendServerLifecycle:
|
||||
"""测试 BackendServer 生命周期"""
|
||||
|
||||
def test_backend_server_start_stop(self):
|
||||
from core import BackendServer
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
server.start(api_key="")
|
||||
|
||||
assert server._running is True
|
||||
assert server._graph is not None
|
||||
assert server._tool_limiter is not None
|
||||
|
||||
server.shutdown()
|
||||
assert server._running is False
|
||||
|
||||
def test_create_server_custom_db(self):
|
||||
def test_backend_server_start_with_api_key(self):
|
||||
from core import BackendServer
|
||||
server = BackendServer(db_path="custom.db")
|
||||
assert server._db_path == "custom.db"
|
||||
|
||||
def test_create_server_no_embedded(self):
|
||||
from core import BackendServer
|
||||
server = BackendServer(use_embedded_db=False)
|
||||
assert server._use_embedded_db is False
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
server.start(api_key="test-key", base_url="https://api.deepseek.com")
|
||||
|
||||
assert server._client is not None
|
||||
assert server._config["api_key"] == "test-key"
|
||||
|
||||
server.shutdown()
|
||||
|
||||
|
||||
class TestBackendServerStart:
|
||||
def test_start_without_api_key(self):
|
||||
from core import BackendServer
|
||||
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="")
|
||||
assert server._running is True
|
||||
assert server._graph is not None
|
||||
assert server._client is None
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
def test_start_with_api_key(self):
|
||||
from core import BackendServer
|
||||
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="test-key", base_url="https://api.test.com")
|
||||
assert server._running is True
|
||||
assert server._config["api_key"] == "test-key"
|
||||
assert server._config["base_url"] == "https://api.test.com"
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
def test_start_twice_returns_early(self):
|
||||
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="")
|
||||
running_before = server._running
|
||||
server.start(api_key="")
|
||||
running_after = server._running
|
||||
assert running_before is True
|
||||
assert running_after is True
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestBackendServerShutdown:
|
||||
def test_shutdown_stops_server(self):
|
||||
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="")
|
||||
assert server._running is True
|
||||
server.shutdown()
|
||||
assert server._running is False
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
def test_shutdown_closes_graph(self):
|
||||
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="")
|
||||
server.shutdown()
|
||||
assert server._graph is None
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestBackendServerPacketHandling:
|
||||
def test_send_message_without_client(self):
|
||||
from core import BackendServer, Packet, PacketType
|
||||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
||||
db_path = f.name
|
||||
try:
|
||||
server = BackendServer(db_path=db_path)
|
||||
server.start(api_key="")
|
||||
packet = Packet(id="1", type=PacketType.MESSAGE, body={"message": "hello"})
|
||||
response = server.send(packet)
|
||||
assert response.body["success"] is True
|
||||
assert "API Key not configured" in response.body["error"]
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
def test_send_status_packet(self):
|
||||
from core import BackendServer, Packet, PacketType
|
||||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
||||
db_path = f.name
|
||||
try:
|
||||
server = BackendServer(db_path=db_path)
|
||||
server.start(api_key="")
|
||||
packet = Packet(id="2", type=PacketType.STATUS, body={})
|
||||
response = server.send(packet)
|
||||
assert response.body["success"] is True
|
||||
assert response.body["running"] is True
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestBackendClientInit:
|
||||
def test_create_client_with_server(self):
|
||||
class TestBackendClientCreation:
|
||||
"""测试 BackendClient 创建"""
|
||||
|
||||
def test_backend_client_init(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)
|
||||
assert client._server is server
|
||||
assert client._counter == 0
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
client = BackendClient(server)
|
||||
|
||||
assert client._server is server
|
||||
assert client._counter == 0
|
||||
|
||||
|
||||
class TestBackendClientMethods:
|
||||
class TestBackendClientAPI:
|
||||
"""测试 BackendClient API"""
|
||||
|
||||
def test_get_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)
|
||||
status = client.get_status()
|
||||
assert status["success"] is True
|
||||
assert status["running"] is True
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
server = BackendServer(db_path=":memory:", 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
|
||||
|
||||
server.shutdown()
|
||||
|
||||
def test_update_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="new-key", base_url="https://new-api.test.com")
|
||||
assert result["success"] is True
|
||||
assert result["status"] == "config_updated"
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
result = client.update_config(api_key="new-key", base_url="https://api.deepseek.com")
|
||||
assert result.get("success") is True
|
||||
|
||||
server.shutdown()
|
||||
|
||||
def test_save_and_get_history(self):
|
||||
def test_get_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)
|
||||
|
||||
test_messages = [{"role": "user", "content": "hello"}]
|
||||
client.save_history(test_messages)
|
||||
|
||||
history = client.get_history()
|
||||
assert history == test_messages
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
server.start(api_key="test-key")
|
||||
client = BackendClient(server)
|
||||
|
||||
result = client.get_config()
|
||||
assert result.get("success") is True
|
||||
assert result.get("data", {}).get("api_key") == "test-key"
|
||||
|
||||
server.shutdown()
|
||||
|
||||
|
||||
class TestMultipleClients:
|
||||
def test_multiple_clients_thread_safety(self):
|
||||
def test_process_message_no_api_key(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)
|
||||
|
||||
status1 = client.get_status()
|
||||
status2 = client.get_status()
|
||||
|
||||
assert status1["success"] is True
|
||||
assert status2["success"] is True
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
server = BackendServer(db_path=":memory:", use_embedded_db=True)
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
# 无 API key 应该返回错误（success 为 False）
|
||||
result = client.process_message("hello")
|
||||
# 由于 API 调用失败，success 应该是 False
|
||||
assert result.get("success") is False
|
||||
|
||||
server.shutdown()
|
||||
|
||||
def test_execute_tool(self):
|
||||
from core import BackendServer, BackendClient
|
||||
server = BackendServer(db_path=":memory:", 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()
|
||||
|
||||
|
||||
class TestServerStateAfterShutdown:
|
||||
def test_server_state_not_running(self):
|
||||
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="")
|
||||
assert server._running is True
|
||||
server.shutdown()
|
||||
assert server._running is False
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
class TestToolLimiter:
|
||||
"""测试工具限制器"""
|
||||
|
||||
def test_tool_limiter_init(self):
|
||||
from core.tool_limiter import ToolLimiter
|
||||
limiter = ToolLimiter()
|
||||
assert limiter.counts.persona_query == 0
|
||||
assert limiter.counts.persona_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()
|
||||
@ -6,8 +6,10 @@ os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
|
||||
|
||||
|
||||
class TestIntegrationPacketFlow:
|
||||
def test_packet_round_trip_message(self):
|
||||
from core import BackendServer, BackendClient, Packet, PacketType
|
||||
"""测试 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:
|
||||
@ -15,9 +17,10 @@ class TestIntegrationPacketFlow:
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
result = client.send_message("test message")
|
||||
assert result["success"] is True
|
||||
assert "API Key not configured" in result.get("error", "")
|
||||
# 无 API key 时应该返回错误而非抛异常
|
||||
result = client.process_message("test message")
|
||||
assert result.get("success") is False
|
||||
assert "error" in result
|
||||
|
||||
server.shutdown()
|
||||
finally:
|
||||
@ -34,12 +37,12 @@ class TestIntegrationPacketFlow:
|
||||
client = BackendClient(server)
|
||||
|
||||
result = client.update_config(api_key="test-api", base_url="https://test.com")
|
||||
assert result["success"] is True
|
||||
assert result["status"] == "config_updated"
|
||||
assert result.get("success") is True
|
||||
|
||||
status = client.get_status()
|
||||
assert status["config"]["api_key"] == "test-api"
|
||||
assert status["config"]["base_url"] == "https://test.com"
|
||||
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:
|
||||
@ -55,17 +58,18 @@ class TestIntegrationPacketFlow:
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
status = client.get_status()
|
||||
assert status["success"] is True
|
||||
assert status["running"] is True
|
||||
assert status["client_ready"] is False
|
||||
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_history(self):
|
||||
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
|
||||
@ -74,14 +78,8 @@ class TestIntegrationPacketFlow:
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "hello"},
|
||||
{"role": "assistant", "content": "hi there"}
|
||||
]
|
||||
client.save_history(messages)
|
||||
|
||||
retrieved = client.get_history()
|
||||
assert retrieved == messages
|
||||
result = client.execute_tool("memory_introspect", {})
|
||||
assert result.get("success") is True
|
||||
|
||||
server.shutdown()
|
||||
finally:
|
||||
@ -89,8 +87,10 @@ class TestIntegrationPacketFlow:
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestIntegrationSequentialOperations:
|
||||
def test_sequential_config_updates(self):
|
||||
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
|
||||
@ -99,36 +99,38 @@ class TestIntegrationSequentialOperations:
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
client.update_config(api_key="key1", base_url="https://api1.com")
|
||||
status1 = client.get_status()
|
||||
assert status1["config"]["api_key"] == "key1"
|
||||
|
||||
client.update_config(api_key="key2", base_url="https://api2.com")
|
||||
status2 = client.get_status()
|
||||
assert status2["config"]["api_key"] == "key2"
|
||||
# 外部调用多次应该成功
|
||||
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_save_history_override(self):
|
||||
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="")
|
||||
server.start(api_key="fake-key") # 假 key 会失败但不影响测试
|
||||
client = BackendClient(server)
|
||||
|
||||
client.save_history([{"role": "user", "content": "first"}])
|
||||
history1 = client.get_history()
|
||||
assert len(history1) == 1
|
||||
# 内部调用受限，tool_limiter 存在
|
||||
assert server._tool_limiter is not None
|
||||
|
||||
client.save_history([{"role": "user", "content": "second"}])
|
||||
history2 = client.get_history()
|
||||
assert len(history2) == 1
|
||||
assert history2[0]["content"] == "second"
|
||||
# 初始状态
|
||||
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:
|
||||
@ -136,117 +138,10 @@ class TestIntegrationSequentialOperations:
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestIntegrationMultipleClients:
|
||||
def test_two_clients_same_server(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="")
|
||||
|
||||
client1 = BackendClient(server)
|
||||
client2 = BackendClient(server)
|
||||
|
||||
status1 = client1.get_status()
|
||||
status2 = client2.get_status()
|
||||
|
||||
assert status1["success"] is True
|
||||
assert status2["success"] is True
|
||||
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestIntegrationDatabase:
|
||||
def test_server_creates_database(self):
|
||||
from core import BackendServer
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = os.path.join(tmpdir, "test.db")
|
||||
server = BackendServer(db_path=db_path, use_embedded_db=True)
|
||||
server.start(api_key="")
|
||||
|
||||
assert os.path.exists(db_path)
|
||||
|
||||
server.shutdown()
|
||||
assert os.path.exists(db_path)
|
||||
|
||||
def test_server_persists_across_restart(self):
|
||||
from core import BackendServer, BackendClient
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = os.path.join(tmpdir, "persist.db")
|
||||
|
||||
server1 = BackendServer(db_path=db_path, use_embedded_db=True)
|
||||
server1.start(api_key="")
|
||||
server1.shutdown()
|
||||
|
||||
server2 = BackendServer(db_path=db_path, use_embedded_db=True)
|
||||
server2.start(api_key="")
|
||||
assert os.path.exists(db_path)
|
||||
server2.shutdown()
|
||||
|
||||
|
||||
class TestIntegrationEntryPoint:
|
||||
def test_entry_import(self):
|
||||
import trulymem_entry
|
||||
assert trulymem_entry is not None
|
||||
|
||||
def test_entry_has_main(self):
|
||||
import trulymem_entry
|
||||
assert hasattr(trulymem_entry, 'main')
|
||||
assert callable(trulymem_entry.main)
|
||||
|
||||
|
||||
class TestIntegrationAllPacketTypes:
|
||||
def test_message_type_string(self):
|
||||
from core import PacketType
|
||||
assert PacketType.MESSAGE.value == "message"
|
||||
|
||||
def test_config_type_string(self):
|
||||
from core import PacketType
|
||||
assert PacketType.CONFIG.value == "config"
|
||||
|
||||
def test_tool_type_string(self):
|
||||
from core import PacketType
|
||||
assert PacketType.TOOL.value == "tool"
|
||||
|
||||
def test_status_type_string(self):
|
||||
from core import PacketType
|
||||
assert PacketType.STATUS.value == "status"
|
||||
|
||||
def test_history_type_string(self):
|
||||
from core import PacketType
|
||||
assert PacketType.HISTORY.value == "history"
|
||||
|
||||
|
||||
class TestIntegrationErrorHandling:
|
||||
def test_timeout_on_slow_response(self):
|
||||
from core import BackendServer, Packet, PacketType
|
||||
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
||||
db_path = f.name
|
||||
try:
|
||||
server = BackendServer(db_path=db_path)
|
||||
server.start(api_key="")
|
||||
server._running = False
|
||||
|
||||
packet = Packet(id="timeout-test", type=PacketType.MESSAGE, body={"message": "test"})
|
||||
|
||||
original_timeout = 30.0
|
||||
server.send = lambda p, timeout=original_timeout: (
|
||||
setattr(server, '_running', True),
|
||||
server.send(p)
|
||||
)[1]
|
||||
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestIntegrationFullWorkflow:
|
||||
def test_complete_workflow(self):
|
||||
"""测试错误处理"""
|
||||
|
||||
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
|
||||
@ -255,24 +150,30 @@ class TestIntegrationFullWorkflow:
|
||||
server.start(api_key="")
|
||||
client = BackendClient(server)
|
||||
|
||||
status_before = client.get_status()
|
||||
assert status_before["success"] is True
|
||||
|
||||
client.update_config(api_key="workflow-key", base_url="https://workflow.com")
|
||||
status_after_config = client.get_status()
|
||||
assert status_after_config["config"]["api_key"] == "workflow-key"
|
||||
|
||||
history = [{"role": "user", "content": "test workflow"}]
|
||||
client.save_history(history)
|
||||
retrieved_history = client.get_history()
|
||||
assert retrieved_history == history
|
||||
|
||||
message_result = client.send_message("test")
|
||||
assert message_result["success"] is True
|
||||
# 应该返回错误，而不是抛出异常
|
||||
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()
|
||||
status_after_shutdown = client.get_status()
|
||||
assert status_after_shutdown["running"] is False
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
@ -6,7 +6,9 @@ from pathlib import Path
|
||||
os.environ["DEEPSEEK_API_KEY"] = "fake-test-key"
|
||||
|
||||
|
||||
class TestUIImportFull:
|
||||
class TestUIImport:
|
||||
"""测试 UI 模块导入"""
|
||||
|
||||
def test_import_graphmemoryapp(self):
|
||||
from ui import GraphMemoryApp
|
||||
assert GraphMemoryApp is not None
|
||||
@ -15,52 +17,24 @@ class TestUIImportFull:
|
||||
from ui import AppConfig
|
||||
assert AppConfig is not None
|
||||
|
||||
def test_import_from_models(self):
|
||||
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_from_models_config(self):
|
||||
def test_import_config(self):
|
||||
from ui.models.config import AppConfig
|
||||
assert AppConfig is not None
|
||||
|
||||
def test_import_from_models_log_entry(self):
|
||||
def test_import_log_entry(self):
|
||||
from ui.models.log_entry import LogEntry
|
||||
assert LogEntry is not None
|
||||
|
||||
def test_import_from_widgets(self):
|
||||
from ui.widgets.left_panel import LeftPanel
|
||||
from ui.widgets.right_panel import RightPanel
|
||||
from ui.widgets.input_box import InputBox
|
||||
from ui.widgets.message_history import MessageHistory
|
||||
from ui.widgets.status_bar import StatusBar
|
||||
assert LeftPanel is not None
|
||||
assert RightPanel is not None
|
||||
assert InputBox is not None
|
||||
assert MessageHistory is not None
|
||||
assert StatusBar is not None
|
||||
|
||||
def test_import_from_handlers(self):
|
||||
from ui.handlers.focus_handler import FocusHandler
|
||||
from ui.handlers.key_handler import KeyHandler
|
||||
from ui.handlers.message_handler import MessageHandler
|
||||
assert FocusHandler is not None
|
||||
assert KeyHandler is not None
|
||||
assert MessageHandler is not None
|
||||
|
||||
def test_import_from_services(self):
|
||||
from ui.services.config_manager import ConfigManager
|
||||
from ui.services.config_service import ConfigService
|
||||
from ui.services.chat_service import ChatService
|
||||
from ui.services.tool_service import ToolService
|
||||
assert ConfigManager is not None
|
||||
assert ConfigService is not None
|
||||
assert ChatService is not None
|
||||
assert ToolService is not None
|
||||
|
||||
|
||||
class TestAppConfigFull:
|
||||
class TestAppConfig:
|
||||
"""测试配置模型"""
|
||||
|
||||
def test_config_default_values(self):
|
||||
from ui.models.config import AppConfig
|
||||
config = AppConfig()
|
||||
@ -68,25 +42,15 @@ class TestAppConfigFull:
|
||||
assert config.model == "deepseek-chat"
|
||||
assert config.base_url == "https://api.deepseek.com"
|
||||
|
||||
def test_config_from_env_with_key(self):
|
||||
def test_config_from_env(self):
|
||||
from ui.models.config import AppConfig
|
||||
config = AppConfig.from_env()
|
||||
assert "fake-test-key" in config.api_key
|
||||
|
||||
def test_config_from_env_custom(self):
|
||||
os.environ["DEEPSEEK_API_KEY"] = "my-key"
|
||||
os.environ["MODEL_NAME"] = "my-model"
|
||||
os.environ["DEEPSEEK_BASE_URL"] = "https://my-api.com"
|
||||
|
||||
from ui.models.config import AppConfig
|
||||
config = AppConfig.from_env()
|
||||
|
||||
assert config.api_key == "my-key"
|
||||
assert config.model == "my-model"
|
||||
assert config.base_url == "https://my-api.com"
|
||||
|
||||
|
||||
class TestMessageModelFull:
|
||||
class TestMessageModel:
|
||||
"""测试消息模型"""
|
||||
|
||||
def test_message_creation_user(self):
|
||||
from ui.models.message import Message
|
||||
from datetime import datetime
|
||||
@ -99,119 +63,36 @@ class TestMessageModelFull:
|
||||
from ui.models.message import Message
|
||||
msg = Message(role="assistant", content="assistant response")
|
||||
assert msg.role == "assistant"
|
||||
assert msg.content == "assistant response"
|
||||
|
||||
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.role == "assistant"
|
||||
assert msg.tool_calls is not None
|
||||
assert len(msg.tool_calls) == 1
|
||||
assert msg.tool_calls[0].name == "memory_recall"
|
||||
|
||||
def test_message_with_tool_results(self):
|
||||
from ui.models.message import Message, ToolResult
|
||||
tr = ToolResult(tool_call_id="call-1", name="memory_recall", arguments={}, result="result", success=True)
|
||||
msg = Message(role="assistant", content="response", tool_results=[tr])
|
||||
assert msg.tool_results is not None
|
||||
assert len(msg.tool_results) == 1
|
||||
assert msg.tool_results[0].success is True
|
||||
|
||||
|
||||
class TestToolCallModelFull:
|
||||
def test_toolcall_creation(self):
|
||||
from ui.models.message import ToolCall
|
||||
tc = ToolCall(id="call-1", name="memory_recall", arguments={"query": "test"})
|
||||
assert tc.id == "call-1"
|
||||
assert tc.name == "memory_recall"
|
||||
assert tc.arguments["query"] == "test"
|
||||
|
||||
|
||||
class TestToolResultModelFull:
|
||||
def test_toolresult_creation_success(self):
|
||||
from ui.models.message import ToolResult
|
||||
tr = ToolResult(tool_call_id="call-1", name="memory_recall", arguments={}, result="success result", success=True)
|
||||
assert tr.tool_call_id == "call-1"
|
||||
assert tr.name == "memory_recall"
|
||||
assert tr.success is True
|
||||
assert tr.result == "success result"
|
||||
|
||||
def test_toolresult_creation_failure(self):
|
||||
from ui.models.message import ToolResult
|
||||
tr = ToolResult(tool_call_id="call-1", name="memory_recall", arguments={}, result="error", success=False)
|
||||
assert tr.success is False
|
||||
assert tr.result == "error"
|
||||
|
||||
|
||||
class TestLogEntryModelFull:
|
||||
def test_logentry_creation(self):
|
||||
from ui.models.log_entry import LogEntry
|
||||
from datetime import datetime
|
||||
entry = LogEntry(
|
||||
timestamp=datetime.now(),
|
||||
tool_name="memory_recall",
|
||||
arguments={"query": "test"},
|
||||
result="result",
|
||||
duration=0.5
|
||||
)
|
||||
assert entry.tool_name == "memory_recall"
|
||||
assert entry.duration == 0.5
|
||||
|
||||
def test_logentry_args_summary(self):
|
||||
from ui.models.log_entry import LogEntry
|
||||
from datetime import datetime
|
||||
entry = LogEntry(
|
||||
timestamp=datetime.now(),
|
||||
tool_name="memory_recall",
|
||||
arguments={"query": "test query with many characters"},
|
||||
result="result",
|
||||
duration=0.5
|
||||
)
|
||||
summary = entry.args_summary
|
||||
assert isinstance(summary, str)
|
||||
assert "query" in summary
|
||||
|
||||
def test_logentry_result_summary(self):
|
||||
from ui.models.log_entry import LogEntry
|
||||
from datetime import datetime
|
||||
entry = LogEntry(
|
||||
timestamp=datetime.now(),
|
||||
tool_name="memory_recall",
|
||||
arguments={},
|
||||
result="a" * 200,
|
||||
duration=0.5
|
||||
)
|
||||
summary = entry.result_summary
|
||||
assert len(summary) <= 103
|
||||
|
||||
|
||||
class TestAppCSSPathFull:
|
||||
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
|
||||
|
||||
def test_css_path_are_paths(self):
|
||||
from ui import GraphMemoryApp
|
||||
for path in GraphMemoryApp.CSS_PATH:
|
||||
assert isinstance(path, Path)
|
||||
|
||||
|
||||
class TestAppBindingsFull:
|
||||
class TestAppBindings:
|
||||
"""测试 App 快捷键"""
|
||||
|
||||
def test_app_has_bindings(self):
|
||||
from ui import GraphMemoryApp
|
||||
assert hasattr(GraphMemoryApp, 'BINDINGS')
|
||||
assert len(GraphMemoryApp.BINDINGS) > 0
|
||||
|
||||
def test_bindings_have_required_keys(self):
|
||||
from ui import GraphMemoryApp
|
||||
for binding in GraphMemoryApp.BINDINGS:
|
||||
assert hasattr(binding, 'key')
|
||||
assert hasattr(binding, 'action')
|
||||
|
||||
|
||||
class TestWidgetImportsFull:
|
||||
class TestWidgetImports:
|
||||
"""测试组件导入"""
|
||||
|
||||
def test_import_left_panel(self):
|
||||
from ui.widgets.left_panel import LeftPanel
|
||||
assert LeftPanel is not None
|
||||
@ -228,28 +109,14 @@ class TestWidgetImportsFull:
|
||||
from ui.widgets.message_history import MessageHistory
|
||||
assert MessageHistory is not None
|
||||
|
||||
def test_import_message_widget(self):
|
||||
from ui.widgets.message_widget import MessageWidget
|
||||
assert MessageWidget is not None
|
||||
|
||||
def test_import_status_bar(self):
|
||||
from ui.widgets.status_bar import StatusBar
|
||||
assert StatusBar is not None
|
||||
|
||||
def test_import_config_section(self):
|
||||
from ui.widgets.config_section import ConfigSection
|
||||
assert ConfigSection is not None
|
||||
|
||||
def test_import_operation_log(self):
|
||||
from ui.widgets.operation_log import OperationLog
|
||||
assert OperationLog is not None
|
||||
|
||||
def test_import_cypher_query_box(self):
|
||||
from ui.widgets.cypher_query_box import CypherQueryBox
|
||||
assert CypherQueryBox is not None
|
||||
|
||||
|
||||
class TestHandlerImportsFull:
|
||||
class TestHandlerImports:
|
||||
"""测试处理器导入"""
|
||||
|
||||
def test_import_focus_handler(self):
|
||||
from ui.handlers.focus_handler import FocusHandler
|
||||
assert FocusHandler is not None
|
||||
@ -258,30 +125,22 @@ class TestHandlerImportsFull:
|
||||
from ui.handlers.key_handler import KeyHandler
|
||||
assert KeyHandler is not None
|
||||
|
||||
def test_import_message_handler(self):
|
||||
from ui.handlers.message_handler import MessageHandler
|
||||
assert MessageHandler is not None
|
||||
|
||||
|
||||
class TestServiceImportsFull:
|
||||
def test_import_config_manager(self):
|
||||
from ui.services.config_manager import ConfigManager
|
||||
assert ConfigManager 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_chat_service(self):
|
||||
from ui.services.chat_service import ChatService
|
||||
assert ChatService is not None
|
||||
|
||||
def test_import_tool_service(self):
|
||||
from ui.services.tool_service import ToolService
|
||||
assert ToolService is not None
|
||||
def test_import_config_manager(self):
|
||||
from ui.services.config_manager import ConfigManager
|
||||
assert ConfigManager is not None
|
||||
|
||||
|
||||
class TestAppInitializationFull:
|
||||
class TestAppInitialization:
|
||||
"""测试 App 初始化"""
|
||||
|
||||
def test_app_without_backend(self):
|
||||
from ui import GraphMemoryApp
|
||||
app = GraphMemoryApp()
|
||||
@ -305,18 +164,56 @@ class TestAppInitializationFull:
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
class TestConfigManagerFull:
|
||||
def test_config_manager_creation(self):
|
||||
from ui.services.config_manager import ConfigManager
|
||||
cm = ConfigManager()
|
||||
assert cm is not None
|
||||
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
|
||||
|
||||
# 测试 get_status
|
||||
status = client.get_status()
|
||||
assert status.get("success") is True
|
||||
|
||||
# 测试 update_config
|
||||
result = client.update_config(api_key="sk-test", base_url="https://api.deepseek.com")
|
||||
assert result.get("success") is True
|
||||
|
||||
server.shutdown()
|
||||
finally:
|
||||
if os.path.exists(db_path):
|
||||
os.unlink(db_path)
|
||||
|
||||
def test_config_manager_config_path(self):
|
||||
from ui.services.config_manager import ConfigManager
|
||||
cm = ConfigManager()
|
||||
assert cm._config_path is not None
|
||||
|
||||
def test_config_manager_exists_false(self):
|
||||
from ui.services.config_manager import ConfigManager
|
||||
cm = ConfigManager()
|
||||
assert cm.exists() is False
|
||||
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