Doctest 60 closes #249 (#529)

* Improve doctest coverage

Signed-off-by: Mihai Criveti <[email protected]>

* Improve doctest coverage 60%

Signed-off-by: Mihai Criveti <[email protected]>

---------

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

.github/workflows/pytest.yml

@@ -70,7 +70,7 @@ jobs:
           pip install pytest pytest-cov pytest-asyncio coverage[toml]
 
       # -----------------------------------------------------------
-      # 3️⃣  Run the tests with coverage
+      # 3️⃣  Run the tests with coverage (fail under 80% coverage)
       # -----------------------------------------------------------
       - name: 🧪  Run pytest
         run: |
@@ -80,14 +80,20 @@ jobs:
             --cov-report=html \
             --cov-report=term \
             --cov-branch \
-            --cov-fail-under=40
+            --cov-fail-under=80
 
       # -----------------------------------------------------------
-      # 4️⃣  Run doctests
+      # 4️⃣  Run doctests (fail under 55% coverage)
       # -----------------------------------------------------------
-      - name: 🧪  Run doctests
+      - name: 📊  Doctest coverage with threshold
         run: |
-          pytest --doctest-modules mcpgateway/ --tb=short
+          # Run doctests with coverage measurement
+          pytest --doctest-modules mcpgateway/ \
+            --cov=mcpgateway \
+            --cov-report=term \
+            --cov-report=json:doctest-coverage.json \
+            --cov-fail-under=55 \
+            --tb=short
 
       # -----------------------------------------------------------
       # 5️⃣  Doctest coverage check

mcpgateway/main.py

@@ -342,6 +342,28 @@ async def dispatch(self, request: Request, call_next):
 
         Returns:
             Response: Either the standard route response or a 401/403 error response.
+
+        Examples:
+            >>> import asyncio
+            >>> from unittest.mock import Mock, AsyncMock, patch
+            >>> from fastapi import HTTPException
+            >>> from fastapi.responses import JSONResponse
+            >>>
+            >>> # Test unprotected path - should pass through
+            >>> middleware = DocsAuthMiddleware(None)
+            >>> request = Mock()
+            >>> request.url.path = "/api/tools"
+            >>> request.headers.get.return_value = None
+            >>> call_next = AsyncMock(return_value="response")
+            >>>
+            >>> result = asyncio.run(middleware.dispatch(request, call_next))
+            >>> result
+            'response'
+            >>>
+            >>> # Test that middleware checks protected paths
+            >>> request.url.path = "/docs"
+            >>> isinstance(middleware, DocsAuthMiddleware)
+            True
         """
         protected_paths = ["/docs", "/redoc", "/openapi.json"]
 
@@ -386,6 +408,36 @@ async def __call__(self, scope, receive, send):
             scope (dict): The ASGI connection scope.
             receive (Callable): Awaitable that yields events from the client.
             send (Callable): Awaitable used to send events to the client.
+
+        Examples:
+            >>> import asyncio
+            >>> from unittest.mock import AsyncMock, patch
+            >>>
+            >>> # Test non-HTTP request passthrough
+            >>> app_mock = AsyncMock()
+            >>> middleware = MCPPathRewriteMiddleware(app_mock)
+            >>> scope = {"type": "websocket", "path": "/ws"}
+            >>> receive = AsyncMock()
+            >>> send = AsyncMock()
+            >>>
+            >>> asyncio.run(middleware(scope, receive, send))
+            >>> app_mock.assert_called_once_with(scope, receive, send)
+            >>>
+            >>> # Test path rewriting for /servers/123/mcp
+            >>> app_mock.reset_mock()
+            >>> scope = {"type": "http", "path": "/servers/123/mcp"}
+            >>> with patch('mcpgateway.main.streamable_http_auth', return_value=True):
+            ...     with patch.object(streamable_http_session, 'handle_streamable_http') as mock_handler:
+            ...         asyncio.run(middleware(scope, receive, send))
+            ...         scope["path"]
+            '/mcp'
+            >>>
+            >>> # Test regular path (no rewrite)
+            >>> scope = {"type": "http", "path": "/tools"}
+            >>> with patch('mcpgateway.main.streamable_http_auth', return_value=True):
+            ...     asyncio.run(middleware(scope, receive, send))
+            ...     scope["path"]
+            '/tools'
         """
         # Only handle HTTP requests, HTTPS uses scope["type"] == "http" in ASGI
         if scope["type"] != "http":

