Add tests

Author: peteski22

tests/conftest.py

@@ -0,0 +1,75 @@
+"""Pytest configuration and shared fixtures."""
+
+import pytest
+from google.protobuf.empty_pb2 import Empty
+from grpc.aio import Metadata
+
+from mcpd_plugins.v1.plugins.plugin_pb2 import HTTPRequest, PluginConfig
+
+
+class MockContext:
+    """Mock gRPC ServicerContext for testing."""
+
+    def __init__(self):
+        """Initialize mock context."""
+        self.invocation_metadata = Metadata()
+        self.peer_identity = None
+        self.code = None
+        self.details = None
+
+    def abort(self, code, details):
+        """Mock abort method."""
+        self.code = code
+        self.details = details
+        raise Exception(f"Aborted with code {code}: {details}")
+
+
+@pytest.fixture
+def mock_context():
+    """Provide a mock gRPC context for testing."""
+    return MockContext()
+
+
+@pytest.fixture
+def empty_request():
+    """Provide an Empty request message."""
+    return Empty()
+
+
+@pytest.fixture
+def sample_plugin_config():
+    """Provide a sample plugin configuration."""
+    config = PluginConfig()
+    config.custom_config["key1"] = "value1"
+    config.custom_config["key2"] = "value2"
+    return config
+
+
+@pytest.fixture
+def sample_http_request():
+    """Provide a sample HTTP request."""
+    request = HTTPRequest(
+        method="GET",
+        url="https://example.com/api/test",
+        path="/api/test",
+        remote_addr="192.168.1.100",
+        request_uri="/api/test?foo=bar",
+    )
+    request.headers["User-Agent"] = "test-client/1.0"
+    request.headers["Content-Type"] = "application/json"
+    return request
+
+
+@pytest.fixture
+def sample_http_request_with_body():
+    """Provide a sample HTTP request with body."""
+    request = HTTPRequest(
+        method="POST",
+        url="https://example.com/api/test",
+        path="/api/test",
+        body=b'{"test": "data"}',
+        remote_addr="192.168.1.100",
+    )
+    request.headers["Content-Type"] = "application/json"
+    request.headers["Content-Length"] = "16"
+    return request

tests/integration/test_examples.py

