feat(fdc): Add constructor validation and unit tests for Impersonation

Added an explicit __init__ constructor to Impersonation to validate parameter configurations at object instantiation time. Enforced choosing either unauthenticated=True or auth_claims, with support for both auth_claims (snake_case) and authClaims (camelCase). Updated class docstring to recommend factory methods. Also added unit test suite TestImpersonation in tests/test_data_connect.py.

Author: mk2023

firebase_admin/dataconnect.py

@@ -106,7 +106,39 @@ def __post_init__(self):
 
 
 class Impersonation(dict):
-    """Represents impersonation configuration for DataConnect requests."""
+    """Represents impersonation configuration for DataConnect requests.
+
+    It is recommended to construct instances using the static factory methods
+    :meth:`unauthenticated` or :meth:`authenticated`.
+    """
+
+    def __init__(
+        self,
+        *,
+        unauthenticated: Optional[bool] = None,
+        auth_claims: Optional[Dict[str, Any]] = None,
+        authClaims: Optional[Dict[str, Any]] = None
+    ) -> None:
+        if auth_claims is not None and authClaims is not None:
+            raise ValueError("Cannot specify both 'auth_claims' and 'authClaims'.")
+
+        claims = auth_claims if auth_claims is not None else authClaims
+
+        if unauthenticated is None and claims is None:
+            raise ValueError(
+                "Impersonation requires either 'unauthenticated=True' or 'auth_claims'."
+            )
+        if unauthenticated is not None and claims is not None:
+            raise ValueError("Cannot specify both 'unauthenticated' and 'auth_claims'.")
+
+        if unauthenticated is not None:
+            if not isinstance(unauthenticated, bool):
+                raise ValueError("'unauthenticated' must be a boolean.")
+            super().__init__(unauthenticated=unauthenticated)
+        else:
+            if not isinstance(claims, dict):
+                raise ValueError("'auth_claims' must be a dictionary.")
+            super().__init__(authClaims=claims)
 
     @staticmethod
     def unauthenticated() -> 'Impersonation':
@@ -119,7 +151,7 @@ def authenticated(auth_claims: Dict[str, Any]) -> 'Impersonation':
 
         # TODO: More strongly type auth_claims later.
         """
-        return Impersonation(authClaims=auth_claims)
+        return Impersonation(auth_claims=auth_claims)
 
 
 @dataclass
@@ -129,6 +161,7 @@ class GraphqlOptions(Generic[_Variables]):
     impersonate: Optional[Union[Impersonation, Dict[str, Any]]] = None
 
 
+# TODO(b/406281627): Add support for partial errors.
 @dataclass
 class ExecuteGraphqlResponse(Generic[_Data]):
     """Represents the response from a DataConnect GraphQL execution.
@@ -141,6 +174,7 @@ class ExecuteGraphqlResponse(Generic[_Data]):
 
 class DataConnect:
 
+
     """Represents a Firebase Data Connect client instance. 
     
        This client provides access to the Firebase Data Connect service 
@@ -479,11 +513,13 @@ def _parse_graphql_response(
         """Parses a raw GraphQL response payload into ExecuteGraphqlResponse."""
         if not isinstance(resp_dict, dict):
             raise exceptions.InternalError(
-                message="Response payload is not a valid JSON dictionary."
+                message=f"Response payload is not a valid JSON dictionary: {resp_dict}"
             )
 
+        # TODO(b/406281627): Add support for partial errors.
         return ExecuteGraphqlResponse(data=resp_dict.get("data"))
 
+
     def _execute_graphql_helper(
         self,
         query: str,

tests/test_data_connect.py

@@ -883,8 +883,72 @@ def test_parse_graphql_response_non_dict_error(self):
         assert str(excinfo.value) == "Response payload is not a valid JSON dictionary."
 
 
+class TestImpersonation:
+    """Unit tests for Impersonation class and constructor validation."""
+
+    def test_unauthenticated_factory(self):
+        """Tests factory method for unauthenticated impersonation."""
+        imp = dataconnect.Impersonation.unauthenticated()
+        assert imp == {"unauthenticated": True}
+
+    def test_authenticated_factory(self):
+        """Tests factory method for authenticated impersonation."""
+        claims = {"sub": "user_123"}
+        imp = dataconnect.Impersonation.authenticated(claims)
+        assert imp == {"authClaims": {"sub": "user_123"}}
+
+    def test_constructor_unauthenticated(self):
+        """Tests direct constructor with unauthenticated=True."""
+        imp = dataconnect.Impersonation(unauthenticated=True)
+        assert imp == {"unauthenticated": True}
+
+    def test_constructor_auth_claims(self):
+        """Tests direct constructor with auth_claims dict."""
+        claims = {"sub": "user_123"}
+        imp = dataconnect.Impersonation(auth_claims=claims)
+        assert imp == {"authClaims": {"sub": "user_123"}}
+
+    def test_constructor_auth_claims_camel_case(self):
+        """Tests direct constructor with authClaims dict."""
+        claims = {"sub": "user_123"}
+        imp = dataconnect.Impersonation(authClaims=claims)
+        assert imp == {"authClaims": {"sub": "user_123"}}
+
+    def test_constructor_both_claims_and_camel_case_fails(self):
+        """Tests specifying both auth_claims and authClaims raises ValueError."""
+        with pytest.raises(ValueError, match="Cannot specify both 'auth_claims' and 'authClaims'."):
+            dataconnect.Impersonation(auth_claims={"sub": "1"}, authClaims={"sub": "2"})
+
+    def test_constructor_neither_unauth_nor_claims_fails(self):
+        """Tests specifying neither unauthenticated nor claims raises ValueError."""
+        with pytest.raises(
+            ValueError,
+            match="Impersonation requires either 'unauthenticated=True' or 'auth_claims'."
+        ):
+            dataconnect.Impersonation()
+
+    def test_constructor_both_unauth_and_claims_fails(self):
+        """Tests specifying both unauthenticated and claims raises ValueError."""
+        with pytest.raises(
+            ValueError,
+            match="Cannot specify both 'unauthenticated' and 'auth_claims'."
+        ):
+            dataconnect.Impersonation(unauthenticated=True, auth_claims={"sub": "123"})
+
+    def test_constructor_invalid_unauthenticated_type(self):
+        """Tests non-boolean unauthenticated raises ValueError."""
+        with pytest.raises(ValueError, match="'unauthenticated' must be a boolean."):
+            dataconnect.Impersonation(unauthenticated="not-a-bool")
+
+    def test_constructor_invalid_auth_claims_type(self):
+        """Tests non-dict auth_claims raises ValueError."""
+        with pytest.raises(ValueError, match="'auth_claims' must be a dictionary."):
+            dataconnect.Impersonation(auth_claims="not-a-dict")
+
+
 class TestDataConnectExecuteGraphql:
 
+
     def setup_method(self):
         self.cred = testutils.MockCredential()
         self.app = firebase_admin.initialize_app(
@@ -990,9 +1054,10 @@ class User:
             name: str
 
         options = dataconnect.GraphqlOptions(variables={"name": "Fred"})
-        with pytest.raises(ValueError, match="Expected variables of type User"):
+        with pytest.raises(ValueError, match="variables must be of type User"):
             self.api_client.execute_graphql("query { foo }", options=options, variables_type=User)
 
+
     @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request")
 
     def test_execute_graphql_success(self, mock_make_gql_request):