import asyncio import json import unittest from unittest.mock import AsyncMock, MagicMock, patch from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool from app.agent.tools.impl.update_system_settings import UpdateSystemSettingsTool from app.agent.tools.manager import MoviePilotToolsManager from app.core.config import settings from app.schemas.types import SystemConfigKey class TestAgentSystemSettingsTools(unittest.TestCase): def test_query_system_settings_returns_exact_systemconfig_value(self): tool = QuerySystemSettingsTool(session_id="session-1", user_id="10001") with patch( "app.agent.tools.impl.query_system_settings.SystemConfigOper" ) as system_config_oper: system_config_oper.return_value.get.return_value = [{"name": "qb", "enabled": True}] result = asyncio.run(tool.run(setting_key="Downloaders")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["matched_count"], 1) self.assertEqual(payload["settings"][0]["setting_key"], "Downloaders") self.assertEqual(payload["settings"][0]["value"][0]["name"], "qb") system_config_oper.return_value.get.assert_called_once_with( SystemConfigKey.Downloaders ) def test_query_system_settings_redacts_secret_values_by_default(self): """查询系统设置默认应脱敏 API Key、Token、Cookie 等敏感字段。""" tool = QuerySystemSettingsTool(session_id="session-1", user_id="10001") with patch( "app.agent.tools.impl.query_system_settings.SystemConfigOper" ) as system_config_oper: system_config_oper.return_value.get.return_value = [ { "name": "site-a", "apikey": "site-api-key", "token": "site-token", "cookie": "uid=1; passkey=secret", "url": "https://example.com", } ] result = asyncio.run( tool.run(setting_key="UserSiteAuthParams", include_values=True) ) payload = json.loads(result) item = payload["settings"][0] self.assertTrue(item["redacted"]) self.assertFalse(payload["show_secrets"]) self.assertEqual("***", item["value"][0]["apikey"]) self.assertEqual("***", item["value"][0]["token"]) self.assertEqual("***", item["value"][0]["cookie"]) self.assertEqual("https://example.com", item["value"][0]["url"]) def test_query_system_settings_show_secrets_requires_admin_context(self): """只有管理员显式请求 show_secrets 时才返回敏感配置原值。""" tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin") tool.set_agent_context({"is_admin": True}) with patch( "app.agent.tools.impl.query_system_settings.SystemConfigOper" ) as system_config_oper: system_config_oper.return_value.get.return_value = [ {"name": "site-a", "apikey": "site-api-key"} ] result = asyncio.run( tool.run( setting_key="UserSiteAuthParams", include_values=True, show_secrets=True, ) ) payload = json.loads(result) item = payload["settings"][0] self.assertTrue(payload["show_secrets"]) self.assertFalse(item["redacted"]) self.assertEqual("site-api-key", item["value"][0]["apikey"]) def test_query_system_settings_group_defaults_to_summary_for_multiple_items(self): tool = QuerySystemSettingsTool(session_id="session-1", user_id="10001") with patch( "app.agent.tools.impl.query_system_settings.SystemConfigOper" ) as system_config_oper: system_config_oper.return_value.get.return_value = [] result = asyncio.run(tool.run(group="systemconfig")) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertFalse(payload["include_values"]) self.assertGreater(payload["matched_count"], 1) def test_update_system_settings_merges_dict_and_emits_event(self): tool = UpdateSystemSettingsTool(session_id="session-1", user_id="10001") config_oper = MagicMock() config_oper.get.side_effect = [ {"chatgpt": {"enabled": True}}, {"chatgpt": {"enabled": False}, "gemini": {"enabled": True}}, ] config_oper.async_set = AsyncMock(return_value=True) with patch( "app.agent.tools.impl.update_system_settings.SystemConfigOper", return_value=config_oper, ), patch( "app.agent.tools.impl.update_system_settings.eventmanager.async_send_event", new=AsyncMock(), ) as send_event: result = asyncio.run( tool.run( setting_key="AIAgentConfig", operation="merge_dict", value={"chatgpt": {"enabled": False}, "gemini": {"enabled": True}}, ) ) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertTrue(payload["changed"]) config_oper.async_set.assert_awaited_once_with( SystemConfigKey.AIAgentConfig, {"chatgpt": {"enabled": False}, "gemini": {"enabled": True}}, ) send_event.assert_awaited_once() def test_update_system_settings_upserts_named_list_item(self): tool = UpdateSystemSettingsTool(session_id="session-1", user_id="10001") config_oper = MagicMock() config_oper.get.side_effect = [ [{"name": "qb", "enabled": False}], [{"name": "qb", "enabled": True}], ] config_oper.async_set = AsyncMock(return_value=True) with patch( "app.agent.tools.impl.update_system_settings.SystemConfigOper", return_value=config_oper, ), patch( "app.agent.tools.impl.update_system_settings.eventmanager.async_send_event", new=AsyncMock(), ): result = asyncio.run( tool.run( setting_key="downloaders", operation="upsert_list_item", value={"name": "qb", "enabled": True}, ) ) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertEqual(payload["saved_value"], [{"name": "qb", "enabled": True}]) config_oper.async_set.assert_awaited_once_with( SystemConfigKey.Downloaders, [{"name": "qb", "enabled": True}], ) def test_update_system_settings_redacts_secret_values_in_response(self): """更新敏感系统设置后响应不应回显旧值和新值中的密钥。""" tool = UpdateSystemSettingsTool(session_id="session-1", user_id="10001") config_oper = MagicMock() config_oper.get.side_effect = [ [{"name": "site-a", "apikey": "old-key"}], [{"name": "site-a", "apikey": "new-key"}], ] config_oper.async_set = AsyncMock(return_value=True) with patch( "app.agent.tools.impl.update_system_settings.SystemConfigOper", return_value=config_oper, ), patch( "app.agent.tools.impl.update_system_settings.eventmanager.async_send_event", new=AsyncMock(), ): result = asyncio.run( tool.run( setting_key="UserSiteAuthParams", operation="upsert_list_item", value={"name": "site-a", "apikey": "new-key"}, match_field="name", ) ) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertTrue(payload["values_redacted"]) self.assertEqual("***", payload["previous_value"][0]["apikey"]) self.assertEqual("***", payload["saved_value"][0]["apikey"]) def test_update_system_settings_updates_basic_settings(self): tool = UpdateSystemSettingsTool(session_id="session-1", user_id="10001") # settings 是 pydantic 模型实例，不能直接 patch 其实例方法（__setattr__ 会拦截）， # 改 patch 类上的方法；经实例调用时不带 self，断言参数不受影响。 with patch.object( type(settings), "update_setting", return_value=(True, ""), ) as update_setting, patch.object( UpdateSystemSettingsTool, "_load_setting_value", side_effect=["https://old.example.com", "https://new.example.com"], ), patch( "app.agent.tools.impl.update_system_settings.eventmanager.async_send_event", new=AsyncMock(), ) as send_event: result = asyncio.run( tool.run(setting_key="APP_DOMAIN", value="https://new.example.com") ) payload = json.loads(result) self.assertTrue(payload["success"]) self.assertTrue(payload["changed"]) update_setting.assert_called_once_with("APP_DOMAIN", "https://new.example.com") send_event.assert_awaited_once() def test_tool_manager_blocks_admin_tools_for_non_admin_context(self): tool = QuerySystemSettingsTool(session_id="session-1", user_id="10001") with patch( "app.agent.tools.manager.MoviePilotToolFactory.create_tools", return_value=[tool], ): manager = MoviePilotToolsManager(is_admin=False) result = asyncio.run( manager.call_tool( "query_system_settings", {"setting_key": "Downloaders"}, ) ) payload = json.loads(result) self.assertIn("error", payload) self.assertIn("系统管理员", payload["error"])