test_health_cache_rbac.py•1.79 kB
import json
import pytest
from unittest import mock
import grpc
from src.server import MCPServicer
from src.generated import mcp_pb2
@pytest.fixture
def servicer():
return MCPServicer()
def test_health_check(servicer):
"""Test health check returns status for all adapters."""
req = mcp_pb2.HealthRequest()
resp = servicer.HealthCheck(req, grpc.ServicerContext())
assert len(resp.adapters) > 0
for adapter in resp.adapters:
assert adapter.name in ["kubernetes", "github", "azure_identity"]
assert adapter.ok is True
assert adapter.detail == "ok"
def test_cache_behavior(servicer):
"""Test that caching works correctly."""
with mock.patch('kubernetes.client.CoreV1Api') as mock_api:
mock_instance = mock_api.return_value
mock_instance.list_namespace.return_value = mock.MagicMock(
items=[mock.MagicMock(metadata=mock.MagicMock(name="default"))]
)
# First call
req = mcp_pb2.InvokeRequest(
fq_name="kubernetes.CoreV1Api.list_namespace",
json_args='{}'
)
resp1 = servicer.Invoke(req, grpc.ServicerContext())
# Second call should use cache
resp2 = servicer.Invoke(req, grpc.ServicerContext())
assert resp1.json_result == resp2.json_result
def test_rbac_deny(servicer):
"""Test that RBAC policy denies access when configured."""
# This would require setting up a deny policy
# For now, just test that the permission check exists
from src.utils import check_permission
# Default policy should allow everything
assert check_permission("kubernetes.CoreV1Api.list_namespace") is True
assert check_permission("github.MainClass.Github.get_user") is True