1- """Unit tests for channel_mcp.configure_channel_mcp — Linear/Jira MCP gating + merge."""
1+ """Unit tests for channel_mcp.configure_channel_mcp — Jira MCP gating + merge.
2+
3+ Linear is NOT tested here: ABCA runs Linear 100% deterministically (ADR-016),
4+ so there is no Linear MCP entry. The gate below asserts that channel_source=='linear'
5+ is now a no-op (no .mcp.json written).
6+ """
27
38from __future__ import annotations
49
914 JIRA_API_TOKEN_ENV ,
1015 JIRA_MCP_SERVER_KEY ,
1116 JIRA_MCP_URL ,
12- LINEAR_API_TOKEN_ENV ,
13- LINEAR_MCP_SERVER_KEY ,
14- LINEAR_MCP_URL ,
1517 configure_channel_mcp ,
1618)
1719
@@ -23,7 +25,23 @@ def _read_mcp(repo_dir: str) -> dict:
2325
2426
2527class TestChannelGate :
26- """Only channel_source=='linear' writes anything — everything else is a no-op."""
28+ """Only channel_source with a wired MCP writes anything — everything else is a no-op."""
29+
30+ def test_no_op_for_linear_channel (self , tmp_path ):
31+ # ADR-016: Linear is fully deterministic — no Linear MCP is written.
32+ wrote = configure_channel_mcp (str (tmp_path ), "linear" )
33+ assert wrote is False
34+ assert not (tmp_path / ".mcp.json" ).exists ()
35+
36+ def test_no_op_for_linear_channel_ignores_gateway_url (self , tmp_path ):
37+ # A stale gateway_url in metadata must not resurrect a Linear MCP entry.
38+ wrote = configure_channel_mcp (
39+ str (tmp_path ),
40+ "linear" ,
41+ {"gateway_url" : "https://gw.example/mcp" },
42+ )
43+ assert wrote is False
44+ assert not (tmp_path / ".mcp.json" ).exists ()
2745
2846 def test_no_op_for_slack_channel (self , tmp_path ):
2947 wrote = configure_channel_mcp (str (tmp_path ), "slack" )
@@ -46,101 +64,16 @@ def test_no_op_for_empty_channel(self, tmp_path):
4664 assert not (tmp_path / ".mcp.json" ).exists ()
4765
4866
49- class TestLinearWrite :
50- """channel_source=='linear' writes .mcp.json with the linear-server entry."""
51-
52- def test_creates_mcp_json_with_linear_server_key (self , tmp_path ):
53- wrote = configure_channel_mcp (str (tmp_path ), "linear" )
54- assert wrote is True
55- config = _read_mcp (str (tmp_path ))
56- assert LINEAR_MCP_SERVER_KEY in config ["mcpServers" ]
57-
58- def test_renders_linear_url_and_token_placeholder (self , tmp_path ):
59- configure_channel_mcp (str (tmp_path ), "linear" )
60- entry = _read_mcp (str (tmp_path ))["mcpServers" ][LINEAR_MCP_SERVER_KEY ]
61- assert entry ["type" ] == "http"
62- assert entry ["url" ] == LINEAR_MCP_URL
63- assert entry ["headers" ]["Authorization" ] == f"Bearer ${{{ LINEAR_API_TOKEN_ENV }
64-
65- def test_server_key_is_linear_server (self ):
66- # If this ever changes, tools surface under a different mcp__ prefix and
67- # the agent prompt (prompt_builder._channel_prompt_addendum) must be
68- # updated in lockstep.
69- assert LINEAR_MCP_SERVER_KEY == "linear-server"
70-
71-
72- class TestMerge :
73- """Existing .mcp.json must not be clobbered."""
74-
75- def test_adds_linear_to_existing_empty_mcp_json (self , tmp_path ):
76- (tmp_path / ".mcp.json" ).write_text ("{}" )
77- wrote = configure_channel_mcp (str (tmp_path ), "linear" )
78- assert wrote is True
79- assert LINEAR_MCP_SERVER_KEY in _read_mcp (str (tmp_path ))["mcpServers" ]
80-
81- def test_preserves_existing_mcp_servers (self , tmp_path ):
82- existing = {
83- "mcpServers" : {
84- "other-server" : {"type" : "stdio" , "command" : "/usr/bin/my-mcp" },
85- },
86- }
87- (tmp_path / ".mcp.json" ).write_text (json .dumps (existing ))
88-
89- configure_channel_mcp (str (tmp_path ), "linear" )
90- merged = _read_mcp (str (tmp_path ))
91- assert "other-server" in merged ["mcpServers" ]
92- assert merged ["mcpServers" ]["other-server" ]["command" ] == "/usr/bin/my-mcp"
93- assert LINEAR_MCP_SERVER_KEY in merged ["mcpServers" ]
94-
95- def test_overwrites_existing_linear_server_entry (self , tmp_path ):
96- # If someone committed a stale Linear entry with a wrong token var, we
97- # want the fresh ABCA-written entry to win — otherwise the MCP would
98- # fail to auth.
99- existing = {
100- "mcpServers" : {
101- LINEAR_MCP_SERVER_KEY : {
102- "type" : "http" ,
103- "url" : "https://stale.example" ,
104- "headers" : {"Authorization" : "Bearer stale" },
105- },
106- },
107- }
108- (tmp_path / ".mcp.json" ).write_text (json .dumps (existing ))
109-
110- configure_channel_mcp (str (tmp_path ), "linear" )
111- entry = _read_mcp (str (tmp_path ))["mcpServers" ][LINEAR_MCP_SERVER_KEY ]
112- assert entry ["url" ] == LINEAR_MCP_URL
113- assert "stale" not in entry ["headers" ]["Authorization" ]
114-
115- def test_tolerates_mcp_json_without_mcpservers_key (self , tmp_path ):
116- # A .mcp.json that only has unrelated top-level keys should still
117- # gain an mcpServers map.
118- (tmp_path / ".mcp.json" ).write_text (json .dumps ({"version" : 1 }))
119- configure_channel_mcp (str (tmp_path ), "linear" )
120- merged = _read_mcp (str (tmp_path ))
121- assert merged ["version" ] == 1
122- assert LINEAR_MCP_SERVER_KEY in merged ["mcpServers" ]
123-
124- def test_malformed_mcp_json_is_replaced (self , tmp_path ):
125- # Malformed JSON is treated as absent (logged as a warning in shell.log)
126- # rather than crashing the pipeline.
127- (tmp_path / ".mcp.json" ).write_text ("{not json" )
128- wrote = configure_channel_mcp (str (tmp_path ), "linear" )
129- assert wrote is True
130- merged = _read_mcp (str (tmp_path ))
131- assert LINEAR_MCP_SERVER_KEY in merged ["mcpServers" ]
132-
133-
13467class TestRepoDirGuard :
13568 """Missing repo_dir must not raise — the pipeline should keep going."""
13669
13770 def test_missing_repo_dir (self , tmp_path ):
13871 missing = tmp_path / "does-not-exist"
139- wrote = configure_channel_mcp (str (missing ), "linear " )
72+ wrote = configure_channel_mcp (str (missing ), "jira " )
14073 assert wrote is False
14174
14275 def test_empty_repo_dir_string (self ):
143- wrote = configure_channel_mcp ("" , "linear " )
76+ wrote = configure_channel_mcp ("" , "jira " )
14477 assert wrote is False
14578
14679
@@ -169,6 +102,12 @@ def test_server_key_is_jira_server(self):
169102class TestJiraMerge :
170103 """Jira entry must coexist with other servers and overwrite stale jira entries."""
171104
105+ def test_adds_jira_to_existing_empty_mcp_json (self , tmp_path ):
106+ (tmp_path / ".mcp.json" ).write_text ("{}" )
107+ wrote = configure_channel_mcp (str (tmp_path ), "jira" )
108+ assert wrote is True
109+ assert JIRA_MCP_SERVER_KEY in _read_mcp (str (tmp_path ))["mcpServers" ]
110+
172111 def test_preserves_existing_mcp_servers (self , tmp_path ):
173112 existing = {
174113 "mcpServers" : {
@@ -200,51 +139,18 @@ def test_overwrites_existing_jira_server_entry(self, tmp_path):
200139 assert entry ["url" ] == JIRA_MCP_URL
201140 assert "stale" not in entry ["headers" ]["Authorization" ]
202141
203- def test_linear_and_jira_can_coexist (self , tmp_path ):
204- # Belt-and-braces: a repo that committed a Linear entry and then
205- # gets onboarded to Jira (or vice-versa) must keep both. The current
206- # code path only writes one channel per run, but this test guards
207- # against a future refactor that writes the wrong key.
208- configure_channel_mcp (str (tmp_path ), "linear" )
142+ def test_tolerates_mcp_json_without_mcpservers_key (self , tmp_path ):
143+ (tmp_path / ".mcp.json" ).write_text (json .dumps ({"version" : 1 }))
209144 configure_channel_mcp (str (tmp_path ), "jira" )
210145 merged = _read_mcp (str (tmp_path ))
211- assert LINEAR_MCP_SERVER_KEY in merged ["mcpServers" ]
146+ assert merged ["version" ] == 1
212147 assert JIRA_MCP_SERVER_KEY in merged ["mcpServers" ]
213148
214-
215- class TestLinearGatewayRouting :
216- """channel_metadata.gateway_url routes the Linear MCP through the workspace
217- gateway (F15/F16); absence keeps the direct hosted path."""
218-
219- def _entry (self , tmp_path ):
220- from channel_mcp import LINEAR_MCP_SERVER_KEY
221-
222- return _read_mcp (str (tmp_path ))["mcpServers" ][LINEAR_MCP_SERVER_KEY ]
223-
224- def test_routes_through_gateway_when_url_and_token_present (self , tmp_path , monkeypatch ):
225- # _build_linear_entry imports gateway_auth lazily, so patch it there.
226- monkeypatch .setattr ("gateway_auth.get_gateway_bearer_token" , lambda : "m2m-token-xyz" )
227- gw = "https://gw-abc.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp"
228- wrote = configure_channel_mcp (str (tmp_path ), "linear" , {"gateway_url" : gw })
149+ def test_malformed_mcp_json_is_replaced (self , tmp_path ):
150+ # Malformed JSON is treated as absent (logged as a warning in shell.log)
151+ # rather than crashing the pipeline.
152+ (tmp_path / ".mcp.json" ).write_text ("{not json" )
153+ wrote = configure_channel_mcp (str (tmp_path ), "jira" )
229154 assert wrote is True
230- entry = self ._entry (tmp_path )
231- assert entry ["url" ] == gw
232- assert entry ["headers" ]["Authorization" ] == "Bearer m2m-token-xyz"
233-
234- def test_falls_back_to_direct_when_no_gateway_url (self , tmp_path ):
235- configure_channel_mcp (str (tmp_path ), "linear" , {"linear_oauth_secret_arn" : "arn:x" })
236- entry = self ._entry (tmp_path )
237- assert entry ["url" ] == LINEAR_MCP_URL
238- assert entry ["headers" ]["Authorization" ] == f"Bearer ${{{ LINEAR_API_TOKEN_ENV }
239-
240- def test_falls_back_to_direct_when_token_mint_fails (self , tmp_path , monkeypatch ):
241- # gateway_url present but the M2M token mint returns "" → direct path.
242- monkeypatch .setattr ("gateway_auth.get_gateway_bearer_token" , lambda : "" )
243- gw = "https://gw-abc.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp"
244- configure_channel_mcp (str (tmp_path ), "linear" , {"gateway_url" : gw })
245- entry = self ._entry (tmp_path )
246- assert entry ["url" ] == LINEAR_MCP_URL
247-
248- def test_none_metadata_is_direct_path (self , tmp_path ):
249- configure_channel_mcp (str (tmp_path ), "linear" , None )
250- assert self ._entry (tmp_path )["url" ] == LINEAR_MCP_URL
155+ merged = _read_mcp (str (tmp_path ))
156+ assert JIRA_MCP_SERVER_KEY in merged ["mcpServers" ]
0 commit comments