fix(event_handler): match AppSync Events wildcard routes on segment boundaries

aws_lambda_powertools/event_handler/events_appsync/functions.py

@@ -71,7 +71,7 @@
         """
         Convert a route pattern to a regex pattern with caching.
         Examples:
-            /default/v1/*         -> ^/default/v1/[^/]+$
+            /default/v1/*         -> ^/default/v1/.*$
             /default/v1/users/*   -> ^/default/v1/users/.*$
 
         Parameters
@@ -85,11 +85,13 @@
             Compiled regex pattern
         """
         # Escape special regex chars but convert * to regex pattern
         pattern = re.escape(route).replace("\\*", "[^/]+")
 
-        # If pattern ends with [^/]+, replace with .* for multi-segment match
+        # If pattern ends with [^/]+, replace with .* for multi-segment match.
+        # Only the wildcard is replaced: the preceding "/" must be kept so that
+        # "/default/users/*" does not match "/default/usersettings".
         if pattern.endswith("[^/]+"):
-            pattern = pattern[:-6] + ".*"
+            pattern = pattern[: -len("[^/]+")] + ".*"
 
         # Compile and return the regex pattern
         return re.compile(f"^{pattern}$")

tests/functional/event_handler/required_dependencies/appsync/test_appsync_events_resolvers.py

@@ -1024,6 +1024,37 @@ def phones_handler(payload):
     assert result == expected_result
 
 
+def test_wildcard_route_does_not_match_sibling_segment_with_same_prefix(lambda_context, mock_event):
+    """Test that a wildcard route only matches whole path segments."""
+    # GIVEN a publish event on a channel that only shares a string prefix with a more specific route
+    mock_event["info"]["channel"]["path"] = "/default/usersettings"
+    mock_event["events"] = [
+        {"id": "123", "payload": {"data": "settings data"}},
+    ]
+
+    # GIVEN an AppSyncEventsResolver with a generic and a more specific route
+    app = AppSyncEventsResolver()
+
+    @app.on_publish(path="/default/*")
+    def default_handler(payload):
+        return {"handler": "default", "data": payload["data"]}
+
+    @app.on_publish(path="/default/users/*")
+    def users_handler(payload):
+        return {"handler": "users", "data": payload["data"]}
+
+    # WHEN we resolve the event
+    result = app.resolve(mock_event, lambda_context)
+
+    # THEN the generic route should be used, since /default/usersettings is not under /default/users/
+    expected_result = {
+        "events": [
+            {"id": "123", "payload": {"handler": "default", "data": "settings data"}},
+        ],
+    }
+    assert result == expected_result
+
+
 def test_integration_with_external_service(lambda_context, mock_event):
     """Test integration with an external service using mocks."""
     # GIVEN a sample publish event

tests/unit/event_handler/_required_dependencies/appsync_events/test_functions.py

@@ -0,0 +1,28 @@
+import pytest
+
+from aws_lambda_powertools.event_handler.events_appsync.functions import find_best_route
+
+
+@pytest.mark.parametrize(
+    "path, expected_route",
+    [
+        ("/default/users/123", "/default/users/*"),
+        ("/default/users/123/profile", "/default/users/*"),
+        ("/default/usersettings", "/default/*"),
+        ("/default/users-archive/1", "/default/*"),
+        ("/other/channel", "/*"),
+    ],
+)
+def test_find_best_route_wildcard_matches_whole_segments(path, expected_route):
+    # GIVEN routes where a more specific route shares a string prefix with sibling channels
+    routes = {
+        "/*": {"func": lambda x: x, "aggregate": False},
+        "/default/*": {"func": lambda x: x, "aggregate": False},
+        "/default/users/*": {"func": lambda x: x, "aggregate": False},
+    }
+
+    # WHEN finding the best route
+    result = find_best_route(routes, path)
+
+    # THEN the wildcard only matches whole path segments
+    assert result == expected_route