mcpgateway/schemas.py

@@ -60,6 +60,12 @@ def to_camel_case(s: str) -> str:
         'alreadyCamel'
         >>> to_camel_case("")
         ''
+        >>> to_camel_case("single")
+        'single'
+        >>> to_camel_case("_leading_underscore")
+        'LeadingUnderscore'
+        >>> to_camel_case("trailing_underscore_")
+        'trailingUnderscore'
     """
     return "".join(word.capitalize() if i else word for i, word in enumerate(s.split("_")))
 
@@ -120,6 +126,22 @@ def to_dict(self, use_alias: bool = False) -> Dict[str, Any]:
             >>> m = ExampleModel(foo=1, bar='baz')
             >>> m.to_dict()
             {'foo': 1, 'bar': 'baz'}
+
+            >>> # Test with alias
+            >>> m.to_dict(use_alias=True)
+            {'foo': 1, 'bar': 'baz'}
+
+            >>> # Test with nested model
+            >>> class NestedModel(BaseModelWithConfigDict):
+            ...     nested_field: int
+            >>> class ParentModel(BaseModelWithConfigDict):
+            ...     parent_field: str
+            ...     child: NestedModel
+            >>> nested = NestedModel(nested_field=42)
+            >>> parent = ParentModel(parent_field="test", child=nested)
+            >>> result = parent.to_dict()
+            >>> result['child']
+            {'nested_field': 42}
         """
         output = {}
         for key, value in self.model_dump(by_alias=use_alias).items():
@@ -407,6 +429,35 @@ def validate_request_type(cls, v: str, info: ValidationInfo) -> str:
 
         Raises:
             ValueError: When value is unsafe
