Cleanup: remove redundant files, add comprehensive tests

Deleted:
- install.bat (Windows-only, use pip directly)
- test_message_order.py (outdated, replaced by pytest)
- test_tool_call_flow.py (outdated documentation-style test)

Added tests (57 total, all passing):
- test_embedded_db.py: SQLite database operations (11 tests)
- test_tool_limiter.py: tool call limiting logic (14 tests)
- test_memory_tools.py: tool definitions validation (17 tests)
This commit is contained in:
root
2026-04-12 16:07:39 +08:00
parent 7fe56e8319
commit 28aa696224
6 changed files with 448 additions and 176 deletions

View File

@ -0,0 +1,162 @@
"""嵌入式数据库测试"""
import pytest
import tempfile
import os
from graph_memory_tui.core.embedded_db import EmbeddedGraphDB
@pytest.fixture
def db():
"""创建临时数据库用于测试"""
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
db_path = f.name
db = EmbeddedGraphDB(db_path)
yield db
db.close()
os.unlink(db_path)
def test_db_init(db):
"""测试数据库初始化"""
assert db.conn is not None
assert db.db_path.exists()
def test_commit_and_recall(db):
"""测试写入和检索记忆"""
result = db.commit(
triplets=[
{"subject": "用户", "relation": "喜欢", "object": "Python"},
{"subject": "用户", "relation": "正在学习", "object": "AI"}
],
session_id="test-session",
turn_id=1
)
assert result["created_entities"] >= 2
assert result["created_relations"] >= 2
def test_recall_with_keywords(db):
"""测试关键词检索"""
db.commit(
triplets=[
{"subject": "项目A", "relation": "使用技术", "object": "React"}
]
)
result = db.recall("React")
assert len(result["entities"]) > 0
def test_recall_empty_keywords(db):
"""测试空关键词检索"""
db.commit(
triplets=[
{"subject": "测试实体", "relation": "关系", "object": "测试对象"}
]
)
result = db.recall("")
assert len(result["entities"]) > 0
def test_purge_soft(db):
"""测试软删除"""
db.commit(
triplets=[
{"subject": "待删除", "relation": "测试", "object": "删除内容"}
]
)
result = db.purge(
criteria={"source": "待删除"},
mode="soft"
)
assert result["deleted"] >= 0
assert result["mode"] == "soft"
def test_introspect(db):
"""测试状态查看"""
db.commit(
triplets=[
{"subject": "实体1", "relation": "关系", "object": "实体2"}
]
)
result = db.introspect()
assert "entity_count" in result
assert "relation_count" in result
assert result["entity_count"] >= 1
def test_archive(db):
"""测试归档"""
result = db.archive(days=30)
assert "archived" in result
def test_cleanup_dry_run(db):
"""测试清理(预览模式)"""
result = db.cleanup(dry_run=True)
assert result["dry_run"] is True
assert "deleted_relations" in result
def test_multiple_triplets(db):
"""测试批量写入"""
result = db.commit(
triplets=[
{"subject": "实体A", "relation": "关系1", "object": "实体B"},
{"subject": "实体B", "relation": "关系2", "object": "实体C"},
{"subject": "实体C", "relation": "关系3", "object": "实体A"}
],
session_id="batch-test",
turn_id=1
)
assert result["created_entities"] >= 3
assert result["created_relations"] == 3
def test_entity_mention_count(db):
"""测试实体提及次数增加"""
db.commit(
triplets=[{"subject": "热门实体", "relation": "关系", "object": "对象1"}]
)
db.commit(
triplets=[{"subject": "热门实体", "relation": "关系", "object": "对象2"}]
)
result = db.recall("热门实体")
entity = next((e for e in result["entities"] if e["name"] == "热门实体"), None)
assert entity is not None
assert entity["mention_count"] >= 2
def test_close_and_context_manager():
"""测试关闭和上下文管理器"""
with tempfile.NamedTemporaryFile(suffix='.db', delete=False) as f:
db_path = f.name
try:
with EmbeddedGraphDB(db_path) as db:
db.commit(
triplets=[{"subject": "测试", "relation": "上下文", "object": "管理器"}]
)
assert db.conn is not None
with EmbeddedGraphDB(db_path) as db:
result = db.introspect()
assert result["entity_count"] >= 1
finally:
if os.path.exists(db_path):
os.unlink(db_path)

