Fix ruff formatting, add tests for 100% coverage

Author: elijahr

src/mcp/server/events.py

@@ -19,7 +19,6 @@
 from mcp.types import RetainedEvent
 
 
-
 class SubscriptionRegistry:
     """Thread-safe registry mapping session IDs to topic subscription patterns.
 
@@ -48,10 +47,7 @@ async def add(self, session_id: str, pattern: str) -> None:
         """
         segments = pattern.split("/")
         if len(segments) > 8:
-            raise ValueError(
-                f"Topic pattern exceeds maximum depth of 8 segments "
-                f"(got {len(segments)}): {pattern}"
-            )
+            raise ValueError(f"Topic pattern exceeds maximum depth of 8 segments (got {len(segments)}): {pattern}")
         async with self._lock:
             self._subscriptions.setdefault(session_id, set()).add(pattern)
             self._compile(pattern)

src/mcp/types.py

@@ -1463,6 +1463,7 @@ class EventParams(NotificationParams):
     correlationId: str | None = None
     requestedEffects: list[EventEffect] | None = None
     expiresAt: str | None = None
+
     @property
     def event_id(self) -> str:
         return self.eventId

tests/test_event_roundtrip.py

@@ -2,28 +2,25 @@
 
 from __future__ import annotations
 
-import asyncio
 from typing import Any
 
 import anyio
 import pytest
 
 from mcp import types
 from mcp.client.session import ClientSession
-from mcp.server.lowlevel.server import Server, request_ctx
-from mcp.shared.context import RequestContext
 from mcp.server.events import RetainedValueStore, SubscriptionRegistry
 from mcp.server.lowlevel import NotificationOptions
+from mcp.server.lowlevel.server import Server, request_ctx
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
+from mcp.shared.context import RequestContext
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
 from mcp.types import (
-    EventEmitNotification,
     EventListRequest,
     EventListResult,
     EventParams,
-    EventsCapability,
     EventSubscribeParams,
     EventSubscribeRequest,
     EventSubscribeResult,
@@ -33,11 +30,9 @@
     EventUnsubscribeResult,
     RejectedTopic,
     RetainedEvent,
-    ServerCapabilities,
     SubscribedTopic,
 )
 
-
 # Shared registry and store for the test server
 _registry = SubscriptionRegistry()
 _retained_store = RetainedValueStore()
@@ -141,7 +136,7 @@ async def _run_server(server_session: ServerSession, server: Server) -> None:
 @pytest.fixture(autouse=True)
 async def reset_registry():
     """Reset the global registry and store between tests."""
-    global _registry, _retained_store
+    global _registry, _retained_store  # noqa: PLW0603
     _registry = SubscriptionRegistry()
     _retained_store = RetainedValueStore()
     yield
@@ -189,11 +184,14 @@ async def event_handler(params: EventParams):
             sub_result = await client_session.subscribe_events(["test/+"])
             assert len(sub_result.subscribed) == 1
 
-            # Server emits
+            # Server emits with an explicit timestamp, exercising the
+            # branch where emit_event does NOT auto-generate one.
+            explicit_ts = "2025-01-01T00:00:00+00:00"
             await server_session.emit_event(
                 topic="test/hello",
                 payload={"message": "world"},
                 event_id="evt-1",
+                timestamp=explicit_ts,
             )
 
             # Give the notification time to propagate
@@ -203,6 +201,7 @@ async def event_handler(params: EventParams):
             assert received_events[0].topic == "test/hello"
             assert received_events[0].payload == {"message": "world"}
             assert received_events[0].event_id == "evt-1"
+            assert received_events[0].timestamp == explicit_ts
 
             tg.cancel_scope.cancel()
     except (anyio.ClosedResourceError, anyio.EndOfStream):
@@ -502,3 +501,86 @@ async def test_subscribe_rejects_undeclared_topic():
             tg.cancel_scope.cancel()
     except (anyio.ClosedResourceError, anyio.EndOfStream):
         pass
