| 1 | import asyncio |
| 2 | import sys |
| 3 | from pathlib import Path |
| 4 | from types import ModuleType |
| 5 | |
| 6 | PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| 7 | if str(PROJECT_ROOT) not in sys.path: |
| 8 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 9 | |
| 10 | import helpers.settings as settings_module |
| 11 | |
| 12 | |
| 13 | def test_apply_settings_updates_mcp_from_current_settings(monkeypatch): |
| 14 | base_settings = settings_module.get_default_settings() |
| 15 | previous_mcp_servers = '{"mcpServers": {}}' |
| 16 | current_mcp_servers = '{"mcpServers": {"deepwiki": {"url": "https://mcp.deepwiki.com/mcp"}}}' |
| 17 | previous = { |
| 18 | **base_settings, |
| 19 | "mcp_servers": previous_mcp_servers, |
| 20 | "mcp_server_token": "unchanged-token", |
| 21 | } |
| 22 | current = { |
| 23 | **base_settings, |
| 24 | "mcp_servers": current_mcp_servers, |
| 25 | "mcp_server_token": "unchanged-token", |
| 26 | } |
| 27 | received_mcp_servers: list[str] = [] |
| 28 | |
| 29 | class FakeDeferredTask: |
| 30 | def start_task(self, func, *args, **kwargs): |
| 31 | asyncio.run(func(*args, **kwargs)) |
| 32 | return self |
| 33 | |
| 34 | class FakePrintStyle: |
| 35 | def __init__(self, *args, **kwargs): |
| 36 | pass |
| 37 | |
| 38 | def print(self, *args, **kwargs): |
| 39 | pass |
| 40 | |
| 41 | class FakeMCPConfig: |
| 42 | @classmethod |
| 43 | def get_instance(cls): |
| 44 | return cls() |
| 45 | |
| 46 | @classmethod |
| 47 | def update(cls, mcp_servers): |
| 48 | received_mcp_servers.append(mcp_servers) |
| 49 | |
| 50 | def model_dump_json(self): |
| 51 | return "{}" |
| 52 | |
| 53 | agent_stub = ModuleType("agent") |
| 54 | agent_stub.Agent = object |
| 55 | |
| 56 | class FakeAgentContext: |
| 57 | @staticmethod |
| 58 | def all(): |
| 59 | return [] |
| 60 | |
| 61 | agent_stub.AgentContext = FakeAgentContext |
| 62 | |
| 63 | initialize_stub = ModuleType("initialize") |
| 64 | initialize_stub.initialize_agent = lambda override_settings=None: None |
| 65 | |
| 66 | mcp_handler_stub = ModuleType("helpers.mcp_handler") |
| 67 | mcp_handler_stub.MCPConfig = FakeMCPConfig |
| 68 | |
| 69 | monkeypatch.setitem(sys.modules, "agent", agent_stub) |
| 70 | monkeypatch.setitem(sys.modules, "initialize", initialize_stub) |
| 71 | monkeypatch.setitem(sys.modules, "helpers.mcp_handler", mcp_handler_stub) |
| 72 | monkeypatch.setattr(settings_module, "_settings", current) |
| 73 | monkeypatch.setattr(settings_module, "_apply_timezone_setting", lambda *args, **kwargs: None) |
| 74 | monkeypatch.setattr(settings_module.defer, "DeferredTask", FakeDeferredTask) |
| 75 | monkeypatch.setattr(settings_module, "PrintStyle", FakePrintStyle) |
| 76 | monkeypatch.setattr(settings_module.NotificationManager, "send_notification", lambda **kwargs: None) |
| 77 | monkeypatch.setattr(settings_module, "create_auth_token", lambda: "unchanged-token") |
| 78 | |
| 79 | settings_module._apply_settings(previous) |
| 80 | |
| 81 | assert received_mcp_servers == [current_mcp_servers] |