View File

@ -0,0 +1,138 @@
"""记忆工具测试"""
import pytest
from graph_memory_tui.core.tools.memory_tools import (
MEMORY_TOOLS,
PERSONA_TOOLS,
WORKING_MEMORY_TOOLS,
TOOLS
)
def test_memory_tools_exist():
"""测试记忆工具存在"""
assert len(MEMORY_TOOLS) >= 6
def test_persona_tools_exist():
"""测试人设工具存在"""
assert len(PERSONA_TOOLS) >= 2
def test_working_memory_tools_exist():
"""测试工作记忆工具存在"""
assert len(WORKING_MEMORY_TOOLS) >= 4
def test_all_tools_combined():
"""测试工具合并"""
assert len(TOOLS) == len(MEMORY_TOOLS) + len(PERSONA_TOOLS) + len(WORKING_MEMORY_TOOLS)
def test_memory_recall_tool():
"""测试 memory_recall 工具定义"""
recall = next((t for t in TOOLS if t['function']['name'] == 'memory_recall'), None)
assert recall is not None
params = recall['function']['parameters']['properties']
assert 'query_intent' in params
assert 'seed_entities' in params
assert 'depth' in params
def test_memory_commit_tool():
"""测试 memory_commit 工具定义"""
commit = next((t for t in TOOLS if t['function']['name'] == 'memory_commit'), None)
assert commit is not None
params = commit['function']['parameters']['properties']
assert 'triplets' in params
def test_memory_purge_tool():
"""测试 memory_purge 工具定义"""
purge = next((t for t in TOOLS if t['function']['name'] == 'memory_purge'), None)
assert purge is not None
params = purge['function']['parameters']['properties']
assert 'criteria' in params
assert 'mode' in params
def test_memory_introspect_tool():
"""测试 memory_introspect 工具定义"""
introspect = next((t for t in TOOLS if t['function']['name'] == 'memory_introspect'), None)
assert introspect is not None
def test_persona_update_tool():
"""测试 persona_update 工具定义"""
update = next((t for t in TOOLS if t['function']['name'] == 'persona_update'), None)
assert update is not None
params = update['function']['parameters']['properties']
assert 'attributes' in params
def test_persona_clear_tool():
"""测试 persona_clear 工具定义"""
clear = next((t for t in TOOLS if t['function']['name'] == 'persona_clear'), None)
assert clear is not None
def test_task_create_tool():
"""测试 task_create 工具定义"""
create = next((t for t in TOOLS if t['function']['name'] == 'task_create'), None)
assert create is not None
params = create['function']['parameters']['properties']
assert 'task_id' in params
assert 'description' in params
def test_task_set_state_tool():
"""测试 task_set_state 工具定义"""
set_state = next((t for t in TOOLS if t['function']['name'] == 'task_set_state'), None)
assert set_state is not None
params = set_state['function']['parameters']['properties']
assert 'task_id' in params
assert 'state' in params
def test_task_delete_tool():
"""测试 task_delete 工具定义"""
delete = next((t for t in TOOLS if t['function']['name'] == 'task_delete'), None)
assert delete is not None
def test_task_link_info_tool():
"""测试 task_link_info 工具定义"""
link = next((t for t in TOOLS if t['function']['name'] == 'task_link_info'), None)
assert link is not None
params = link['function']['parameters']['properties']
assert 'task_id' in params
assert 'info_node_names' in params
def test_tool_has_required_fields():
"""测试工具都有必需字段"""
for tool in TOOLS:
assert 'type' in tool
assert tool['type'] == 'function'
assert 'function' in tool
assert 'name' in tool['function']
assert 'description' in tool['function']
assert 'parameters' in tool['function']
def test_tool_state_enum():
"""测试 task_set_state 的状态枚举"""
set_state = next((t for t in TOOLS if t['function']['name'] == 'task_set_state'), None)
state_enum = set_state['function']['parameters']['properties']['state']['enum']
assert '进行中' in state_enum
assert '已完成' in state_enum
assert '已暂停' in state_enum
assert '已取消' in state_enum

