"""Gateway writer coverage for the shared ``mcpLifecycle`` generation.""" from __future__ import annotations import json import pytest # Reuse the process-local isolation fixture so these writer tests do not leak # cache globals or the session-pool singleton. from test_mcp_cache_reconciliation import reconciler # noqa: F401 from app.gateway.routers import mcp as mcp_router from app.gateway.routers.mcp import ( McpConfigUpdateRequest, McpServerConfigResponse, McpServerConfigUpdateRequest, McpServerStateUpdateRequest, ) from deerflow.config.extensions_config import ExtensionsConfig from deerflow.mcp.lifecycle import _stdio_fingerprints, legacy_token, parse_mcp_lifecycle def _stdio(command: str) -> dict: return {"enabled": True, "type": "stdio", "command": command} def _write(cfg, servers: dict) -> None: cfg.write_text(json.dumps({"mcpServers": servers, "skills": {}}), encoding="utf-8") def _ledger(cfg): return parse_mcp_lifecycle(json.loads(cfg.read_text(encoding="utf-8"))) def _legacy(cfg, name: str) -> str: fingerprint = _stdio_fingerprints(ExtensionsConfig.from_file(str(cfg)))[name] return legacy_token(name, fingerprint) def test_state_disable_and_reenable_advance_the_generation( reconciler, # noqa: F811 - fixture imported from the sibling suite monkeypatch, tmp_path, ): """Raw in-place mutation must not hide the enabled -> disabled transition.""" cfg = tmp_path / "extensions_config.json" _write(cfg, {"A": _stdio("npx")}) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) baseline = _legacy(cfg, "A") mcp_router._apply_mcp_server_state_update(McpServerStateUpdateRequest(server_name="A", enabled=False)) disabled = _ledger(cfg).servers["A"] assert disabled != baseline mcp_router._apply_mcp_server_state_update(McpServerStateUpdateRequest(server_name="A", enabled=True)) reenabled = _ledger(cfg).servers["A"] assert reenabled not in {baseline, disabled} def test_delete_then_identical_readd_advance_the_generation( reconciler, # noqa: F811 - fixture imported from the sibling suite monkeypatch, tmp_path, ): """Delete tombstones the token and an identical re-add must rotate it again.""" cfg = tmp_path / "extensions_config.json" _write(cfg, {"A": _stdio("npx")}) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) mcp_router._apply_mcp_server_delete("A") tombstone = _ledger(cfg).servers["A"] mcp_router._apply_mcp_servers_create( McpConfigUpdateRequest( mcp_servers={"A": McpServerConfigResponse(enabled=True, type="stdio", command="npx")}, ) ) rebound = _ledger(cfg).servers["A"] assert rebound != tombstone def test_commit_helper_requires_previous_raw(): """A controlled write must not be able to skip the lifecycle commit.""" import inspect parameters = inspect.signature(mcp_router._commit_mcp_config_write).parameters assert parameters["previous_raw"].default is inspect.Parameter.empty def test_bulk_update_rotates_only_the_changed_server_token( reconciler, # noqa: F811 - fixture imported from the sibling suite monkeypatch, tmp_path, ): """A connection change rotates only that server's token; $VAR stays raw.""" cfg = tmp_path / "extensions_config.json" cfg.write_text( json.dumps( { "mcpServers": { "A": {"enabled": True, "type": "stdio", "command": "npx", "args": ["pkg-v1"]}, "B": {"enabled": True, "type": "stdio", "command": "npx", "args": ["pkg-b"], "env": {"TOKEN": "$PE_VAR"}}, }, "skills": {}, } ), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) monkeypatch.setenv("PE_VAR", "secret-value") baseline_a = _legacy(cfg, "A") mcp_router._apply_mcp_config_update( McpConfigUpdateRequest( mcp_servers={ "A": McpServerConfigResponse(enabled=True, type="stdio", command="npx", args=["pkg-v2"]), "B": McpServerConfigResponse(enabled=True, type="stdio", command="npx", args=["pkg-b"], env={"TOKEN": "$PE_VAR"}), } ) ) ledger = _ledger(cfg) assert ledger is not None assert ledger.servers["A"] != baseline_a # the connection changed assert "B" not in ledger.servers # unchanged legacy server stays derived written = cfg.read_text(encoding="utf-8") assert "$PE_VAR" in written assert "secret-value" not in written @pytest.mark.asyncio async def test_targeted_metadata_update_keeps_token_and_session( reconciler, # noqa: F811 - fixture imported from the sibling suite monkeypatch, tmp_path, ): """A metadata-only targeted update must not rotate the token or the session.""" import test_mcp_cache_reconciliation as harness from deerflow.mcp import session_pool as session_pool_module from deerflow.mcp.session_pool import MCPSessionPool cfg = tmp_path / "extensions_config.json" pool = MCPSessionPool() session_pool_module._pool = pool log = harness._session_log() await harness._initialize(monkeypatch, cfg, {"A": _stdio("npx")}, log) binding_before = harness._binding(pool, "A") entry_before = harness._entry(pool, "A") mcp_router._apply_mcp_server_config_update( McpServerConfigUpdateRequest( server_name="A", server=McpServerConfigResponse(enabled=True, type="stdio", command="npx", description="renamed"), ) ) ledger = _ledger(cfg) assert ledger is not None and "A" not in ledger.servers # metadata-only: no token assert harness._binding(pool, "A") is binding_before assert harness._entry(pool, "A") is entry_before assert json.loads(cfg.read_text(encoding="utf-8"))["mcpServers"]["A"]["description"] == "renamed" def test_skills_writer_preserves_the_lifecycle_ledger( reconciler, # noqa: F811 - fixture imported from the sibling suite monkeypatch, tmp_path, ): """A skills-only write must round-trip mcpLifecycle exactly (no recompute).""" from unittest.mock import MagicMock from app.gateway.routers import skills as skills_router cfg = tmp_path / "extensions_config.json" ledger = {"version": 1, "servers": {"A": "a" * 32, "B": "b" * 32}} cfg.write_text( json.dumps( { "mcpServers": {"A": {"enabled": True, "type": "stdio", "command": "npx"}}, "skills": {"demo": {"enabled": True}}, "mcpLifecycle": ledger, } ), encoding="utf-8", ) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) monkeypatch.setattr("app.gateway.routers.skills.reload_extensions_config", lambda *a, **k: None) skills_router._write_extensions_skill_state(MagicMock(_app_config=None), "demo", False, rebuild_public_projection=False) written = json.loads(cfg.read_text(encoding="utf-8")) assert written["mcpLifecycle"] == ledger # preserved byte-for-value, never recomputed assert written["skills"]["demo"]["enabled"] is False def test_removed_servers_drop_their_stored_oauth_tokens_and_disabled_ones_keep_them( reconciler, # noqa: F811 - fixture imported from the sibling suite monkeypatch, tmp_path, ): """A removed server's mcp_oauth_tokens row can never be read again; a disabled one may be re-enabled.""" from unittest.mock import MagicMock from deerflow.mcp.oauth_store import set_mcp_oauth_token_store cfg = tmp_path / "extensions_config.json" _write(cfg, {"A": _stdio("npx"), "B": _stdio("uvx")}) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) store = MagicMock() store.forget_unconfigured.return_value = 0 previous = set_mcp_oauth_token_store(store) try: mcp_router._apply_mcp_server_state_update(McpServerStateUpdateRequest(server_name="A", enabled=False)) mcp_router._apply_mcp_server_delete("B") finally: set_mcp_oauth_token_store(previous) calls = [(call.args[0], set(call.args[1])) for call in store.forget_unconfigured.call_args_list] assert calls == [(None, {"A", "B"}), (None, {"A"})]