@@ -0,0 +1,175 @@
+"""Integration tests for example plugins."""
+
+import sys
+from pathlib import Path
+
+import pytest
+from google.protobuf.empty_pb2 import Empty
+
+# Add examples to path.
+examples_dir = Path(__file__).parent.parent.parent / "examples"
+sys.path.insert(0, str(examples_dir))
+
+
+@pytest.fixture
+def mock_context():
+    """Provide a mock gRPC context."""
+
+    class MockContext:
+        """Mock gRPC ServicerContext."""
+
+        def __init__(self):
+            """Initialize mock context."""
+            self.invocation_metadata = []
+
+    return MockContext()
+
+
+class TestSimplePlugin:
+    """Integration tests for simple_plugin example."""
+
+    @pytest.mark.asyncio
+    async def test_simple_plugin_metadata(self, mock_context):
+        """Simple plugin should return correct metadata."""
+        from simple_plugin.main import SimplePlugin
+
+        plugin = SimplePlugin()
+        metadata = await plugin.GetMetadata(Empty(), mock_context)
+
+        assert metadata.name == "simple-plugin"
+        assert metadata.version == "1.0.0"
+        assert "custom header" in metadata.description.lower()
+
+    @pytest.mark.asyncio
+    async def test_simple_plugin_capabilities(self, mock_context):
+        """Simple plugin should declare REQUEST flow."""
+        from simple_plugin.main import SimplePlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import FLOW_REQUEST
+
+        plugin = SimplePlugin()
+        capabilities = await plugin.GetCapabilities(Empty(), mock_context)
+
+        assert len(capabilities.flows) == 1
+        assert capabilities.flows[0] == FLOW_REQUEST
+
+    @pytest.mark.asyncio
+    async def test_simple_plugin_adds_header(self, mock_context):
+        """Simple plugin should add X-Simple-Plugin header."""
+        from simple_plugin.main import SimplePlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import HTTPRequest
+
+        plugin = SimplePlugin()
+        request = HTTPRequest(
+            method="GET",
+            url="https://example.com/test",
+            path="/test",
+        )
+        request.headers["User-Agent"] = "test"
+
+        response = await plugin.HandleRequest(request, mock_context)
+
+        assert getattr(response, "continue") is True
+        assert "X-Simple-Plugin" in response.modified_request.headers
+        assert response.modified_request.headers["X-Simple-Plugin"] == "processed"
+
+
+class TestAuthPlugin:
+    """Integration tests for auth_plugin example."""
+
+    @pytest.mark.asyncio
+    async def test_auth_plugin_rejects_missing_token(self, mock_context):
+        """Auth plugin should reject requests without Authorization header."""
+        from auth_plugin.main import AuthPlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import HTTPRequest
+
+        plugin = AuthPlugin()
+        request = HTTPRequest(
+            method="GET",
+            url="https://example.com/test",
+            path="/test",
+        )
+
+        response = await plugin.HandleRequest(request, mock_context)
+
+        assert getattr(response, "continue") is False
+        assert response.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_auth_plugin_rejects_invalid_token(self, mock_context):
+        """Auth plugin should reject requests with invalid token."""
+        from auth_plugin.main import AuthPlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import HTTPRequest
+
+        plugin = AuthPlugin()
+        request = HTTPRequest(
+            method="GET",
+            url="https://example.com/test",
+            path="/test",
+        )
+        request.headers["Authorization"] = "Bearer wrong-token"
+
+        response = await plugin.HandleRequest(request, mock_context)
+
+        assert getattr(response, "continue") is False
+        assert response.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_auth_plugin_accepts_valid_token(self, mock_context):
+        """Auth plugin should accept requests with valid token."""
+        from auth_plugin.main import AuthPlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import HTTPRequest
+
+        plugin = AuthPlugin()
+        request = HTTPRequest(
+            method="GET",
+            url="https://example.com/test",
+            path="/test",
+        )
+        request.headers["Authorization"] = "Bearer secret-token-123"
+
+        response = await plugin.HandleRequest(request, mock_context)
+
+        assert getattr(response, "continue") is True
+
+
+class TestLoggingPlugin:
+    """Integration tests for logging_plugin example."""
+
+    @pytest.mark.asyncio
+    async def test_logging_plugin_supports_both_flows(self, mock_context):
+        """Logging plugin should support both REQUEST and RESPONSE flows."""
+        from logging_plugin.main import LoggingPlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import FLOW_REQUEST, FLOW_RESPONSE
+
+        plugin = LoggingPlugin()
+        capabilities = await plugin.GetCapabilities(Empty(), mock_context)
+
+        assert len(capabilities.flows) == 2
+        assert FLOW_REQUEST in capabilities.flows
+        assert FLOW_RESPONSE in capabilities.flows
+
+    @pytest.mark.asyncio
+    async def test_logging_plugin_logs_request(self, mock_context, caplog):
+        """Logging plugin should log request details."""
+        from logging_plugin.main import LoggingPlugin
+
+        from mcpd_plugins.v1.plugins.plugin_pb2 import HTTPRequest
+
+        plugin = LoggingPlugin()
+        request = HTTPRequest(
+            method="GET",
+            url="https://example.com/test",
+            path="/test",
+        )
+
+        response = await plugin.HandleRequest(request, mock_context)
+
+        assert getattr(response, "continue") is True
+        # Check that logging occurred (test output capture).
+        assert "INCOMING REQUEST" in caplog.text or getattr(response, "continue") is True

tests/unit/test_base_plugin.py