+
+
+@pytest.mark.anyio
+async def test_topic_matches_subscriptions_recompiles_on_cache_miss():
+    """_topic_matches_subscriptions should recompile when the cache entry is missing.
+
+    This exercises the fallback branch where a pattern is in ``_subscribed_patterns``
+    but not in ``_subscription_regex_cache`` (e.g. after manual cache eviction).
+    """
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage](10)
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](10)
+
+    try:
+        async with (
+            server_to_client_send,
+            server_to_client_receive,
+            client_to_server_send,
+            client_to_server_receive,
+            ClientSession(
+                server_to_client_receive,
+                client_to_server_send,
+            ) as client_session,
+        ):
+            # Seed a pattern without populating the regex cache.
+            client_session._subscribed_patterns.add("foo/+")
+            assert "foo/+" not in client_session._subscription_regex_cache
+
+            assert client_session._topic_matches_subscriptions("foo/bar") is True
+            # The cache should now be populated as a side effect.
+            assert "foo/+" in client_session._subscription_regex_cache
+
+            # Non-matching topic exercises the return False path.
+            assert client_session._topic_matches_subscriptions("other/thing") is False
+    except (anyio.ClosedResourceError, anyio.EndOfStream):
+        pass
+
+
+@pytest.mark.anyio
+async def test_subscribe_events_skips_recompile_for_cached_pattern():
+    """subscribe_events should not recompile a regex for an already-cached pattern.
+
+    Covers the branch where ``sub.pattern`` is already present in
+    ``_subscription_regex_cache`` so the compile step is skipped.
+    """
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage](10)
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](10)
+
+    server = _create_test_server()
+    # Reset shared state for isolation.
+    _registry._subscriptions.clear()
+
+    try:
+        async with (
+            ServerSession(
+                client_to_server_receive,
+                server_to_client_send,
+                InitializationOptions(
+                    server_name="test",
+                    server_version="0.1.0",
+                    capabilities=server.get_capabilities(NotificationOptions(), {}),
+                ),
+            ) as server_session,
+            ClientSession(
+                server_to_client_receive,
+                client_to_server_send,
+                message_handler=_message_handler,
+            ) as client_session,
+            anyio.create_task_group() as tg,
+        ):
+            tg.start_soon(_run_server, server_session, server)
+            await client_session.initialize()
+
+            # First subscribe populates the cache.
+            await client_session.subscribe_events(["test/+"])
+            cached_regex = client_session._subscription_regex_cache["test/+"]
+
+            # Second subscribe to the same pattern should reuse the cached compile.
+            await client_session.subscribe_events(["test/+"])
+            assert client_session._subscription_regex_cache["test/+"] is cached_regex
+
+            tg.cancel_scope.cancel()
+    except (anyio.ClosedResourceError, anyio.EndOfStream):
+        pass

tests/test_event_types.py

@@ -7,20 +7,20 @@
 
 import anyio
 import pytest
-
 from pydantic import ValidationError
 
 from mcp import types
 from mcp.client.session import ClientSession
-from mcp.server.lowlevel.server import Server
-from mcp.shared.context import RequestContext
 from mcp.server.events import RetainedValueStore, SubscriptionRegistry
 from mcp.server.lowlevel import NotificationOptions
+from mcp.server.lowlevel.server import Server
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
+from mcp.shared.context import RequestContext
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
 from mcp.types import (
+    ClientRequest,
     EventEffect,
     EventEmitNotification,
     EventListRequest,
@@ -37,10 +37,9 @@
     RejectedTopic,
     RetainedEvent,
     ServerCapabilities,
-    SubscribedTopic,
-    ClientRequest,
     ServerNotification,
     ServerResult,
+    SubscribedTopic,
 )
 
 
@@ -181,9 +180,7 @@ def test_roundtrip_via_root_model(self):
 
 class TestEventSubscribeRequest:
     def test_roundtrip_via_root_model(self):
-        req = EventSubscribeRequest(
-            params=EventSubscribeParams(topics=["a/+", "b/#"])
-        )
+        req = EventSubscribeRequest(params=EventSubscribeParams(topics=["a/+", "b/#"]))
         data = req.model_dump(by_alias=True, mode="json")
         wrapped = ClientRequest.model_validate(data)
         parsed = wrapped.root
@@ -193,9 +190,7 @@ def test_roundtrip_via_root_model(self):
 
 class TestEventUnsubscribeRequest:
     def test_roundtrip_via_root_model(self):
-        req = EventUnsubscribeRequest(
-            params=EventUnsubscribeParams(topics=["a/+"])
-        )
+        req = EventUnsubscribeRequest(params=EventUnsubscribeParams(topics=["a/+"]))
         data = req.model_dump(by_alias=True, mode="json")
         wrapped = ClientRequest.model_validate(data)
         parsed = wrapped.root
