| import unittest | |
| from importlib.metadata import PackageNotFoundError | |
| from unittest.mock import patch | |
| from gerrit_mcp_server.http_headers import ( | |
| RequestObservabilityContext, | |
| build_curl_header_args, | |
| ) | |
| def _make_obs(**kwargs): | |
| defaults = dict( | |
| tool_name="my_tool", | |
| client_name="test-client", | |
| client_version="1.2.3", | |
| protocol_version="2025-11-25", | |
| ) | |
| defaults.update(kwargs) | |
| return RequestObservabilityContext(**defaults) | |
| class TestBuildCurlHeaderArgs(unittest.TestCase): | |
| def test_no_obs_returns_static_ua_only(self): | |
| result = build_curl_header_args(None) | |
| self.assertEqual(result[0], "-A") | |
| ua = result[1] | |
| self.assertIn("gerrit-mcp-server/", ua) | |
| self.assertIn("mcp-python-sdk/", ua) | |
| # No X- headers | |
| self.assertNotIn("-H", result) | |
| def test_full_obs_has_three_ua_tokens_and_all_x_headers(self): | |
| result = build_curl_header_args(_make_obs()) | |
| ua = result[1] | |
| tokens = ua.split(" ") | |
| self.assertEqual(len(tokens), 3) | |
| self.assertTrue(tokens[0].startswith("gerrit-mcp-server/")) | |
| self.assertTrue(tokens[1].startswith("mcp-python-sdk/")) | |
| self.assertEqual(tokens[2], "test-client/1.2.3") | |
| self.assertIn("X-MCP-Tool: my_tool", result) | |
| self.assertIn("X-MCP-Client: test-client", result) | |
| self.assertIn("X-MCP-Protocol-Version: 2025-11-25", result) | |
| def test_missing_client_name_omits_x_mcp_client_and_ua_token(self): | |
| result = build_curl_header_args(_make_obs(client_name=None)) | |
| ua = result[1] | |
| tokens = ua.split(" ") | |
| self.assertEqual(len(tokens), 2) | |
| self.assertNotIn("X-MCP-Client: None", result) | |
| header_values = [result[i + 1] for i, v in enumerate(result) if v == "-H"] | |
| self.assertFalse(any("X-MCP-Client" in h for h in header_values)) | |
| def test_missing_client_version_uses_unknown_in_ua(self): | |
| result = build_curl_header_args(_make_obs(client_version=None)) | |
| ua = result[1] | |
| self.assertIn("test-client/unknown", ua) | |
| def test_missing_protocol_version_omits_x_mcp_protocol_version(self): | |
| result = build_curl_header_args(_make_obs(protocol_version=None)) | |
| header_values = [result[i + 1] for i, v in enumerate(result) if v == "-H"] | |
| self.assertFalse(any("X-MCP-Protocol-Version" in h for h in header_values)) | |
| def test_version_fallback_on_package_not_found(self): | |
| with patch( | |
| "gerrit_mcp_server.http_headers.version", | |
| side_effect=PackageNotFoundError("pkg"), | |
| ): | |
| result = build_curl_header_args(None) | |
| ua = result[1] | |
| self.assertIn("gerrit-mcp-server/unknown", ua) | |
| self.assertIn("mcp-python-sdk/unknown", ua) | |
| def test_client_name_with_crlf_becomes_unparseable(self): | |
| result = build_curl_header_args( | |
| _make_obs(client_name="evil\r\nAuthorization: Bearer x") | |
| ) | |
| ua = result[1] | |
| self.assertIn("unparseable/", ua) | |
| self.assertNotIn("\r", ua) | |
| self.assertNotIn("\n", ua) | |
| self.assertIn("X-MCP-Client: unparseable", result) | |
| def test_client_version_with_crlf_becomes_unparseable(self): | |
| result = build_curl_header_args( | |
| _make_obs(client_version="1.0\r\nX-Injected: yes") | |
| ) | |
| ua = result[1] | |
| self.assertIn("test-client/unparseable", ua) | |
| self.assertNotIn("\r", ua) | |
| self.assertNotIn("\n", ua) | |
| def test_protocol_version_with_crlf_becomes_unparseable(self): | |
| result = build_curl_header_args( | |
| _make_obs(protocol_version="2025-11-25\r\nX-Injected: yes") | |
| ) | |
| header_values = [result[i + 1] for i, v in enumerate(result) if v == "-H"] | |
| proto_header = next(h for h in header_values if "X-MCP-Protocol-Version" in h) | |
| self.assertEqual(proto_header, "X-MCP-Protocol-Version: unparseable") | |
| self.assertNotIn("\r", proto_header) | |
| self.assertNotIn("\n", proto_header) | |
| def test_tool_name_with_control_char_becomes_unparseable(self): | |
| result = build_curl_header_args(_make_obs(tool_name="tool\x00name")) | |
| header_values = [result[i + 1] for i, v in enumerate(result) if v == "-H"] | |
| tool_header = next(h for h in header_values if "X-MCP-Tool" in h) | |
| self.assertEqual(tool_header, "X-MCP-Tool: unparseable") | |
| def test_clean_values_pass_through_unchanged(self): | |
| result = build_curl_header_args(_make_obs()) | |
| self.assertIn("X-MCP-Client: test-client", result) | |
| self.assertIn("X-MCP-Protocol-Version: 2025-11-25", result) | |
| self.assertIn("X-MCP-Tool: my_tool", result) |