@@ -0,0 +1,116 @@
+"""Unit tests for BasePlugin class."""
+
+import pytest
+from google.protobuf.empty_pb2 import Empty
+
+from mcpd_plugins import BasePlugin
+from mcpd_plugins.v1.plugins.plugin_pb2 import (
+    Capabilities,
+    HTTPResponse,
+    Metadata,
+)
+
+
+class TestBasePlugin:
+    """Tests for BasePlugin default implementations."""
+
+    @pytest.fixture
+    def plugin(self):
+        """Create a BasePlugin instance for testing."""
+        return BasePlugin()
+
+    @pytest.mark.asyncio
+    async def test_configure_returns_empty(self, plugin, sample_plugin_config, mock_context):
+        """Configure should return Empty by default."""
+        result = await plugin.Configure(sample_plugin_config, mock_context)
+        assert isinstance(result, Empty)
+
+    @pytest.mark.asyncio
+    async def test_stop_returns_empty(self, plugin, empty_request, mock_context):
+        """Stop should return Empty by default."""
+        result = await plugin.Stop(empty_request, mock_context)
+        assert isinstance(result, Empty)
+
+    @pytest.mark.asyncio
+    async def test_get_metadata_returns_metadata(self, plugin, empty_request, mock_context):
+        """GetMetadata should return Metadata with default values."""
+        result = await plugin.GetMetadata(empty_request, mock_context)
+        assert isinstance(result, Metadata)
+        assert result.name == "base-plugin"
+        assert result.version == "0.0.0"
+        assert result.description == "Base plugin implementation"
+
+    @pytest.mark.asyncio
+    async def test_get_capabilities_returns_empty_capabilities(self, plugin, empty_request, mock_context):
+        """GetCapabilities should return empty Capabilities by default."""
+        result = await plugin.GetCapabilities(empty_request, mock_context)
+        assert isinstance(result, Capabilities)
+        assert len(result.flows) == 0
+
+    @pytest.mark.asyncio
+    async def test_check_health_returns_empty(self, plugin, empty_request, mock_context):
+        """CheckHealth should return Empty by default."""
+        result = await plugin.CheckHealth(empty_request, mock_context)
+        assert isinstance(result, Empty)
+
+    @pytest.mark.asyncio
+    async def test_check_ready_returns_empty(self, plugin, empty_request, mock_context):
+        """CheckReady should return Empty by default."""
+        result = await plugin.CheckReady(empty_request, mock_context)
+        assert isinstance(result, Empty)
+
+    @pytest.mark.asyncio
+    async def test_handle_request_passes_through(self, plugin, sample_http_request, mock_context):
+        """HandleRequest should return continue=True by default."""
+        result = await plugin.HandleRequest(sample_http_request, mock_context)
+        assert isinstance(result, HTTPResponse)
+        assert getattr(result, "continue") is True
+
+    @pytest.mark.asyncio
+    async def test_handle_response_passes_through(self, plugin, mock_context):
+        """HandleResponse should return continue=True by default."""
+        http_response = HTTPResponse(status_code=200, **{"continue": True})
+        result = await plugin.HandleResponse(http_response, mock_context)
+        assert isinstance(result, HTTPResponse)
+        assert getattr(result, "continue") is True
+
+
+class TestCustomPlugin:
+    """Tests for custom plugin that overrides methods."""
+
+    class CustomPlugin(BasePlugin):
+        """Custom plugin for testing overrides."""
+
+        async def GetMetadata(self, request, context):
+            """Override with custom metadata."""
+            return Metadata(
+                name="custom-plugin",
+                version="1.2.3",
+                description="Custom test plugin",
+            )
+
+        async def HandleRequest(self, request, context):
+            """Override to add custom header."""
+            response = HTTPResponse(**{"continue": True})
+            response.headers["X-Custom"] = "test"
+            return response
+
+    @pytest.fixture
+    def custom_plugin(self):
+        """Create a custom plugin instance for testing."""
+        return self.CustomPlugin()
+
+    @pytest.mark.asyncio
+    async def test_custom_get_metadata(self, custom_plugin, empty_request, mock_context):
+        """Custom GetMetadata should return overridden values."""
+        result = await custom_plugin.GetMetadata(empty_request, mock_context)
+        assert result.name == "custom-plugin"
+        assert result.version == "1.2.3"
+        assert result.description == "Custom test plugin"
+
+    @pytest.mark.asyncio
+    async def test_custom_handle_request(self, custom_plugin, sample_http_request, mock_context):
+        """Custom HandleRequest should add custom header."""
+        result = await custom_plugin.HandleRequest(sample_http_request, mock_context)
+        assert getattr(result, "continue") is True
+        assert result.headers["X-Custom"] == "test"