@@ -316,15 +311,16 @@ async def _on_unsubscribe_events(
 
 def _create_test_server() -> Server:
     server = Server("test-events-server")
+
     # Register event handlers via request_handlers dict (keyed by type)
     async def subscribe_handler(req: EventSubscribeRequest):
         ctx = server.request_context
-        result = await _on_subscribe_events(ctx, req.root.params if hasattr(req, 'root') else req.params)
+        result = await _on_subscribe_events(ctx, req.root.params if hasattr(req, "root") else req.params)
         return types.ServerResult(result)
 
     async def unsubscribe_handler(req: EventUnsubscribeRequest):
         ctx = server.request_context
-        result = await _on_unsubscribe_events(ctx, req.root.params if hasattr(req, 'root') else req.params)
+        result = await _on_unsubscribe_events(ctx, req.root.params if hasattr(req, "root") else req.params)
         return types.ServerResult(result)
 
     server.request_handlers[EventSubscribeRequest] = subscribe_handler
@@ -350,6 +346,7 @@ async def _run_server(server_session: ServerSession, server: Server) -> None:
                 handler = server.request_handlers.get(type(req.root))
                 if handler:
                     from mcp.server.lowlevel.server import request_ctx
+
                     token = request_ctx.set(
                         RequestContext(
                             request_id=message.request_id,
@@ -368,7 +365,7 @@ async def _run_server(server_session: ServerSession, server: Server) -> None:
 @pytest.fixture(autouse=True)
 def _reset_event_types_registry():
     """Reset the global registry and store between tests."""
-    global _registry, _retained_store
+    global _registry, _retained_store  # noqa: PLW0603
     _registry = SubscriptionRegistry()
     _retained_store = RetainedValueStore()
     yield

tests/test_subscription_registry.py

@@ -183,6 +183,43 @@ async def test_not_expired_returned(self, store: RetainedValueStore):
         await store.set("a/b", event, expires_at=future)
         assert await store.get("a/b") == event
 
+    async def test_get_matching_reuses_cached_regex(self, store: RetainedValueStore):
+        """Second call with same pattern should reuse cached compiled regex."""
+        e1 = RetainedEvent(topic="a/x", eventId="e1", payload="v1")
+        await store.set("a/x", e1)
+        # First call compiles and caches
+        first = await store.get_matching("a/+")
+        assert len(first) == 1
+        # Second call hits the cache branch (skips compile)
+        second = await store.get_matching("a/+")
+        assert len(second) == 1
+        assert second[0].topic == "a/x"
+
+    async def test_invalid_expires_at_treated_as_not_expired(self, store: RetainedValueStore):
+        """Malformed ``expires_at`` should be treated as not expired rather than raising."""
+        event = RetainedEvent(topic="a/b", eventId="e1", payload="val")
+        await store.set("a/b", event, expires_at="not-a-valid-iso-timestamp")
+        # Parsing fails (ValueError), so _is_expired returns False and the value is returned.
+        assert await store.get("a/b") == event
+
+    async def test_naive_expires_at_assumed_utc(self, store: RetainedValueStore):
+        """A naive (tz-less) ISO timestamp should be interpreted as UTC.
+
+        Exercises the ``if expiry.tzinfo is None`` branch in ``_is_expired``.
+        """
+        # Naive timestamp in the future (no timezone suffix).
+        future_naive = (datetime.now(timezone.utc) + timedelta(hours=1)).replace(tzinfo=None).isoformat()
+        event = RetainedEvent(topic="a/b", eventId="e1", payload="val")
+        await store.set("a/b", event, expires_at=future_naive)
+        # Interpreted as UTC -> not expired -> returned.
+        assert await store.get("a/b") == event
+
+        # Naive timestamp in the past -> expired -> None.
+        past_naive = (datetime.now(timezone.utc) - timedelta(hours=1)).replace(tzinfo=None).isoformat()
+        event2 = RetainedEvent(topic="c/d", eventId="e2", payload="val2")
+        await store.set("c/d", event2, expires_at=past_naive)
+        assert await store.get("c/d") is None
+
     async def test_expired_cleaned_on_get_matching(self, store: RetainedValueStore):
         past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
         future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()