View File

@ -0,0 +1,148 @@
"""工具限制器测试"""
import pytest
from graph_memory_tui.core.tools.tool_limiter import (
ToolLimiter,
ToolLimits,
ToolCallCount
)
@pytest.fixture
def limiter():
"""创建限制器实例"""
return ToolLimiter()
def test_classify_persona_tools(limiter):
"""测试人设工具分类"""
category, operation = limiter._classify_tool('persona_update', {})
assert category == 'persona'
assert operation == 'update'
category, operation = limiter._classify_tool('persona_clear', {})
assert category == 'persona'
assert operation == 'update'
def test_classify_task_tools(limiter):
"""测试任务工具分类"""
category, operation = limiter._classify_tool('task_create', {})
assert category == 'task'
assert operation == 'update'
category, operation = limiter._classify_tool('task_set_state', {})
assert category == 'task'
assert operation == 'update'
category, operation = limiter._classify_tool('task_delete', {})
assert category == 'task'
assert operation == 'update'
category, operation = limiter._classify_tool('task_link_info', {})
assert category == 'task'
assert operation == 'update'
def test_classify_memory_recall(limiter):
"""测试 memory_recall 分类"""
category, operation = limiter._classify_tool('memory_recall', {'query_intent': 'Python'})
assert category == 'memory'
assert operation == 'query'
def test_classify_memory_recall_persona_query(limiter):
"""测试 memory_recall 查询人设图"""
category, operation = limiter._classify_tool(
'memory_recall',
{'query_intent': 'AI,人设,角色'}
)
assert category == 'persona'
assert operation == 'query'
def test_classify_memory_recall_task_query(limiter):
"""测试 memory_recall 查询工作记忆链"""
category, operation = limiter._classify_tool(
'memory_recall',
{'query_intent': 'TaskNode,工作记忆'}
)
assert category == 'task'
assert operation == 'query'
def test_can_call_allowed(limiter):
"""测试允许调用"""
allowed, reason = limiter.can_call('memory_recall', {'query_intent': 'test'})
assert allowed is True
def test_can_call_limit_reached(limiter):
"""测试达到限制"""
for _ in range(20):
limiter.record_call('memory_recall', {'query_intent': 'test'})
allowed, reason = limiter.can_call('memory_recall', {'query_intent': 'test'})
assert allowed is False
assert '上限' in reason
def test_record_call(limiter):
"""测试记录调用"""
initial_count = limiter.counts.memory_query
limiter.record_call('memory_recall', {'query_intent': 'test'})
assert limiter.counts.memory_query == initial_count + 1
def test_reset(limiter):
"""测试重置计数"""
limiter.record_call('memory_recall', {'query_intent': 'test'})
limiter.record_call('memory_recall', {'query_intent': 'test'})
limiter.reset()
assert limiter.counts.memory_query == 0
assert limiter.counts.memory_update == 0
def test_get_summary(limiter):
"""测试获取统计摘要"""
limiter.record_call('memory_recall', {'query_intent': 'test'})
summary = limiter.get_summary()
assert isinstance(summary, str)
assert '一般记忆' in summary
assert '查询1' in summary
def test_custom_limits():
"""测试自定义限制"""
limits = ToolLimits(
memory_query_max=5,
memory_update_max=3
)
limiter = ToolLimiter(limits)
assert limiter.limits.memory_query_max == 5
assert limiter.limits.memory_update_max == 3
def test_persona_query_limit(limiter):
"""测试人设图查询限制"""
for _ in range(1):
limiter.record_call('memory_recall', {'query_intent': '人设'})
allowed, _ = limiter.can_call('memory_recall', {'query_intent': '人设'})
assert allowed is False
def test_task_query_limit(limiter):
"""测试工作记忆链查询限制"""
for _ in range(4):
limiter.record_call('memory_recall', {'query_intent': 'TaskNode'})
allowed, _ = limiter.can_call('memory_recall', {'query_intent': 'TaskNode'})
assert allowed is False