"""ACEManager 批量操作与告警相关逻辑的单测，用于提升 ace_manager 覆盖率."""
import pytest
from unittest.mock import MagicMock, patch
from src.ace_manager import ACEManager
from src.config_manager import ConfigManager
class TestACEManagerBulkAndAlerts:
"""验证 bulk_update_strategies / bulk_export_strategies / get_strategy_alerts 逻辑."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建独立配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
def _create_manager(self, config_manager, graphiti_client):
"""创建禁用 ACE 初始化的 ACEManager 以便注入依赖."""
with patch.object(ACEManager, "_initialize_ace", lambda self: None):
manager = ACEManager(config_manager, graphiti_client)
manager.enabled = True
return manager
def _mock_session(self, graphiti_client):
"""为 graphiti_client.driver.session 配置 MagicMock 上下文管理器."""
session = MagicMock()
graphiti_client.driver.session.return_value.__enter__.return_value = session
return session
def test_get_strategy_alerts_success(self, config_manager):
"""在有 StrategyTrend 数据时，应按失败次数返回告警列表."""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
# 模拟聚合后的告警记录
session.run.return_value = [
{
"tool_name": "search_entities",
"arguments_hash": "hash1",
"failures": 12.0,
"usage": 20.0,
"last_date": "2025-12-01",
},
{
"tool_name": "add_episode",
"arguments_hash": "hash2",
"failures": 4.0,
"usage": 10.0,
"last_date": "2025-12-01",
},
]
manager = self._create_manager(config_manager, graphiti_client)
alerts = manager.get_strategy_alerts(failure_threshold=3, days=7)
assert len(alerts) == 2
# 第一条失败次数较多，应被标记为 warning 或更高
assert alerts[0]["tool_name"] == "search_entities"
assert alerts[0]["failures"] == 12.0
assert alerts[0]["severity"] in {"warning", "critical"}
# 第二条失败次数较少，但仍 >= threshold
assert alerts[1]["tool_name"] == "add_episode"
assert alerts[1]["failures"] == 4.0
def test_get_strategy_alerts_not_connected(self, config_manager):
"""未连接 Neo4j 时应直接返回空列表."""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = False
manager = self._create_manager(config_manager, graphiti_client)
alerts = manager.get_strategy_alerts()
assert alerts == []
def test_get_strategy_alerts_exception(self, config_manager):
"""查询告警时出现异常应返回空列表。"""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
session.run.side_effect = Exception("alert error")
manager = self._create_manager(config_manager, graphiti_client)
alerts = manager.get_strategy_alerts()
assert alerts == []
def test_bulk_update_strategies_enable(self, config_manager):
"""批量启用策略时，应构造查询并触发缓存刷新."""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
# single() 返回影响条数
session.run.return_value.single.return_value = {"affected": 3}
manager = self._create_manager(config_manager, graphiti_client)
with patch.object(manager, "_reload_strategies_async") as mock_reload:
result = manager.bulk_update_strategies(
action="enable",
tool_name="search_entities",
success_rate_min=0.5,
usage_min=10,
limit=50,
)
assert result is not None
assert result["action"] == "enable"
assert result["affected"] == 3
assert result["limit"] == 50
mock_reload.assert_called_once()
def test_bulk_update_strategies_invalid_action(self, config_manager):
"""无效 action 应直接返回 None，不触发查询."""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
self._mock_session(graphiti_client)
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_update_strategies(action="unknown")
assert result is None
graphiti_client.driver.session.assert_not_called()
def test_bulk_update_strategies_not_connected(self, config_manager):
"""未连接时返回 None。"""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = False
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_update_strategies(action="enable")
assert result is None
def test_bulk_update_strategies_exception(self, config_manager):
"""更新出现异常时返回 None。"""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
session.run.side_effect = Exception("bulk error")
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_update_strategies(action="enable")
assert result is None
def test_bulk_export_strategies_in_memory(self, config_manager, tmp_path):
"""不带 file_path 时，应返回内存中的策略列表."""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
# 模拟返回两个策略节点
session.run.return_value = [
{
"s": {
"tool_name": "search_entities",
"arguments_hash": "hash1",
"content": "strategy 1",
"success_rate": 0.8,
"usage_count": 10,
}
},
{
"s": {
"tool_name": "add_episode",
"arguments_hash": "hash2",
"content": "strategy 2",
"success_rate": 0.6,
"usage_count": 5,
}
},
]
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_export_strategies(tool_name=None, limit=10, file_path=None)
assert result is not None
assert result["count"] == 2
assert result["file_path"] is None
assert len(result["strategies"]) == 2
names = {s["tool_name"] for s in result["strategies"]}
assert names == {"search_entities", "add_episode"}
def test_bulk_export_strategies_to_file(self, config_manager, tmp_path):
"""带 file_path 时，应写入文件并且返回空 strategies 列表."""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
session.run.return_value = [
{
"s": {
"tool_name": "search_entities",
"arguments_hash": "hash1",
"content": "strategy 1",
"success_rate": 0.8,
"usage_count": 10,
}
}
]
file_path = tmp_path / "export.json"
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_export_strategies(
tool_name="search_entities",
limit=5,
file_path=str(file_path),
)
assert result is not None
assert result["count"] == 1
assert result["file_path"] == str(file_path)
assert result["strategies"] == []
assert file_path.exists()
def test_bulk_export_strategies_not_connected(self, config_manager):
"""未连接时返回 None。"""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = False
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_export_strategies()
assert result is None
def test_bulk_export_strategies_exception(self, config_manager):
"""导出时出现异常应返回 None。"""
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = self._mock_session(graphiti_client)
session.run.side_effect = Exception("export error")
manager = self._create_manager(config_manager, graphiti_client)
result = manager.bulk_export_strategies()
assert result is None
