import asyncio import json import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from app.agent.tools.impl.install_plugin import InstallPluginTool from app.agent.tools.impl.query_installed_plugins import QueryInstalledPluginsTool from app.agent.tools.impl.query_market_plugins import QueryMarketPluginsTool from app.agent.tools.impl.query_plugin_config import QueryPluginConfigTool from app.agent.tools.impl.query_plugin_data import QueryPluginDataTool from app.agent.tools.impl.reload_plugin import ReloadPluginTool from app.agent.tools.impl.uninstall_plugin import UninstallPluginTool from app.agent.tools.impl.update_plugin_config import UpdatePluginConfigTool class TestAgentPluginTools(unittest.TestCase): @staticmethod def _plugin_snapshot(state: bool = True) -> dict: return { "plugin_id": "DemoPlugin", "plugin_name": "Demo Plugin", "plugin_version": "1.0.0", "state": state, } @staticmethod def _market_plugin(plugin_id: str, plugin_name: str, installed: bool = False): return SimpleNamespace( id=plugin_id, plugin_name=plugin_name, plugin_desc=f"{plugin_name} description", plugin_version="1.0.0", plugin_author="author", installed=installed, has_update=False, state=installed, repo_url="https://example.com/market", add_time=1, ) def test_query_market_plugins_filters_candidates(self): tool = QueryMarketPluginsTool(session_id="session-1", user_id="10001") plugins = [ self._market_plugin("DemoPlugin", "Demo Plugin"), self._market_plugin("OtherPlugin", "Other Plugin"), ] with patch( "app.agent.tools.impl.query_market_plugins.load_market_plugins", new=AsyncMock(return_value=plugins), ): result = asyncio.run(tool.run(query="demo")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["match_count"], 1) self.assertEqual(payload["plugins"][0]["id"], "DemoPlugin") def test_query_installed_plugins_filters_candidates(self): tool = QueryInstalledPluginsTool(session_id="session-1", user_id="10001") plugins = [ self._market_plugin("DemoPlugin", "Demo Plugin", installed=True), self._market_plugin("OtherPlugin", "Other Plugin", installed=True), ] with patch( "app.agent.tools.impl.query_installed_plugins.list_installed_plugins", return_value=plugins, ): result = asyncio.run(tool.run(query="demo")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["match_count"], 1) self.assertEqual(payload["plugins"][0]["id"], "DemoPlugin") def test_query_plugin_config_returns_saved_config_and_default_model(self): tool = QueryPluginConfigTool(session_id="session-1", user_id="10001") plugin_manager = MagicMock() plugin_manager.get_plugin_config.return_value = {"enabled": True} plugin_instance = MagicMock() plugin_instance.get_form.return_value = (None, {"enabled": False, "interval": 10}) plugin_manager.running_plugins = {"DemoPlugin": plugin_instance} with patch( "app.agent.tools.impl.query_plugin_config.get_plugin_snapshot", return_value=self._plugin_snapshot(), ), patch( "app.agent.tools.impl.query_plugin_config.PluginManager", return_value=plugin_manager, ): result = asyncio.run(tool.run(plugin_id="DemoPlugin")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["config"], {"enabled": True}) self.assertEqual(payload["default_model"], {"enabled": False, "interval": 10}) def test_update_plugin_config_merges_and_removes_keys_without_reloading(self): tool = UpdatePluginConfigTool(session_id="session-1", user_id="10001") plugin_manager = MagicMock() plugin_manager.get_plugin_config.return_value = { "enabled": False, "interval": 30, "token": "legacy-token", } plugin_manager.async_save_plugin_config = AsyncMock(return_value=True) with patch( "app.agent.tools.impl.update_plugin_config.get_plugin_snapshot", return_value=self._plugin_snapshot(), ), patch( "app.agent.tools.impl.update_plugin_config.PluginManager", return_value=plugin_manager, ): result = asyncio.run( tool.run( plugin_id="DemoPlugin", updates={"enabled": True}, remove_keys=["token"], ) ) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertTrue(payload["config_requires_reload"]) self.assertEqual(payload["saved_config"], {"enabled": True, "interval": 30}) plugin_manager.async_save_plugin_config.assert_awaited_once_with( "DemoPlugin", {"enabled": True, "interval": 30}, ) def test_reload_plugin_triggers_runtime_refresh(self): tool = ReloadPluginTool(session_id="session-1", user_id="10001") with patch( "app.agent.tools.impl.reload_plugin.get_plugin_snapshot", side_effect=[self._plugin_snapshot(), self._plugin_snapshot(state=False)], ), patch( "app.agent.tools.impl.reload_plugin.reload_plugin_runtime" ) as reload_plugin_runtime: result = asyncio.run(tool.run(plugin_id="DemoPlugin")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertFalse(payload["state"]) reload_plugin_runtime.assert_called_once_with("DemoPlugin") def test_install_plugin_installs_market_candidate(self): tool = InstallPluginTool(session_id="session-1", user_id="10001") candidate = self._market_plugin("DemoPlugin", "Demo Plugin") with patch( "app.agent.tools.impl.install_plugin.load_market_plugins", new=AsyncMock(return_value=[candidate]), ), patch( "app.agent.tools.impl.install_plugin.install_plugin_runtime", new=AsyncMock(return_value=(True, "插件安装完成", False)), ) as install_runtime, patch( "app.agent.tools.impl.install_plugin.get_plugin_snapshot", return_value=self._plugin_snapshot(), ): result = asyncio.run(tool.run(plugin_id="DemoPlugin")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["plugin"]["id"], "DemoPlugin") install_runtime.assert_awaited_once_with( "DemoPlugin", "https://example.com/market", force=False ) def test_uninstall_plugin_uninstalls_installed_candidate(self): tool = UninstallPluginTool(session_id="session-1", user_id="10001") installed_plugin = self._market_plugin( "DemoPlugin", "Demo Plugin", installed=True ) with patch( "app.agent.tools.impl.uninstall_plugin.list_installed_plugins", return_value=[installed_plugin], ), patch( "app.agent.tools.impl.uninstall_plugin.uninstall_plugin_runtime", new=AsyncMock( return_value={"was_clone": False, "clone_files_removed": False} ), ) as uninstall_runtime: result = asyncio.run(tool.run(plugin_id="DemoPlugin")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["plugin"]["id"], "DemoPlugin") uninstall_runtime.assert_awaited_once_with("DemoPlugin") def test_query_plugin_data_truncates_large_payload(self): tool = QueryPluginDataTool(session_id="session-1", user_id="10001") plugin_data_oper = MagicMock() plugin_data_oper.async_get_data_all = AsyncMock(return_value=[ SimpleNamespace(key="payload", value={"text": "x" * 5000}) ]) with patch( "app.agent.tools.impl.query_plugin_data.get_plugin_snapshot", return_value=self._plugin_snapshot(), ), patch( "app.agent.tools.impl.query_plugin_data.PluginDataOper", return_value=plugin_data_oper, ): result = asyncio.run(tool.run(plugin_id="DemoPlugin", max_chars=200)) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertTrue(payload["truncated"]) self.assertIn("data_preview", payload) self.assertNotIn("data", payload) self.assertIn("已截断", payload["data_preview"]) if __name__ == "__main__": unittest.main()