import unittest import json import socket import urllib.error from unittest.mock import patch, MagicMock from agents.mcp_server.server import ( _fetch_opax_deployment_service_data, _sanitize_opax_deployment_status, ) class TestSanitizeOpaxDeploymentStatus(unittest.TestCase): def test_valid_inputs(self): """Tests valid integer/string generations and a valid UTC timestamp.""" service_data = { "generation": 10, "observedGeneration": "9", "updateTime": "2023-10-27T10:00:00Z" } result = _sanitize_opax_deployment_status(service_data) self.assertEqual(result["generation"], 10) self.assertEqual(result["observed_generation"], 9) self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z") self.assertIsNone(result["reason_code"]) def test_zero_values(self): """Tests handling of zero for generation values.""" service_data = {"generation": 0, "observedGeneration": "0"} result = _sanitize_opax_deployment_status(service_data) self.assertEqual(result["generation"], 0) self.assertEqual(result["observed_generation"], 0) def test_invalid_generation_values(self): """Tests various invalid generation inputs.""" invalid_inputs = [ True, False, -1, 1.5, "-1", " 1 ", "abc", "١", None, {}, [] ] for val in invalid_inputs: with self.subTest(val=val): res1 = _sanitize_opax_deployment_status({"generation": val}) res2 = _sanitize_opax_deployment_status({"observedGeneration": val}) self.assertIsNone(res1["generation"]) self.assertIsNone(res2["observed_generation"]) def test_fractional_second_timestamp_is_truncated(self): """Tests that fractional seconds are correctly truncated.""" service_data = {"updateTime": "2023-10-27T10:00:00.123456Z"} result = _sanitize_opax_deployment_status(service_data) self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z") def test_invalid_and_non_z_timestamps(self): """Tests invalid, tz-naive, missing, and non-Z timestamps.""" invalid_timestamps = [ "2023-10-27T10:00:00", # Missing Z (naive) "2023-10-27T10:00:00ZZ", # Multiple Z "2023-10-27T11:00:00+01:00", # Valid but not Z-suffix "2023-10-27 10:00:00Z", # Malformed "not-a-timestamp", # Invalid 1698399600, # Not a string None, # Missing ] for ts in invalid_timestamps: with self.subTest(ts=ts): result = _sanitize_opax_deployment_status({"updateTime": ts}) self.assertIsNone(result["last_update_time"]) def test_exact_key_set_and_fixed_values(self): """Tests for exact key set and fixed service/region/reason_code.""" result = _sanitize_opax_deployment_status({}) expected_keys = { "service", "region", "generation", "observed_generation", "last_update_time", "reason_code" } self.assertEqual(set(result.keys()), expected_keys) self.assertEqual(result["service"], "opax-mcp") self.assertEqual(result["region"], "us-central1") self.assertIsNone(result["reason_code"]) class TestFetchOpaxDeploymentServiceData(unittest.TestCase): @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_success_filters_fields(self, mock_get_token, mock_urlopen): mock_get_token.return_value = "fake-token" mock_response = MagicMock() mock_response.status = 200 mock_response.read.return_value = json.dumps({ "generation": 5, "observedGeneration": 5, "updateTime": "T", "uri": "forbidden" }).encode() mock_urlopen.return_value.__enter__.return_value = mock_response result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"generation": 5, "observedGeneration": 5, "updateTime": "T"}) self.assertEqual(set(result.keys()), {"generation", "observedGeneration", "updateTime"}) @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_success_missing_fields(self, mock_get_token, mock_urlopen): mock_get_token.return_value = "fake-token" mock_response = MagicMock() mock_response.status = 200 mock_response.read.return_value = json.dumps({"generation": 1}).encode() mock_urlopen.return_value.__enter__.return_value = mock_response result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"generation": 1, "observedGeneration": None, "updateTime": None}) @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_falsy_token_skips_call(self, mock_get_token, mock_urlopen): for token in [None, ""]: with self.subTest(token=token): mock_get_token.return_value = token result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": "ADC_TOKEN_UNAVAILABLE"}) mock_urlopen.assert_not_called() @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_get_token_exception(self, mock_get_token, mock_urlopen): mock_get_token.side_effect = Exception("local failure") result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": "UNEXPECTED_LOCAL_FAILURE"}) mock_urlopen.assert_not_called() @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_absent_response_status(self, mock_get_token, mock_urlopen): mock_get_token.return_value = "fake-token" class MockResp: def __init__(self): self.read_called = False def read(self): self.read_called = True return b"" def __enter__(self): return self def __exit__(self, *args): return False mock_response = MockResp() mock_urlopen.return_value = mock_response result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": "UNEXPECTED_LOCAL_FAILURE"}) self.assertFalse(mock_response.read_called) @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_non_2xx_status_skips_read(self, mock_get_token, mock_urlopen): mock_get_token.return_value = "fake-token" mock_response = MagicMock() mock_response.status = 503 mock_urlopen.return_value.__enter__.return_value = mock_response result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": "UNEXPECTED_NON_2XX_STATUS"}) mock_response.read.assert_not_called() @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_http_and_network_errors(self, mock_get_token, mock_urlopen): mock_get_token.return_value = "fake-token" errors_to_reasons = [ (urllib.error.HTTPError(None, 401, "", {}, None), "CLOUD_RUN_UNAUTHORIZED"), (urllib.error.HTTPError(None, 403, "", {}, None), "CLOUD_RUN_UNAUTHORIZED"), (urllib.error.HTTPError(None, 404, "", {}, None), "CLOUD_RUN_NOT_FOUND"), (urllib.error.HTTPError(None, 500, "", {}, None), "UNEXPECTED_NON_2XX_STATUS"), (socket.timeout(), "REQUEST_TIMEOUT"), (urllib.error.URLError("DNS failure"), "NETWORK_FAILURE"), ] for error, reason in errors_to_reasons: with self.subTest(error=error.__class__.__name__, code=getattr(error, 'code', 'N/A')): mock_urlopen.side_effect = error result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": reason}) mock_urlopen.side_effect = None @patch('urllib.request.urlopen') @patch('agents.mcp_server.server._get_access_token') def test_invalid_json_body(self, mock_get_token, mock_urlopen): mock_get_token.return_value = "fake-token" mock_response = MagicMock() mock_response.status = 200 mock_urlopen.return_value.__enter__.return_value = mock_response # Malformed JSON with self.subTest(case="malformed"): mock_response.read.return_value = b'{"key":' result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": "INVALID_JSON_RESPONSE"}) # Valid JSON, but not a dictionary object with self.subTest(case="non-object"): mock_response.read.return_value = b'[1, 2, 3]' result = _fetch_opax_deployment_service_data() self.assertEqual(result, {"reason_code": "INVALID_RESPONSE_SHAPE"}) if __name__ == '__main__': unittest.main()