docs(data_classes): correct AppSync identity docstring for OPENID_CONNECT and pin the dispatch

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py

@@ -56,12 +56,12 @@ def stash(self) -> dict:
     @property
     def identity(self) -> AppSyncIdentityIAM | AppSyncIdentityCognito | None:
         """An object that contains information about the caller.
-        Depending on the type of identify found:
+        Depending on the type of identity found:
         - API_KEY authorization - returns None
         - AWS_IAM authorization - returns AppSyncIdentityIAM
         - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
-        - AWS_LAMBDA authorization - returns None - NEED TO TEST
-        - OPENID_CONNECT authorization - returns None - NEED TO TEST
+        - OPENID_CONNECT authorization - returns AppSyncIdentityCognito
+        - AWS_LAMBDA authorization - unverified, the mapping is whatever the authorizer returns
         """
         return get_identity_object(self.get("identity"))
 

tests/unit/data_classes/required_dependencies/test_appsync_resolver_event.py

@@ -143,3 +143,37 @@ def test_appsync_resolver_event_empty():
 
     assert event.info.field_name is None
     assert event.info.parent_type_name is None
+
+
+def test_get_identity_object_openid_connect_identity():
+    # GIVEN an OPENID_CONNECT identity, which always carries `sub`
+    identity = {
+        "sub": "string",
+        "claims": {"sub": "string"},
+        "defaultAuthStrategy": "ALLOW",
+        "groups": None,
+        "issuer": "https://example.com",
+        "sourceIp": ["string"],
+        "username": "string",
+    }
+
+    # WHEN dispatched to a type
+    identity_object = get_identity_object(identity)
+
+    # THEN it resolves to AppSyncIdentityCognito, which already models the OIDC fields
+    assert isinstance(identity_object, AppSyncIdentityCognito)
+    assert identity_object.sub == identity["sub"]
+    assert identity_object.claims == identity["claims"]
+    assert identity_object.default_auth_strategy == identity["defaultAuthStrategy"]
+    assert identity_object.issuer == identity["issuer"]
+
+
+def test_get_identity_object_only_none_identity_returns_none():
+    # GIVEN any non-null identity, whichever authorization mode produced it
+    # THEN a concrete class is returned
+    assert isinstance(get_identity_object({"sub": "string"}), AppSyncIdentityCognito)
+    assert isinstance(get_identity_object({"accountId": "string"}), AppSyncIdentityIAM)
+    assert isinstance(get_identity_object({}), AppSyncIdentityIAM)
+
+    # AND only an absent identity maps to None
+    assert get_identity_object(None) is None