Chris Christiansen
f0039088c2
Some checks are pending
Check Python Version Consistency / Check Python Version (push) Waiting to run
- Add a private helper that performs an authenticated, read-only GET for the configured opax-mcp Cloud Run v2 service. - Return only generation, observedGeneration, and updateTime on success, with sanitized reason codes for token, HTTP, network, timeout, and parsing errors. - Add mocked unit coverage for successful responses and handled failure paths.
203 lines
9.0 KiB
Python
203 lines
9.0 KiB
Python
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()
|