fix: review fixes — 100% coverage, typo, regex hoist, docstring path

- Fix typo: validatated_headers → validated_headers
- Hoist RFC 9110 token regex to module-level compiled constant
- Consolidate redundant CRLF check into unified CTL validation
- Fix integration test docstring path (was unit test path)
- Add tests for _parse_rate minute/hour/error branches
- Add tests for unlimited (no-limit) prompt and tool paths
- Achieve 100% differential test coverage on rate_limiter.py

Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>

Author: crivetimihai

mcpgateway/main.py

@@ -1452,6 +1452,14 @@ async def database_exception_handler(_request: Request, exc: IntegrityError):
     return ORJSONResponse(status_code=409, content=ErrorFormatter.format_database_error(exc))
 
 
+# RFC 9110 §5.6.2 'token' pattern for header field names:
+#   token = 1*tchar
+#   tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*"
+#           / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~"
+#           / DIGIT / ALPHA
+_RFC9110_TOKEN_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")
+
+
 def _validate_http_headers(headers: dict[str, str]) -> Optional[dict[str, str]]:
     """Validate headers according to RFC 9110.
 
@@ -1464,34 +1472,24 @@ def _validate_http_headers(headers: dict[str, str]) -> Optional[dict[str, str]]:
     Rules enforced:
       - Header name must match RFC 9110 'token'.
       - No whitespace before colon (enforced by dictionary usage).
-      - Header value must not contain CTL characters (0x00–0x1F, 0x7F).
+      - Header value must not contain CTL characters (0x00–0x1F, 0x7F),
+        except SP (0x20) and HTAB (0x09) which are allowed.
     """
-
-    # RFC 9110 'token' definition:
-    # token = 1*tchar
-    # tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*"
-    #         / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~"
-    #         / DIGIT / ALPHA
-    header_key = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")
     validated: dict[str, str] = {}
     for key, value in headers.items():
-        # Validate header name (RFC 9110)
-        if not re.match(header_key, key):
+        # Validate header name (RFC 9110 token)
+        if not _RFC9110_TOKEN_RE.match(key):
             logger.warning(f"Invalid header name: {key}")
             continue
-        # Validate header value (no CRLF)
-        if "\r" in value or "\n" in value:
-            logger.warning(f"Header value contains CRLF: {key}")
-            continue
         # RFC 9110: Reject CTLs (0x00–0x1F, 0x7F). Allow SP (0x20) and HTAB (0x09).
-        # Further structure (quoted-string, lists, parameters) is left to higher-level parsers.
         valid = True
         for ch in value:
             code = ord(ch)
             if (0 <= code <= 31 or code == 127) and code not in (9, 32):
                 valid = False
                 break
         if not valid:
+            logger.warning(f"Header value contains invalid characters: {key}")
             continue
         validated[key] = value
     return validated if validated else None
@@ -1576,9 +1574,9 @@ async def plugin_violation_exception_handler(_request: Request, exc: PluginViola
 
     response = ORJSONResponse(status_code=http_status, content={"error": json_rpc_error.model_dump()})
     if headers:
-        validatated_headers = _validate_http_headers(headers)
-        if validatated_headers:
-            response.headers.update(validatated_headers)
+        validated_headers = _validate_http_headers(headers)
+        if validated_headers:
+            response.headers.update(validated_headers)
     return response
 
 

tests/integration/test_rate_limiter.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""Location: ./tests/unit/mcpgateway/plugins/plugins/rate_limiter/test_rate_limiter.py
+"""Location: ./tests/integration/test_rate_limiter.py
 Copyright 2025
 SPDX-License-Identifier: Apache-2.0
 Authors: Mihai Criveti

tests/unit/mcpgateway/plugins/plugins/rate_limiter/test_rate_limiter.py

@@ -17,7 +17,7 @@
     PromptPrehookPayload,
     ToolHookType
 )
-from plugins.rate_limiter.rate_limiter import RateLimiterPlugin, _make_headers, _select_most_restrictive, _store
+from plugins.rate_limiter.rate_limiter import RateLimiterPlugin, _make_headers, _parse_rate, _select_most_restrictive, _store
 
 
 @pytest.fixture(autouse=True)
@@ -509,3 +509,93 @@ def test_tenant_unlimited_user_tool_limited(self):
         assert allowed is True
         assert limit == 50  # Tool is most restrictive
         assert remaining == 30
+
+
+# ============================================================================
+# _parse_rate Tests
+# ============================================================================
+
+
+class TestParseRate:
+    """Tests for _parse_rate helper covering all time units."""
+
+    def test_seconds_short(self):
+        assert _parse_rate("10/s") == (10, 1)
+
+    def test_seconds_medium(self):
+        assert _parse_rate("10/sec") == (10, 1)
+
+    def test_seconds_long(self):
+        assert _parse_rate("10/second") == (10, 1)
+
+    def test_minutes_short(self):
+        assert _parse_rate("60/m") == (60, 60)
+
+    def test_minutes_medium(self):
+        assert _parse_rate("60/min") == (60, 60)
+
+    def test_minutes_long(self):
+        assert _parse_rate("60/minute") == (60, 60)
+
+    def test_hours_short(self):
+        assert _parse_rate("100/h") == (100, 3600)
+
+    def test_hours_medium(self):
+        assert _parse_rate("100/hr") == (100, 3600)
+
+    def test_hours_long(self):
+        assert _parse_rate("100/hour") == (100, 3600)
+
+    def test_unsupported_unit_raises(self):
+        with pytest.raises(ValueError, match="Unsupported rate unit"):
+            _parse_rate("10/d")
+
+    def test_whitespace_stripped(self):
+        assert _parse_rate("5/ M ") == (5, 60)
+
+
+# ============================================================================
+# Unlimited (no-limit) path tests
+# ============================================================================
+
+
+def _mk_unlimited() -> RateLimiterPlugin:
+    """Create a plugin with no rate limits configured."""
+    return RateLimiterPlugin(
+        PluginConfig(
+            name="rl",
+            kind="plugins.rate_limiter.rate_limiter.RateLimiterPlugin",
+            hooks=[PromptHookType.PROMPT_PRE_FETCH, ToolHookType.TOOL_PRE_INVOKE],
+            config={},  # No limits
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_prompt_pre_fetch_unlimited_returns_no_headers():
+    """When no limits are configured, prompt_pre_fetch returns metadata without http_headers."""
+    plugin = _mk_unlimited()
+    ctx = PluginContext(global_context=GlobalContext(request_id="r1", user="u1"))
+    payload = PromptPrehookPayload(prompt_id="p", args={})
+
+    result = await plugin.prompt_pre_fetch(payload, ctx)
+    assert result.violation is None
+    assert result.http_headers is None
+    assert result.metadata is not None
+    assert result.metadata.get("limited") is False
+
+
+@pytest.mark.asyncio
+async def test_tool_pre_invoke_unlimited_returns_no_headers():
+    """When no limits are configured, tool_pre_invoke returns metadata without http_headers."""
+    from mcpgateway.plugins.framework import ToolPreInvokePayload
+
+    plugin = _mk_unlimited()
+    ctx = PluginContext(global_context=GlobalContext(request_id="r1", user="u1"))
+    payload = ToolPreInvokePayload(name="test_tool", arguments={})
+
+    result = await plugin.tool_pre_invoke(payload, ctx)
+    assert result.violation is None
+    assert result.http_headers is None
+    assert result.metadata is not None
+    assert result.metadata.get("limited") is False