+
+        Examples:
+            >>> # Test MCP integration types
+            >>> from pydantic import ValidationInfo
+            >>> info = type('obj', (object,), {'data': {'integration_type': 'MCP'}})
+            >>> ToolCreate.validate_request_type('SSE', info)
+            'SSE'
+
+            >>> # Test REST integration types
+            >>> info = type('obj', (object,), {'data': {'integration_type': 'REST'}})
+            >>> ToolCreate.validate_request_type('GET', info)
+            'GET'
+            >>> ToolCreate.validate_request_type('POST', info)
+            'POST'
+
+            >>> # Test invalid REST type
+            >>> try:
+            ...     ToolCreate.validate_request_type('SSE', info)
+            ... except ValueError as e:
+            ...     "not allowed for REST" in str(e)
+            True
+
+            >>> # Test invalid MCP type
+            >>> info = type('obj', (object,), {'data': {'integration_type': 'MCP'}})
+            >>> try:
+            ...     ToolCreate.validate_request_type('GET', info)
+            ... except ValueError as e:
+            ...     "not allowed for MCP" in str(e)
+            True
         """
         data = info.data
         integration_type = data.get("integration_type")
@@ -434,6 +485,33 @@ def assemble_auth(cls, values: Dict[str, Any]) -> Dict[str, Any]:
 
         Returns:
             Dict: Reformatedd values dict
+
+        Examples:
+            >>> # Test basic auth
+            >>> values = {'auth_type': 'basic', 'auth_username': 'user', 'auth_password': 'pass'}
+            >>> result = ToolCreate.assemble_auth(values)
+            >>> 'auth' in result
+            True
+            >>> result['auth']['auth_type']
+            'basic'
+
+            >>> # Test bearer auth
+            >>> values = {'auth_type': 'bearer', 'auth_token': 'mytoken'}
+            >>> result = ToolCreate.assemble_auth(values)
+            >>> result['auth']['auth_type']
+            'bearer'
+
+            >>> # Test authheaders
+            >>> values = {'auth_type': 'authheaders', 'auth_header_key': 'X-API-Key', 'auth_header_value': 'secret'}
+            >>> result = ToolCreate.assemble_auth(values)
+            >>> result['auth']['auth_type']
+            'authheaders'
+
+            >>> # Test no auth type
+            >>> values = {'name': 'test'}
+            >>> result = ToolCreate.assemble_auth(values)
+            >>> 'auth' in result
+            False
         """
         logger.debug(
             "Assembling auth in ToolCreate with raw values",
@@ -519,6 +597,17 @@ def validate_description(cls, v: Optional[str]) -> Optional[str]:
 
         Raises:
             ValueError: When value is unsafe
+
+        Examples:
+            >>> from mcpgateway.schemas import ResourceCreate
+            >>> ResourceCreate.validate_description('A safe description')
+            'A safe description'
+            >>> ResourceCreate.validate_description(None)  # Test None case
+
+            >>> ResourceCreate.validate_description('x' * 5000)
+            Traceback (most recent call last):
+                ...
+            ValueError: ...
         """
         if v is None:
             return v

mcpgateway/translate.py

@@ -137,6 +137,20 @@ async def publish(self, data: str) -> None:
             ...     return True
             >>> asyncio.run(test_publish())
             True
+
+            >>> # Test queue full handling
+            >>> async def test_full_queue():
+            ...     pubsub = _PubSub()
+            ...     # Create a queue with size 1
+            ...     q = asyncio.Queue(maxsize=1)
+            ...     pubsub._subscribers = [q]
+            ...     # Fill the queue
+            ...     await q.put("first")
+            ...     # This should remove the full queue
+            ...     await pubsub.publish("second")
+            ...     return len(pubsub._subscribers)
+            >>> asyncio.run(test_full_queue())
+            0
         """
         dead: List[asyncio.Queue[str]] = []
         for q in self._subscribers:
@@ -405,6 +419,12 @@ def _build_fastapi(
         True
         >>> "/send" in [r.path for r in app2.routes]
         True
+
+        >>> # Test CORS middleware is added
+        >>> app3 = _build_fastapi(pubsub, stdio, cors_origins=["http://example.com"])
+        >>> # Check that middleware stack includes CORSMiddleware
+        >>> any("CORSMiddleware" in str(m) for m in app3.user_middleware)
+        True
     """
     app = FastAPI()
 

mcpgateway/utils/db_isready.py

@@ -320,6 +320,44 @@ def _parse_cli() -> argparse.Namespace:
 
     Returns:
         Parsed :class:`argparse.Namespace` holding all CLI options.
+
+    Examples:
+        >>> import sys
+        >>> # Save original argv
+        >>> original_argv = sys.argv
+        >>>
+        >>> # Test default values
+        >>> sys.argv = ['db_isready.py']
+        >>> args = _parse_cli()
+        >>> args.database_url == DEFAULT_DB_URL
+        True
+        >>> args.max_tries == DEFAULT_MAX_TRIES
+        True
+        >>> args.interval == DEFAULT_INTERVAL
+        True
+        >>> args.timeout == DEFAULT_TIMEOUT
+        True
+        >>> args.log_level == DEFAULT_LOG_LEVEL
+        True
+
+        >>> # Test custom values
+        >>> sys.argv = ['db_isready.py', '--database-url', 'postgresql://localhost/test',
+        ...             '--max-tries', '5', '--interval', '1.5', '--timeout', '10',
+        ...             '--log-level', 'DEBUG']
+        >>> args = _parse_cli()
+        >>> args.database_url
+        'postgresql://localhost/test'
+        >>> args.max_tries
+        5
+        >>> args.interval
+        1.5
+        >>> args.timeout
+        10
+        >>> args.log_level
+        'DEBUG'
+
+        >>> # Restore original argv
+        >>> sys.argv = original_argv
     """
 
     parser = argparse.ArgumentParser(