From 18a28c17550cc36a9ebfcdc1c8a774093ee33911 Mon Sep 17 00:00:00 2001 From: mk2023 Date: Mon, 20 Jul 2026 14:09:15 -0700 Subject: [PATCH 1/6] feat(fdc): Add internal GraphQL request helper method and tests Implemented _make_gql_request on _DataConnectApiClient to execute and handle responses/errors for GraphQL operations. Added corresponding unit tests in tests/test_data_connect.py. --- firebase_admin/dataconnect.py | 41 +++++++++++++++++++++++- tests/test_data_connect.py | 60 +++++++++++++++++++++++++++++++++-- 2 files changed, 98 insertions(+), 3 deletions(-) diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index a3ee51a5..6c1ea9f1 100644 --- a/firebase_admin/dataconnect.py +++ b/firebase_admin/dataconnect.py @@ -22,8 +22,11 @@ from dataclasses import dataclass, asdict, is_dataclass import typing from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union + +import requests + import firebase_admin -from firebase_admin import _utils, _http_client, App +from firebase_admin import _utils, _http_client, App, exceptions __all__ = [ 'ConnectorConfig', @@ -334,3 +337,39 @@ def _get_headers(self) -> Dict[str, str]: "X-Firebase-Client": f"fire-admin-python/{firebase_admin.__version__}", "x-goog-api-client": _utils.get_metrics_header(), } + + def _make_gql_request( + self, + url: str, + headers: Dict[str, str], + payload: Dict[str, Any] + ) -> Dict[str, Any]: + """Make a GraphQL request to the Data Connect service.""" + if url is None or headers is None or payload is None: + raise ValueError("url, headers, and payload must all be specified.") + + try: + resp_dict, resp = self._http_client.body_and_response( + 'post', + url=url, + headers=headers, + json=payload + ) + except requests.exceptions.RequestException as error: + raise _utils.handle_platform_error_from_requests(error) + + if resp_dict and "errors" in resp_dict: + errors = resp_dict["errors"] + if isinstance(errors, list) and len(errors) > 0: + messages = [ + err.get("message") for err in errors + if isinstance(err, dict) and err.get("message") + ] + all_messages = " ".join(messages) + raise exceptions.FirebaseError( + code="query-error", + message=all_messages, + http_response=resp + ) + + return resp_dict diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index 9e1faf04..7af5bb35 100644 --- a/tests/test_data_connect.py +++ b/tests/test_data_connect.py @@ -20,10 +20,10 @@ from google.auth import credentials as google_auth_credentials import pytest - +import requests import firebase_admin -from firebase_admin import _utils +from firebase_admin import _utils, _http_client, exceptions from firebase_admin import dataconnect from tests import testutils @@ -724,3 +724,59 @@ def test_get_headers(self): assert isinstance(headers, dict) assert headers.get("X-Firebase-Client") == f"fire-admin-python/{firebase_admin.__version__}" assert headers.get("x-goog-api-client") == _utils.get_metrics_header() + + +class TestDataConnectApiClientMakeGqlRequest: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + @mock.patch.object(_http_client.JsonHttpClient, "body_and_response") + def test_make_gql_request_success(self, mock_body_and_response): + mock_response = mock.Mock(spec=requests.Response) + mock_body_and_response.return_value = ({"data": "val"}, mock_response) + url = "https://example.com/endpoint" + headers = {"key": "val"} + payload = {"query": "foo"} + + res = self.api_client._make_gql_request(url, headers, payload) + assert res == {"data": "val"} + mock_body_and_response.assert_called_once_with( + "post", url=url, headers=headers, json=payload + ) + + def test_make_gql_request_missing_url(self): + headers = {"key": "val"} + payload = {"query": "foo"} + with pytest.raises(ValueError, match="url, headers, and payload must all be specified."): + self.api_client._make_gql_request(None, headers, payload) + + def test_make_gql_request_missing_headers(self): + url = "https://example.com/endpoint" + payload = {"query": "foo"} + with pytest.raises(ValueError, match="url, headers, and payload must all be specified."): + self.api_client._make_gql_request(url, None, payload) + + def test_make_gql_request_missing_payload(self): + url = "https://example.com/endpoint" + headers = {"key": "val"} + with pytest.raises(ValueError, match="url, headers, and payload must all be specified."): + self.api_client._make_gql_request(url, headers, None) + + @mock.patch.object(_http_client.JsonHttpClient, "body_and_response") + def test_make_gql_request_error(self, mock_body_and_response): + mock_body_and_response.side_effect = requests.exceptions.RequestException() + url = "https://example.com/endpoint" + headers = {"key": "val"} + payload = {"query": "foo"} + + with pytest.raises(exceptions.FirebaseError): + self.api_client._make_gql_request(url, headers, payload) From 168105d08093466666bf5c37bd5a61eadbf77680 Mon Sep 17 00:00:00 2001 From: mk2023 Date: Mon, 20 Jul 2026 16:09:22 -0700 Subject: [PATCH 2/6] feat(fdc): Add GraphQL response parsing and deserialization logic Refactored _parse_graphql_response and added robust recursive type deserialization to _DataConnectApiClient. - Implemented _deserialize_type and _deserialize_dataclass helper methods to support nested dataclasses, generic lists (List[T]), generic dictionaries (Dict[K, V]), Unions (Union[...]), Enums, and primitive casting. - Enhanced _make_gql_request error handling to prevent silent error swallowing when the errors key is present. - Added comprehensive unit test coverage in tests/test_data_connect.py. --- firebase_admin/dataconnect.py | 140 ++++++++++++++- tests/test_data_connect.py | 312 +++++++++++++++++++++++++++++++++- 2 files changed, 445 insertions(+), 7 deletions(-) diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index 6c1ea9f1..6be15650 100644 --- a/firebase_admin/dataconnect.py +++ b/firebase_admin/dataconnect.py @@ -20,11 +20,17 @@ from collections.abc import Mapping from dataclasses import dataclass, asdict, is_dataclass +import enum import typing from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union import requests +try: + from types import UnionType # pylint: disable=no-name-in-module +except ImportError: + UnionType = None + import firebase_admin from firebase_admin import _utils, _http_client, App, exceptions @@ -358,18 +364,140 @@ def _make_gql_request( except requests.exceptions.RequestException as error: raise _utils.handle_platform_error_from_requests(error) - if resp_dict and "errors" in resp_dict: + if isinstance(resp_dict, dict) and "errors" in resp_dict: errors = resp_dict["errors"] - if isinstance(errors, list) and len(errors) > 0: + all_messages = "" + if isinstance(errors, list): messages = [ err.get("message") for err in errors if isinstance(err, dict) and err.get("message") ] all_messages = " ".join(messages) - raise exceptions.FirebaseError( - code="query-error", - message=all_messages, - http_response=resp + if not all_messages: + all_messages = ( + f"GraphQL execution failed: {errors}" if errors + else "GraphQL execution failed." ) + raise exceptions.FirebaseError( + code="query-error", + message=all_messages, + http_response=resp + ) return resp_dict + + @staticmethod + def _extract_actual_type(field_type: Any) -> Any: + origin = typing.get_origin(field_type) + if origin is Union or (UnionType is not None and origin is UnionType): + args = typing.get_args(field_type) + non_none_args = [arg for arg in args if arg is not type(None)] + if len(non_none_args) == 1: + return non_none_args[0] + return field_type + + @staticmethod + def _deserialize_type(type_hint: Any, data: Any) -> Any: + """Recursively deserializes data into the specified type hint.""" + if data is None or type_hint is Any: + return data + + actual_type = _DataConnectApiClient._extract_actual_type(type_hint) + + origin = typing.get_origin(actual_type) + args = typing.get_args(actual_type) + + # Union (e.g. Union[int, str] or Union[int, List[str]]) + if origin is Union or (UnionType is not None and origin is UnionType): + for arg in args: + try: + res = _DataConnectApiClient._deserialize_type(arg, data) + base = typing.get_origin(arg) or arg + if base is Any or (isinstance(base, type) and isinstance(res, base)): + return res + except (ValueError, TypeError): + continue + return data + + # Dataclass + if is_dataclass(actual_type): + return _DataConnectApiClient._deserialize_dataclass(actual_type, data) + + # List + if (origin is list or actual_type is list) and isinstance(data, list): + if args: + return [_DataConnectApiClient._deserialize_type(args[0], item) for item in data] + return data + + # Dict / Mapping + if (origin in (dict, Mapping) or actual_type in (dict, Mapping)) and isinstance(data, dict): + if len(args) == 2: + k_type, v_type = args + new_dict = {} + for key, val in data.items(): + try: + coerced_key = k_type(key) if isinstance(k_type, type) else key + except (ValueError, TypeError): + coerced_key = key + new_dict[coerced_key] = _DataConnectApiClient._deserialize_type( + v_type, val + ) + return new_dict + return data + + # Enum (fails loudly on invalid value) + if isinstance(actual_type, type) and issubclass(actual_type, enum.Enum): + try: + return actual_type(data) + except (ValueError, TypeError) as err: + raise ValueError( + f"Invalid value {data!r} for Enum '{actual_type.__name__}'." + ) from err + + # Primitive / Class Constructor + if isinstance(actual_type, type): + try: + return actual_type(data) + except (ValueError, TypeError): + return data + + return data + + @staticmethod + def _deserialize_dataclass(type_hint: Any, data: Any) -> Any: + """Deserializes a dictionary payload into a target dataclass instance.""" + if not is_dataclass(type_hint): + return data + if not isinstance(data, dict): + return data + + type_hints = typing.get_type_hints(type_hint) + fields_to_pass = {} + for field_name, _ in type_hint.__dataclass_fields__.items(): + if field_name in data: + val = data[field_name] + field_type = type_hints.get(field_name) + fields_to_pass[field_name] = _DataConnectApiClient._deserialize_type( + field_type, val + ) + return type_hint(**fields_to_pass) + + @staticmethod + def _parse_graphql_response( + resp_dict: Dict[str, Any], + data_type: Type[_Data] = Any + ) -> ExecuteGraphqlResponse[_Data]: + """Parses a raw GraphQL response payload into ExecuteGraphqlResponse.""" + if not isinstance(resp_dict, dict): + raise exceptions.FirebaseError( + code="internal-error", + message="Response payload is not a valid JSON dictionary." + ) + + data = resp_dict.get("data") + + if data is None: + return ExecuteGraphqlResponse(data=None) + + deserialized_data = _DataConnectApiClient._deserialize_type(data_type, data) + return ExecuteGraphqlResponse(data=deserialized_data) diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index 7af5bb35..408a78d5 100644 --- a/tests/test_data_connect.py +++ b/tests/test_data_connect.py @@ -15,7 +15,8 @@ """Test cases for the firebase_admin.dataconnect module.""" from dataclasses import dataclass -from typing import Any, Dict, Mapping +from enum import Enum +from typing import Any, Dict, List, Mapping, Optional, Union from unittest import mock from google.auth import credentials as google_auth_credentials @@ -780,3 +781,312 @@ def test_make_gql_request_error(self, mock_body_and_response): with pytest.raises(exceptions.FirebaseError): self.api_client._make_gql_request(url, headers, payload) + + @mock.patch.object(_http_client.JsonHttpClient, "body_and_response") + def test_make_gql_request_server_errors(self, mock_body_and_response): + mock_response = mock.Mock(spec=requests.Response) + mock_body_and_response.return_value = ( + { + "errors": [ + {"message": "First error."}, + {"message": "Second error."} + ] + }, + mock_response + ) + url = "https://example.com/endpoint" + headers = {"key": "val"} + payload = {"query": "foo"} + + with pytest.raises(exceptions.FirebaseError) as excinfo: + self.api_client._make_gql_request(url, headers, payload) + + assert excinfo.value.code == "query-error" + assert str(excinfo.value) == "First error. Second error." + assert excinfo.value.http_response is mock_response + + @mock.patch.object(_http_client.JsonHttpClient, "body_and_response") + def test_make_gql_request_non_standard_errors(self, mock_body_and_response): + mock_response = mock.Mock(spec=requests.Response) + mock_body_and_response.return_value = ( + {"errors": "String error message"}, + mock_response + ) + url = "https://example.com/endpoint" + headers = {"key": "val"} + payload = {"query": "foo"} + + with pytest.raises(exceptions.FirebaseError) as excinfo: + self.api_client._make_gql_request(url, headers, payload) + + assert excinfo.value.code == "query-error" + assert str(excinfo.value) == "GraphQL execution failed: String error message" + + mock_body_and_response.return_value = ( + {"errors": []}, + mock_response + ) + with pytest.raises(exceptions.FirebaseError) as excinfo: + self.api_client._make_gql_request(url, headers, payload) + + assert str(excinfo.value) == "GraphQL execution failed." + + +class TestParseGraphqlResponse: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_parse_graphql_response_to_dictionary(self): + payload = { + "data": {"name": "Fred", "age": 20} + } + res = self.api_client._parse_graphql_response(payload, dict) + assert isinstance(res, dataconnect.ExecuteGraphqlResponse) + assert res.data == {"name": "Fred", "age": 20} + + def test_parse_graphql_response_list_of_dataclasses(self): + @dataclass + class User: + name: str + + payload = { + "data": [{"name": "Fred"}, {"name": "Bob"}] + } + res = self.api_client._parse_graphql_response(payload, List[User]) + assert isinstance(res.data, list) + assert len(res.data) == 2 + assert isinstance(res.data[0], User) + assert res.data[0].name == "Fred" + assert isinstance(res.data[1], User) + assert res.data[1].name == "Bob" + + def test_parse_graphql_response_dataclass(self): + @dataclass + class User: + name: str + + payload = { + "data": {"name": "Fred"} + } + res = self.api_client._parse_graphql_response(payload, User) + assert isinstance(res.data, User) + assert res.data.name == "Fred" + + def test_parse_graphql_response_primitive_str(self): + payload = { + "data": "Hello World" + } + res = self.api_client._parse_graphql_response(payload, str) + assert res.data == "Hello World" + + def test_parse_graphql_response_nested_dataclasses(self): + @dataclass + class Profile: + bio: str + + @dataclass + class GetProfileData: + name: str + profile: Profile + + payload = { + "data": { + "name": "Fred", + "profile": {"bio": "Developer"} + } + } + res = self.api_client._parse_graphql_response(payload, GetProfileData) + assert isinstance(res.data, GetProfileData) + assert res.data.name == "Fred" + assert isinstance(res.data.profile, Profile) + assert res.data.profile.bio == "Developer" + + def test_parse_graphql_response_lists_inside_dataclasses(self): + @dataclass + class Comment: + text: str + author: str + + @dataclass + class Post: + title: str + tags: List[str] + comments: List[Comment] + + payload = { + "data": { + "title": "My Post", + "tags": ["tech", "firebase"], + "comments": [ + {"text": "Nice post", "author": "Alice"}, + {"text": "Thanks", "author": "Bob"} + ] + } + } + res = self.api_client._parse_graphql_response(payload, Post) + assert isinstance(res.data, Post) + assert res.data.title == "My Post" + assert res.data.tags == ["tech", "firebase"] + assert isinstance(res.data.comments, list) + assert isinstance(res.data.comments[0], Comment) + assert res.data.comments[0].text == "Nice post" + assert res.data.comments[0].author == "Alice" + + def test_parse_graphql_response_optional_fields(self): + @dataclass + class Profile: + name: str + age: Optional[int] = None + email: Optional[str] = None + + payload = { + "data": { + "name": "Fred", + "age": 20 + } + } + res = self.api_client._parse_graphql_response(payload, Profile) + assert isinstance(res.data, Profile) + assert res.data.name == "Fred" + assert res.data.age == 20 + assert res.data.email is None + + def test_parse_graphql_response_list_of_primitives(self): + payload = { + "data": ["tech", "firebase"] + } + res = self.api_client._parse_graphql_response(payload, List[str]) + assert res.data == ["tech", "firebase"] + + def test_parse_graphql_response_empty_lists_and_null_objects(self): + @dataclass + class User: + name: str + + # Null data + payload1 = { + "data": None + } + res1 = self.api_client._parse_graphql_response(payload1, User) + assert res1.data is None + + # Empty list + payload2 = { + "data": [] + } + res2 = self.api_client._parse_graphql_response(payload2, List[User]) + assert res2.data == [] + + def test_parse_graphql_response_dict_and_any(self): + payload = { + "data": {"name": "Fred", "age": 20} + } + res = self.api_client._parse_graphql_response(payload, Dict[str, Any]) + assert res.data == {"name": "Fred", "age": 20} + + res_any = self.api_client._parse_graphql_response(payload, Any) + assert res_any.data == {"name": "Fred", "age": 20} + + def test_parse_graphql_response_nested_dictionary_field(self): + @dataclass + class AppConfig: + name: str + settings: Dict[str, Any] + + payload = { + "data": { + "name": "MyApp", + "settings": {"theme": "dark", "retries": 3} + } + } + res = self.api_client._parse_graphql_response(payload, AppConfig) + assert isinstance(res.data, AppConfig) + assert res.data.name == "MyApp" + assert res.data.settings == {"theme": "dark", "retries": 3} + + def test_parse_graphql_response_dictionary_key_coercion_to_str(self): + @dataclass + class AppConfig: + name: str + settings: Dict[str, Any] + + payload = { + "data": { + "name": "MyApp", + "settings": {1: "first", 2: "second"} + } + } + res = self.api_client._parse_graphql_response(payload, AppConfig) + assert isinstance(res.data, AppConfig) + assert res.data.name == "MyApp" + assert res.data.settings == {"1": "first", "2": "second"} + + def test_parse_graphql_response_dictionary_key_coercion_to_int(self): + @dataclass + class AppConfig: + name: str + settings: Dict[int, str] + + payload = { + "data": { + "name": "MyApp", + "settings": {"1": "first", "2": "second"} + } + } + res = self.api_client._parse_graphql_response(payload, AppConfig) + assert isinstance(res.data, AppConfig) + assert res.data.name == "MyApp" + assert res.data.settings == {1: "first", 2: "second"} + + def test_parse_graphql_response_non_dict_error(self): + with pytest.raises(exceptions.FirebaseError) as excinfo: + self.api_client._parse_graphql_response("not-a-dict", dict) + + assert excinfo.value.code == "internal-error" + assert str(excinfo.value) == "Response payload is not a valid JSON dictionary." + + def test_parse_graphql_response_union_types(self): + payload_int = { + "data": 123 + } + res_int = self.api_client._parse_graphql_response(payload_int, Union[int, str]) + assert res_int.data == 123 + + payload_str = { + "data": "hello" + } + res_str = self.api_client._parse_graphql_response(payload_str, Union[int, str]) + assert res_str.data == "hello" + + # Also testing generic lists inside Union + payload_list = { + "data": ["foo", "bar"] + } + res_list = self.api_client._parse_graphql_response(payload_list, Union[int, List[str]]) + assert res_list.data == ["foo", "bar"] + + def test_parse_graphql_response_enum(self): + class Status(Enum): + ACTIVE = "ACTIVE" + INACTIVE = "INACTIVE" + + payload = {"data": "ACTIVE"} + res = self.api_client._parse_graphql_response(payload, Status) + assert res.data == Status.ACTIVE + + def test_parse_graphql_response_enum_invalid(self): + class Status(Enum): + ACTIVE = "ACTIVE" + + payload = {"data": "UNKNOWN"} + with pytest.raises(ValueError, match="Invalid value 'UNKNOWN' for Enum 'Status'."): + self.api_client._parse_graphql_response(payload, Status) From d300d30a80815a60d83b63af7ffdfaa09ae09dfc Mon Sep 17 00:00:00 2001 From: mk2023 Date: Thu, 23 Jul 2026 10:29:08 -0700 Subject: [PATCH 3/6] fix(fdc): Add custom QueryError exception and GraphQL error checking helper - Introduced QueryError subclass of FirebaseError for Data Connect GraphQL query/mutation errors and exposed it in __all__. - Extracted _check_graphql_errors helper method on _DataConnectApiClient. - Updated error handling for non-dictionary response payloads in _parse_graphql_response to raise InternalError. - Note: Did not edit parse_graphql_response because we are waiting on whether this will even be a function or not. --- firebase_admin/dataconnect.py | 64 +++++++++++++++++++++++------------ tests/test_data_connect.py | 4 +-- 2 files changed, 44 insertions(+), 24 deletions(-) diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index 6be15650..f208fc6d 100644 --- a/firebase_admin/dataconnect.py +++ b/firebase_admin/dataconnect.py @@ -41,6 +41,7 @@ 'GraphqlOptions', 'Impersonation', 'ExecuteGraphqlResponse', + 'QueryError', ] _DATA_CONNECT_ATTRIBUTE = '_data_connect' @@ -61,6 +62,21 @@ _Data = TypeVar("_Data") _Variables = TypeVar("_Variables") + +# Error Codes +_QUERY_ERROR_CODE = 'query-error' + + +class QueryError(exceptions.FirebaseError): + """Raised when a GraphQL query or mutation execution fails.""" + + def __init__(self, message: str, http_response: Any = None) -> None: + super().__init__( + code=_QUERY_ERROR_CODE, + message=message, + http_response=http_response + ) + @dataclass(frozen=True) class ConnectorConfig: """A configuration object for DataConnect. @@ -344,6 +360,30 @@ def _get_headers(self) -> Dict[str, str]: "x-goog-api-client": _utils.get_metrics_header(), } + @staticmethod + def _check_graphql_errors(resp_dict: Any, resp: Any) -> None: + """Raises QueryError if the GraphQL response payload contains an errors key.""" + if isinstance(resp_dict, dict) and "errors" in resp_dict: + errors = resp_dict["errors"] + all_messages = "" + if isinstance(errors, list): + messages = [] + for err in errors: + if isinstance(err, dict): + message = err.get("message") + if message: + messages.append(message) + all_messages = " ".join(messages) + if not all_messages: + all_messages = ( + f"GraphQL execution failed: {errors}" if errors + else "GraphQL execution failed." + ) + raise QueryError( + message=all_messages, + http_response=resp + ) + def _make_gql_request( self, url: str, @@ -364,26 +404,7 @@ def _make_gql_request( except requests.exceptions.RequestException as error: raise _utils.handle_platform_error_from_requests(error) - if isinstance(resp_dict, dict) and "errors" in resp_dict: - errors = resp_dict["errors"] - all_messages = "" - if isinstance(errors, list): - messages = [ - err.get("message") for err in errors - if isinstance(err, dict) and err.get("message") - ] - all_messages = " ".join(messages) - if not all_messages: - all_messages = ( - f"GraphQL execution failed: {errors}" if errors - else "GraphQL execution failed." - ) - raise exceptions.FirebaseError( - code="query-error", - message=all_messages, - http_response=resp - ) - + _DataConnectApiClient._check_graphql_errors(resp_dict, resp) return resp_dict @staticmethod @@ -489,8 +510,7 @@ def _parse_graphql_response( ) -> ExecuteGraphqlResponse[_Data]: """Parses a raw GraphQL response payload into ExecuteGraphqlResponse.""" if not isinstance(resp_dict, dict): - raise exceptions.FirebaseError( - code="internal-error", + raise exceptions.InternalError( message="Response payload is not a valid JSON dictionary." ) diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index 408a78d5..77024884 100644 --- a/tests/test_data_connect.py +++ b/tests/test_data_connect.py @@ -1048,10 +1048,10 @@ class AppConfig: assert res.data.settings == {1: "first", 2: "second"} def test_parse_graphql_response_non_dict_error(self): - with pytest.raises(exceptions.FirebaseError) as excinfo: + with pytest.raises(exceptions.InternalError) as excinfo: self.api_client._parse_graphql_response("not-a-dict", dict) - assert excinfo.value.code == "internal-error" + assert excinfo.value.code == exceptions.INTERNAL assert str(excinfo.value) == "Response payload is not a valid JSON dictionary." def test_parse_graphql_response_union_types(self): From 28a50e20250ba9f40de7f1c406e441bf7e867686 Mon Sep 17 00:00:00 2001 From: mk2023 Date: Thu, 23 Jul 2026 13:56:15 -0700 Subject: [PATCH 4/6] refactor(fdc): Remove response deserialization and use immediate client instantiation - Removed output deserialization helpers (_extract_actual_type, _deserialize_type, _deserialize_dataclass) to return raw JSON payload dictionaries (ExecuteGraphqlResponse.data), aligning Data Connect with Firestore and Realtime Database patterns for user-defined schemas. - Updated DataConnect.__init__ to immediately instantiate _DataConnectApiClient for consistency with Node.js and other Python Admin SDK services. - Updated test suite in tests/test_data_connect.py to cover raw response parsing and immediate client instantiation. --- firebase_admin/dataconnect.py | 119 +-------------- tests/test_data_connect.py | 275 ++++------------------------------ 2 files changed, 39 insertions(+), 355 deletions(-) diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index f208fc6d..01028b8a 100644 --- a/firebase_admin/dataconnect.py +++ b/firebase_admin/dataconnect.py @@ -20,18 +20,14 @@ from collections.abc import Mapping from dataclasses import dataclass, asdict, is_dataclass -import enum import typing from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union -import requests -try: - from types import UnionType # pylint: disable=no-name-in-module -except ImportError: - UnionType = None +import requests import firebase_admin + from firebase_admin import _utils, _http_client, App, exceptions __all__ = [ @@ -121,6 +117,7 @@ def __init__(self, app: App, config: ConnectorConfig) -> None: """Initializes a DataConnect client instance. """ self._app: App = app self._config = config + self._client = _DataConnectApiClient(connector_config=config, app=app) @property def app(self) -> App: @@ -174,7 +171,6 @@ def client(config: ConnectorConfig, app: Optional[App] = None) -> DataConnect: return dc_service.get_client(config) - class Impersonation(dict): """Represents impersonation configuration for DataConnect requests.""" @@ -407,117 +403,14 @@ def _make_gql_request( _DataConnectApiClient._check_graphql_errors(resp_dict, resp) return resp_dict - @staticmethod - def _extract_actual_type(field_type: Any) -> Any: - origin = typing.get_origin(field_type) - if origin is Union or (UnionType is not None and origin is UnionType): - args = typing.get_args(field_type) - non_none_args = [arg for arg in args if arg is not type(None)] - if len(non_none_args) == 1: - return non_none_args[0] - return field_type - - @staticmethod - def _deserialize_type(type_hint: Any, data: Any) -> Any: - """Recursively deserializes data into the specified type hint.""" - if data is None or type_hint is Any: - return data - - actual_type = _DataConnectApiClient._extract_actual_type(type_hint) - - origin = typing.get_origin(actual_type) - args = typing.get_args(actual_type) - - # Union (e.g. Union[int, str] or Union[int, List[str]]) - if origin is Union or (UnionType is not None and origin is UnionType): - for arg in args: - try: - res = _DataConnectApiClient._deserialize_type(arg, data) - base = typing.get_origin(arg) or arg - if base is Any or (isinstance(base, type) and isinstance(res, base)): - return res - except (ValueError, TypeError): - continue - return data - - # Dataclass - if is_dataclass(actual_type): - return _DataConnectApiClient._deserialize_dataclass(actual_type, data) - - # List - if (origin is list or actual_type is list) and isinstance(data, list): - if args: - return [_DataConnectApiClient._deserialize_type(args[0], item) for item in data] - return data - - # Dict / Mapping - if (origin in (dict, Mapping) or actual_type in (dict, Mapping)) and isinstance(data, dict): - if len(args) == 2: - k_type, v_type = args - new_dict = {} - for key, val in data.items(): - try: - coerced_key = k_type(key) if isinstance(k_type, type) else key - except (ValueError, TypeError): - coerced_key = key - new_dict[coerced_key] = _DataConnectApiClient._deserialize_type( - v_type, val - ) - return new_dict - return data - - # Enum (fails loudly on invalid value) - if isinstance(actual_type, type) and issubclass(actual_type, enum.Enum): - try: - return actual_type(data) - except (ValueError, TypeError) as err: - raise ValueError( - f"Invalid value {data!r} for Enum '{actual_type.__name__}'." - ) from err - - # Primitive / Class Constructor - if isinstance(actual_type, type): - try: - return actual_type(data) - except (ValueError, TypeError): - return data - - return data - - @staticmethod - def _deserialize_dataclass(type_hint: Any, data: Any) -> Any: - """Deserializes a dictionary payload into a target dataclass instance.""" - if not is_dataclass(type_hint): - return data - if not isinstance(data, dict): - return data - - type_hints = typing.get_type_hints(type_hint) - fields_to_pass = {} - for field_name, _ in type_hint.__dataclass_fields__.items(): - if field_name in data: - val = data[field_name] - field_type = type_hints.get(field_name) - fields_to_pass[field_name] = _DataConnectApiClient._deserialize_type( - field_type, val - ) - return type_hint(**fields_to_pass) - @staticmethod def _parse_graphql_response( - resp_dict: Dict[str, Any], - data_type: Type[_Data] = Any - ) -> ExecuteGraphqlResponse[_Data]: + resp_dict: Dict[str, Any] + ) -> ExecuteGraphqlResponse[Any]: """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." ) - data = resp_dict.get("data") - - if data is None: - return ExecuteGraphqlResponse(data=None) - - deserialized_data = _DataConnectApiClient._deserialize_type(data_type, data) - return ExecuteGraphqlResponse(data=deserialized_data) + return ExecuteGraphqlResponse(data=resp_dict.get("data")) diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index 77024884..ffff0518 100644 --- a/tests/test_data_connect.py +++ b/tests/test_data_connect.py @@ -15,10 +15,10 @@ """Test cases for the firebase_admin.dataconnect module.""" from dataclasses import dataclass -from enum import Enum -from typing import Any, Dict, List, Mapping, Optional, Union +from typing import Any, Dict, Mapping from unittest import mock + from google.auth import credentials as google_auth_credentials import pytest import requests @@ -101,7 +101,9 @@ def teardown_method(self, method): def test_init_property_assignment(self): cred = testutils.MockCredential() try: - app = firebase_admin.initialize_app(cred, name="starter_app") + app = firebase_admin.initialize_app( + cred, options={'projectId': 'test-project'}, name="starter_app" + ) except ValueError: pytest.fail("initialize app has an error") @@ -114,9 +116,8 @@ def test_init_property_assignment(self): assert data_connect_instance._config is BASE_CONFIG # pylint: disable=protected-access assert data_connect_instance.app is app assert data_connect_instance.config is BASE_CONFIG + assert isinstance(data_connect_instance._client, dataconnect._DataConnectApiClient) # pylint: disable=protected-access - assert data_connect_instance._app.name == "starter_app" # pylint: disable=protected-access - assert data_connect_instance._config.service_id == "starterproject" # pylint: disable=protected-access class TestDataConnectClientFactory: @@ -127,7 +128,9 @@ def teardown_method(self, method): def setup_method(self): self.cred = testutils.MockCredential() - self.app = firebase_admin.initialize_app(self.cred, name="starter_app") + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'}, name="starter_app" + ) self.config1 = BASE_CONFIG self.config2 = dataconnect.ConnectorConfig( service_id="starterproject2", location="us-east4", connector="my_connector2" @@ -149,7 +152,9 @@ def test_client_successful(self, mock_get_client): assert client2.config is self.config2 def test_client_retrieval_different_apps_same_config(self): - app2 = firebase_admin.initialize_app(self.cred, name="app2") + app2 = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'}, name="app2" + ) client1 = dataconnect.client(self.config1, app=self.app) client2 = dataconnect.client(self.config1, app=app2) @@ -168,7 +173,9 @@ def test_invalid_app_type(self): dataconnect.client(self.config1, "not-a-app") def test_client_default_app(self): - default_app = firebase_admin.initialize_app(self.cred) + default_app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) client_instance = dataconnect.client(self.config1) assert client_instance.app is default_app @@ -192,9 +199,12 @@ class TestDataConnectService: def setup_method(self): self.cred = testutils.MockCredential() - self.app = firebase_admin.initialize_app(self.cred, name="starter_app") + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'}, name="starter_app" + ) self.service = dataconnect._DataConnectService(self.app) # pylint: disable=protected-access + def teardown_method(self, method): del method testutils.cleanup_apps() @@ -298,8 +308,12 @@ class TestDataConnectServiceWorkflow: def setup_method(self): self.cred = testutils.MockCredential() - self.app1 = firebase_admin.initialize_app(self.cred, name="integ_app1") - self.app2 = firebase_admin.initialize_app(self.cred, name="integ_app2") + self.app1 = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'}, name="integ_app1" + ) + self.app2 = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'}, name="integ_app2" + ) self.config1 = BASE_CONFIG self.config2 = dataconnect.ConnectorConfig( @@ -309,6 +323,7 @@ def setup_method(self): service_id="starterproject", location="us-east4", connector="my_connector" ) + def teardown_method(self, method): del method testutils.cleanup_apps() @@ -849,244 +864,20 @@ def test_parse_graphql_response_to_dictionary(self): payload = { "data": {"name": "Fred", "age": 20} } - res = self.api_client._parse_graphql_response(payload, dict) + res = self.api_client._parse_graphql_response(payload) assert isinstance(res, dataconnect.ExecuteGraphqlResponse) assert res.data == {"name": "Fred", "age": 20} - def test_parse_graphql_response_list_of_dataclasses(self): - @dataclass - class User: - name: str - - payload = { - "data": [{"name": "Fred"}, {"name": "Bob"}] - } - res = self.api_client._parse_graphql_response(payload, List[User]) - assert isinstance(res.data, list) - assert len(res.data) == 2 - assert isinstance(res.data[0], User) - assert res.data[0].name == "Fred" - assert isinstance(res.data[1], User) - assert res.data[1].name == "Bob" - - def test_parse_graphql_response_dataclass(self): - @dataclass - class User: - name: str - - payload = { - "data": {"name": "Fred"} - } - res = self.api_client._parse_graphql_response(payload, User) - assert isinstance(res.data, User) - assert res.data.name == "Fred" - - def test_parse_graphql_response_primitive_str(self): - payload = { - "data": "Hello World" - } - res = self.api_client._parse_graphql_response(payload, str) - assert res.data == "Hello World" - - def test_parse_graphql_response_nested_dataclasses(self): - @dataclass - class Profile: - bio: str - - @dataclass - class GetProfileData: - name: str - profile: Profile - - payload = { - "data": { - "name": "Fred", - "profile": {"bio": "Developer"} - } - } - res = self.api_client._parse_graphql_response(payload, GetProfileData) - assert isinstance(res.data, GetProfileData) - assert res.data.name == "Fred" - assert isinstance(res.data.profile, Profile) - assert res.data.profile.bio == "Developer" - - def test_parse_graphql_response_lists_inside_dataclasses(self): - @dataclass - class Comment: - text: str - author: str - @dataclass - class Post: - title: str - tags: List[str] - comments: List[Comment] - - payload = { - "data": { - "title": "My Post", - "tags": ["tech", "firebase"], - "comments": [ - {"text": "Nice post", "author": "Alice"}, - {"text": "Thanks", "author": "Bob"} - ] - } - } - res = self.api_client._parse_graphql_response(payload, Post) - assert isinstance(res.data, Post) - assert res.data.title == "My Post" - assert res.data.tags == ["tech", "firebase"] - assert isinstance(res.data.comments, list) - assert isinstance(res.data.comments[0], Comment) - assert res.data.comments[0].text == "Nice post" - assert res.data.comments[0].author == "Alice" - - def test_parse_graphql_response_optional_fields(self): - @dataclass - class Profile: - name: str - age: Optional[int] = None - email: Optional[str] = None - - payload = { - "data": { - "name": "Fred", - "age": 20 - } - } - res = self.api_client._parse_graphql_response(payload, Profile) - assert isinstance(res.data, Profile) - assert res.data.name == "Fred" - assert res.data.age == 20 - assert res.data.email is None - - def test_parse_graphql_response_list_of_primitives(self): - payload = { - "data": ["tech", "firebase"] - } - res = self.api_client._parse_graphql_response(payload, List[str]) - assert res.data == ["tech", "firebase"] - - def test_parse_graphql_response_empty_lists_and_null_objects(self): - @dataclass - class User: - name: str - - # Null data - payload1 = { - "data": None - } - res1 = self.api_client._parse_graphql_response(payload1, User) - assert res1.data is None - - # Empty list - payload2 = { - "data": [] - } - res2 = self.api_client._parse_graphql_response(payload2, List[User]) - assert res2.data == [] - - def test_parse_graphql_response_dict_and_any(self): - payload = { - "data": {"name": "Fred", "age": 20} - } - res = self.api_client._parse_graphql_response(payload, Dict[str, Any]) - assert res.data == {"name": "Fred", "age": 20} - - res_any = self.api_client._parse_graphql_response(payload, Any) - assert res_any.data == {"name": "Fred", "age": 20} - - def test_parse_graphql_response_nested_dictionary_field(self): - @dataclass - class AppConfig: - name: str - settings: Dict[str, Any] - - payload = { - "data": { - "name": "MyApp", - "settings": {"theme": "dark", "retries": 3} - } - } - res = self.api_client._parse_graphql_response(payload, AppConfig) - assert isinstance(res.data, AppConfig) - assert res.data.name == "MyApp" - assert res.data.settings == {"theme": "dark", "retries": 3} - - def test_parse_graphql_response_dictionary_key_coercion_to_str(self): - @dataclass - class AppConfig: - name: str - settings: Dict[str, Any] - - payload = { - "data": { - "name": "MyApp", - "settings": {1: "first", 2: "second"} - } - } - res = self.api_client._parse_graphql_response(payload, AppConfig) - assert isinstance(res.data, AppConfig) - assert res.data.name == "MyApp" - assert res.data.settings == {"1": "first", "2": "second"} - - def test_parse_graphql_response_dictionary_key_coercion_to_int(self): - @dataclass - class AppConfig: - name: str - settings: Dict[int, str] - - payload = { - "data": { - "name": "MyApp", - "settings": {"1": "first", "2": "second"} - } - } - res = self.api_client._parse_graphql_response(payload, AppConfig) - assert isinstance(res.data, AppConfig) - assert res.data.name == "MyApp" - assert res.data.settings == {1: "first", 2: "second"} + def test_parse_graphql_response_none_data(self): + payload = {"data": None} + res = self.api_client._parse_graphql_response(payload) + assert isinstance(res, dataconnect.ExecuteGraphqlResponse) + assert res.data is None def test_parse_graphql_response_non_dict_error(self): with pytest.raises(exceptions.InternalError) as excinfo: - self.api_client._parse_graphql_response("not-a-dict", dict) + self.api_client._parse_graphql_response("not-a-dict") assert excinfo.value.code == exceptions.INTERNAL assert str(excinfo.value) == "Response payload is not a valid JSON dictionary." - - def test_parse_graphql_response_union_types(self): - payload_int = { - "data": 123 - } - res_int = self.api_client._parse_graphql_response(payload_int, Union[int, str]) - assert res_int.data == 123 - - payload_str = { - "data": "hello" - } - res_str = self.api_client._parse_graphql_response(payload_str, Union[int, str]) - assert res_str.data == "hello" - - # Also testing generic lists inside Union - payload_list = { - "data": ["foo", "bar"] - } - res_list = self.api_client._parse_graphql_response(payload_list, Union[int, List[str]]) - assert res_list.data == ["foo", "bar"] - - def test_parse_graphql_response_enum(self): - class Status(Enum): - ACTIVE = "ACTIVE" - INACTIVE = "INACTIVE" - - payload = {"data": "ACTIVE"} - res = self.api_client._parse_graphql_response(payload, Status) - assert res.data == Status.ACTIVE - - def test_parse_graphql_response_enum_invalid(self): - class Status(Enum): - ACTIVE = "ACTIVE" - - payload = {"data": "UNKNOWN"} - with pytest.raises(ValueError, match="Invalid value 'UNKNOWN' for Enum 'Status'."): - self.api_client._parse_graphql_response(payload, Status) From 1a1d12eacec1a3c79a1e1ab4f5841bd4e5fb7d0c Mon Sep 17 00:00:00 2001 From: mk2023 Date: Mon, 27 Jul 2026 15:16:25 -0700 Subject: [PATCH 5/6] feat(fdc): Add execute_graphql signatures and integration test suite Added execute_graphql and execute_graphql_read method signatures and docstrings to DataConnect and _DataConnectApiClient. Also introduced a comprehensive integration test suite in integration/test_data_connect.py translated from Node.js Admin SDK integration tests. --- firebase_admin/dataconnect.py | 155 +++++++++++--- integration/test_data_connect.py | 337 +++++++++++++++++++++++++++++++ tests/test_data_connect.py | 216 ++++++++++++++++++++ 3 files changed, 679 insertions(+), 29 deletions(-) create mode 100644 integration/test_data_connect.py diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index 01028b8a..f26744fc 100644 --- a/firebase_admin/dataconnect.py +++ b/firebase_admin/dataconnect.py @@ -54,6 +54,9 @@ '/services/{service_id}:{endpoint_id}' ) +_EXECUTE_GRAPHQL_ENDPOINT = 'executeGraphql' +_EXECUTE_GRAPHQL_READ_ENDPOINT = 'executeGraphqlRead' + # Generic Type Parameters _Data = TypeVar("_Data") _Variables = TypeVar("_Variables") @@ -102,7 +105,42 @@ def __post_init__(self): raise ValueError("connector cannot be empty") +class Impersonation(dict): + """Represents impersonation configuration for DataConnect requests.""" + + @staticmethod + def unauthenticated() -> 'Impersonation': + """Returns impersonation configuration for unauthenticated requests.""" + return Impersonation(unauthenticated=True) + + @staticmethod + def authenticated(auth_claims: Dict[str, Any]) -> 'Impersonation': + """Returns impersonation configuration for authenticated requests. + + # TODO: More strongly type auth_claims later. + """ + return Impersonation(authClaims=auth_claims) + + +@dataclass +class GraphqlOptions(Generic[_Variables]): + variables: Optional[_Variables] = None + operation_name: Optional[str] = None + impersonate: Optional[Union[Impersonation, Dict[str, Any]]] = None + + +@dataclass +class ExecuteGraphqlResponse(Generic[_Data]): + """Represents the response from a DataConnect GraphQL execution. + + Attributes: + data: The raw JSON dictionary returned by the GraphQL execution. + """ + data: _Data + + class DataConnect: + """Represents a Firebase Data Connect client instance. This client provides access to the Firebase Data Connect service @@ -127,6 +165,66 @@ def app(self) -> App: def config(self) -> ConnectorConfig: return self._config + def execute_graphql( + self, + query: str, + options: Optional[GraphqlOptions[_Variables]] = None, + variables_type: Type[_Variables] = Any, + ) -> ExecuteGraphqlResponse[Any]: + """Executes a GraphQL query or mutation and returns the result. + + Args: + query: string containing the GraphQL query + options: GraphqlOptions instance containing operational parameters such as + variables, operation name, or impersonation context (optional). + variables_type: The expected structure for the request variables + + Returns: + ExecuteGraphqlResponse: An ExecuteGraphqlResponse containing the raw + response data dictionary. + + Raises: + ValueError: If the arguments are invalid from the local inputs side. + InvalidArgumentError: If GraphQL syntax validation fails on the server. + PermissionDeniedError: If an @auth policy directive blocks execution due to + insufficient permission. + NotFoundError: If a specified resource is not found, or the request is rejected + by undisclosed reasons, such as whitelisting. + InternalError: If the server response payload is invalid or malformed. + FirebaseError: The base platform exception. + """ + raise NotImplementedError + + def execute_graphql_read( + self, + query: str, + options: Optional[GraphqlOptions[_Variables]] = None, + variables_type: Type[_Variables] = Any, + ) -> ExecuteGraphqlResponse[Any]: + """Executes a read-only GraphQL query and returns the result. + + Args: + query: string containing the read-only GraphQL query + options: GraphqlOptions instance containing operational parameters such as + variables, operation name, or impersonation context (optional). + variables_type: The expected structure for the request variables + + Returns: + ExecuteGraphqlResponse: An ExecuteGraphqlResponse containing the raw + response data dictionary. + + Raises: + ValueError: If the arguments are invalid from the local inputs side. + InvalidArgumentError: If GraphQL syntax validation fails on the server. + PermissionDeniedError: If an @auth policy directive blocks execution due to + insufficient permission. + NotFoundError: If a specified resource is not found, or the request is rejected + by undisclosed reasons, such as whitelisting. + InternalError: If the server response payload is invalid or malformed. + FirebaseError: The base platform exception. + """ + raise NotImplementedError + class _DataConnectService: """Service that maintains a collection of DataConnect clients.""" @@ -171,35 +269,6 @@ def client(config: ConnectorConfig, app: Optional[App] = None) -> DataConnect: return dc_service.get_client(config) -class Impersonation(dict): - """Represents impersonation configuration for DataConnect requests.""" - - @staticmethod - def unauthenticated() -> 'Impersonation': - """Returns impersonation configuration for unauthenticated requests.""" - return Impersonation(unauthenticated=True) - - @staticmethod - def authenticated(auth_claims: Dict[str, Any]) -> 'Impersonation': - """Returns impersonation configuration for authenticated requests. - - # TODO: More strongly type auth_claims later. - """ - return Impersonation(authClaims=auth_claims) - - -@dataclass -class GraphqlOptions(Generic[_Variables]): - variables: Optional[_Variables] = None - operation_name: Optional[str] = None - impersonate: Optional[Union[Impersonation, Dict[str, Any]]] = None - - -@dataclass -class ExecuteGraphqlResponse(Generic[_Data]): - data: _Data - - def _get_emulator_host() -> Optional[str]: return _utils.get_emulator_host("DATA_CONNECT_EMULATOR_HOST") @@ -414,3 +483,31 @@ def _parse_graphql_response( ) return ExecuteGraphqlResponse(data=resp_dict.get("data")) + + def _execute_graphql_helper( + self, + query: str, + endpoint: str, + options: Optional[GraphqlOptions[_Variables]] = None, + variables_type: Type[_Variables] = Any, + ) -> ExecuteGraphqlResponse[Any]: + """Helper method to execute GraphQL queries or mutations against a specified endpoint.""" + raise NotImplementedError + + def execute_graphql( + self, + query: str, + options: Optional[GraphqlOptions[_Variables]] = None, + variables_type: Type[_Variables] = Any, + ) -> ExecuteGraphqlResponse[Any]: + """Executes a GraphQL query or mutation and returns the result.""" + raise NotImplementedError + + def execute_graphql_read( + self, + query: str, + options: Optional[GraphqlOptions[_Variables]] = None, + variables_type: Type[_Variables] = Any, + ) -> ExecuteGraphqlResponse[Any]: + """Executes a read-only GraphQL query and returns the result.""" + raise NotImplementedError diff --git a/integration/test_data_connect.py b/integration/test_data_connect.py new file mode 100644 index 00000000..bf681952 --- /dev/null +++ b/integration/test_data_connect.py @@ -0,0 +1,337 @@ +# Copyright 2026 Google Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Integration tests for firebase_admin.dataconnect module (execute_graphql).""" + +import pytest + +from firebase_admin import dataconnect, exceptions + +CONNECTOR_CONFIG = dataconnect.ConnectorConfig( + location='us-west2', + service_id='my-service', + connector='my-connector' +) + +FRED_USER = {'id': 'fred_id', 'address': '32 Elm St.', 'name': 'Fred'} +FREDRICK_USER = { + 'id': FRED_USER['id'], + 'address': '64 Elm St. North', + 'name': 'Fredrick' +} +JEFF_USER = {'id': 'jeff_id', 'address': '99 Oak St.', 'name': 'Jeff'} + +FRED_EMAIL = { + 'id': 'email_id', + 'subject': 'free bitcoin inside', + 'date': '1999-12-31', + 'text': 'get pranked! LOL!', + 'from': {'id': FRED_USER['id']} +} + +INITIAL_STATE = { + 'users': [FRED_USER, JEFF_USER], + 'emails': [FRED_EMAIL] +} + +# Queries & Mutations +QUERY_LIST_USERS = ( + 'query ListUsers @auth(level: PUBLIC) { users { id, name, address } }' +) +QUERY_LIST_EMAILS = ( + 'query ListEmails @auth(level: NO_ACCESS) ' + '{ emails { id subject text date from { id } } }' +) +QUERY_GET_EMAIL = ( + 'query GetEmail($id: String!) @auth(level: NO_ACCESS) ' + '{ email(id: $id) { id subject text date from { id } } }' +) +QUERY_GET_USER_BY_ID = ( + 'query GetUser($id: User_Key!) { user(key: $id) { id name address } }' +) + +QUERY_LIST_USERS_IMPERSONATION = """ + query ListUsers @auth(level: USER) { + users(where: { id: { eq_expr: "auth.uid" } }) { id, name, address } + }""" + +MULTIPLE_QUERIES = f"{QUERY_LIST_USERS}\n{QUERY_LIST_EMAILS}" + +UPSERT_FRED_USER = f""" + mutation user {{ + user_upsert(data: {{id: "{FRED_USER['id']}", address: "{FRED_USER['address']}", name: "{FRED_USER['name']}"}}) + }}""" + +UPDATE_FREDRICK_USER_IMPERSONATED = f""" + mutation upsertFredrickUserImpersonated @auth(level: USER) {{ + user_update( + key: {{ id_expr: "auth.uid" }}, + data: {{ address: "{FREDRICK_USER['address']}", name: "{FREDRICK_USER['name']}" }} + ) + }}""" + +UPSERT_JEFF_USER = f""" + mutation user {{ + user_upsert(data: {{id: "{JEFF_USER['id']}", address: "{JEFF_USER['address']}", name: "{JEFF_USER['name']}"}}) + }}""" + +UPSERT_FRED_EMAIL = f""" + mutation email {{ + email_upsert(data: {{ + id:"{FRED_EMAIL['id']}", + subject: "{FRED_EMAIL['subject']}", + date: "{FRED_EMAIL['date']}", + text: "{FRED_EMAIL['text']}", + fromId: "{FRED_EMAIL['from']['id']}" + }}) + }}""" + +DELETE_ALL = """ + mutation delete { + email_deleteMany(all: true) + user_deleteMany(all: true) + }""" + +# Impersonation Options +OPTS_UNAUTHORIZED_CLAIMS = dataconnect.GraphqlOptions( + impersonate=dataconnect.Impersonation.unauthenticated() +) + +OPTS_AUTHORIZED_FRED_CLAIMS = dataconnect.GraphqlOptions( + impersonate=dataconnect.Impersonation.authenticated({ + 'sub': FRED_USER['id'] + }) +) + +OPTS_NON_EXISTING_CLAIMS = dataconnect.GraphqlOptions( + impersonate=dataconnect.Impersonation.authenticated({ + 'sub': 'non-existing-id', + 'email_verified': True + }) +) + + +@pytest.fixture(autouse=True) +def setup_teardown_db(default_app): + """Sets initial database state before each test and cleans up after.""" + del default_app + dc_client = dataconnect.client(CONNECTOR_CONFIG) + dc_client.execute_graphql(UPSERT_FRED_USER) + dc_client.execute_graphql(UPSERT_JEFF_USER) + dc_client.execute_graphql(UPSERT_FRED_EMAIL) + yield + dc_client.execute_graphql(DELETE_ALL) + + +class TestExecuteGraphql: + """Integration tests for execute_graphql method.""" + + def test_execute_graphql_mutation(self): + """Tests executing mutations via execute_graphql.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + fred_resp = dc_client.execute_graphql(UPSERT_FRED_USER) + assert fred_resp.data['user_upsert']['id'] == FRED_USER['id'] + + jeff_resp = dc_client.execute_graphql(UPSERT_JEFF_USER) + assert jeff_resp.data['user_upsert']['id'] == JEFF_USER['id'] + + upsert_email_resp = dc_client.execute_graphql(UPSERT_FRED_EMAIL) + email_id = upsert_email_resp.data['email_upsert']['id'] + assert email_id + + get_email_options = dataconnect.GraphqlOptions(variables={'id': email_id}) + query_email_resp = dc_client.execute_graphql( + QUERY_GET_EMAIL, options=get_email_options + ) + assert query_email_resp.data['email'] == FRED_EMAIL + + delete_resp = dc_client.execute_graphql(DELETE_ALL) + assert delete_resp.data['email_deleteMany'] == 1 + assert delete_resp.data['user_deleteMany'] == 2 + + def test_execute_graphql_query(self): + """Tests executing a query via execute_graphql.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql(QUERY_LIST_USERS) + assert resp.data['users'] == INITIAL_STATE['users'] + + def test_execute_graphql_operation_name_multiple_queries(self): + """Tests operation_name with multi-query document.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + options = dataconnect.GraphqlOptions(operation_name='ListEmails') + resp = dc_client.execute_graphql(MULTIPLE_QUERIES, options=options) + assert resp.data['emails'] == INITIAL_STATE['emails'] + + def test_execute_graphql_query_error_missing_variables(self): + """Tests query error when required variables are missing.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(dataconnect.QueryError) as excinfo: + dc_client.execute_graphql(QUERY_GET_USER_BY_ID) + assert excinfo.value.code == 'query-error' + + def test_execute_graphql_query_with_variables(self): + """Tests query execution with variables.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + user_id = INITIAL_STATE['users'][0]['id'] + options = dataconnect.GraphqlOptions(variables={'id': {'id': user_id}}) + resp = dc_client.execute_graphql(QUERY_GET_USER_BY_ID, options=options) + assert resp.data['user'] == INITIAL_STATE['users'][0] + + +class TestExecuteGraphqlRead: + """Integration tests for execute_graphql_read method.""" + + def test_execute_graphql_read_query(self): + """Tests read-only query execution.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql_read(QUERY_LIST_USERS) + assert resp.data['users'] == INITIAL_STATE['users'] + + def test_execute_graphql_read_mutation_fails(self): + """Tests that execute_graphql_read rejects mutation queries.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.PermissionDeniedError): + dc_client.execute_graphql_read(UPSERT_FRED_USER) + + +class TestExecuteGraphqlImpersonation: + """Integration tests for execute_graphql / execute_graphql_read impersonation.""" + + class TestUserAuthPolicy: + """Integration tests for @auth(level: USER) policy.""" + + def test_execute_graphql_read_impersonated_authenticated(self): + """Tests read query with authenticated impersonation.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql_read( + QUERY_LIST_USERS_IMPERSONATION, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + assert len(resp.data['users']) == 1 + assert resp.data['users'][0] == FRED_USER + + def test_execute_graphql_read_impersonated_unauthenticated_fails(self): + """Tests read query with unauthenticated impersonation fails.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.UnauthenticatedError): + dc_client.execute_graphql_read( + QUERY_LIST_USERS_IMPERSONATION, options=OPTS_UNAUTHORIZED_CLAIMS + ) + + def test_execute_graphql_impersonated_authenticated(self): + """Tests query with authenticated impersonation.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql( + QUERY_LIST_USERS_IMPERSONATION, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + assert len(resp.data['users']) == 1 + assert resp.data['users'][0] == FRED_USER + + def test_execute_graphql_impersonated_unauthenticated_fails(self): + """Tests query with unauthenticated impersonation fails.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.UnauthenticatedError): + dc_client.execute_graphql( + QUERY_LIST_USERS_IMPERSONATION, options=OPTS_UNAUTHORIZED_CLAIMS + ) + + def test_execute_graphql_impersonated_non_existing_claims(self): + """Tests query with non-existing user claims returns empty list.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql( + QUERY_LIST_USERS_IMPERSONATION, options=OPTS_NON_EXISTING_CLAIMS + ) + assert resp.data['users'] == [] + + def test_execute_graphql_impersonated_mutation_authenticated(self): + """Tests mutation with authenticated impersonation.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + update_resp = dc_client.execute_graphql( + UPDATE_FREDRICK_USER_IMPERSONATED, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + assert update_resp.data['user_update']['id'] == FRED_USER['id'] + + user_id = FRED_USER['id'] + query_options = dataconnect.GraphqlOptions(variables={'id': {'id': user_id}}) + query_resp = dc_client.execute_graphql(QUERY_GET_USER_BY_ID, options=query_options) + assert query_resp.data['user'] == FREDRICK_USER + + def test_execute_graphql_impersonated_mutation_unauthenticated_fails(self): + """Tests mutation with unauthenticated impersonation fails.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.UnauthenticatedError): + dc_client.execute_graphql( + UPDATE_FREDRICK_USER_IMPERSONATED, options=OPTS_UNAUTHORIZED_CLAIMS + ) + + def test_execute_graphql_impersonated_mutation_non_existing_claims(self): + """Tests mutation with non-existing claims returns None.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql( + UPDATE_FREDRICK_USER_IMPERSONATED, options=OPTS_NON_EXISTING_CLAIMS + ) + assert resp.data['user_update'] is None + + class TestPublicAuthPolicy: + """Integration tests for @auth(level: PUBLIC) policy.""" + + def test_impersonated_authenticated(self): + """Tests public query with authenticated claims.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql( + QUERY_LIST_USERS, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + assert resp.data['users'] == INITIAL_STATE['users'] + + def test_impersonated_unauthenticated(self): + """Tests public query with unauthenticated claims.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql( + QUERY_LIST_USERS, options=OPTS_UNAUTHORIZED_CLAIMS + ) + assert resp.data['users'] == INITIAL_STATE['users'] + + def test_impersonated_non_existing_claims(self): + """Tests public query with non-existing user claims.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + resp = dc_client.execute_graphql( + QUERY_LIST_USERS, options=OPTS_NON_EXISTING_CLAIMS + ) + assert resp.data['users'] == INITIAL_STATE['users'] + + class TestNoAccessAuthPolicy: + """Integration tests for @auth(level: NO_ACCESS) policy.""" + + def test_impersonated_authenticated_fails(self): + """Tests no-access query with authenticated claims fails.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.PermissionDeniedError): + dc_client.execute_graphql( + QUERY_LIST_EMAILS, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + + def test_impersonated_unauthenticated_fails(self): + """Tests no-access query with unauthenticated claims fails.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.PermissionDeniedError): + dc_client.execute_graphql( + QUERY_LIST_EMAILS, options=OPTS_UNAUTHORIZED_CLAIMS + ) + + def test_impersonated_non_existing_claims_fails(self): + """Tests no-access query with non-existing user claims fails.""" + dc_client = dataconnect.client(CONNECTOR_CONFIG) + with pytest.raises(exceptions.PermissionDeniedError): + dc_client.execute_graphql( + QUERY_LIST_EMAILS, options=OPTS_NON_EXISTING_CLAIMS + ) diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index ffff0518..f35eeb39 100644 --- a/tests/test_data_connect.py +++ b/tests/test_data_connect.py @@ -881,3 +881,219 @@ def test_parse_graphql_response_non_dict_error(self): assert excinfo.value.code == exceptions.INTERNAL assert str(excinfo.value) == "Response payload is not a valid JSON dictionary." + + +class TestDataConnectExecuteGraphql: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) + self.dc_client = dataconnect.DataConnect(self.app, BASE_CONFIG) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + @mock.patch.object(dataconnect._DataConnectApiClient, "execute_graphql") + def test_execute_graphql_delegates_default(self, mock_execute_graphql): + mock_execute_graphql.return_value = dataconnect.ExecuteGraphqlResponse(data={"key": "val"}) + res = self.dc_client.execute_graphql("query { hello }") + mock_execute_graphql.assert_called_once_with( + query="query { hello }", + options=None, + variables_type=Any + ) + assert res.data == {"key": "val"} + + @mock.patch.object(dataconnect._DataConnectApiClient, "execute_graphql") + def test_execute_graphql_delegates_with_options(self, mock_execute_graphql): + mock_execute_graphql.return_value = dataconnect.ExecuteGraphqlResponse(data={"key": "val"}) + options = dataconnect.GraphqlOptions(variables={"id": "1"}) + res = self.dc_client.execute_graphql( + "query { hello }", options=options, variables_type=dict + ) + mock_execute_graphql.assert_called_once_with( + query="query { hello }", + options=options, + variables_type=dict + ) + assert res.data == {"key": "val"} + + @mock.patch.object(dataconnect._DataConnectApiClient, "execute_graphql_read") + def test_execute_graphql_read_delegates_default(self, mock_execute_graphql_read): + mock_execute_graphql_read.return_value = ( + dataconnect.ExecuteGraphqlResponse(data={"key": "val"}) + ) + res = self.dc_client.execute_graphql_read("query { hello }") + mock_execute_graphql_read.assert_called_once_with( + query="query { hello }", + options=None, + variables_type=Any + ) + assert res.data == {"key": "val"} + + @mock.patch.object(dataconnect._DataConnectApiClient, "execute_graphql_read") + def test_execute_graphql_read_delegates_with_options(self, mock_execute_graphql_read): + mock_execute_graphql_read.return_value = ( + dataconnect.ExecuteGraphqlResponse(data={"key": "val"}) + ) + options = dataconnect.GraphqlOptions(variables={"id": "1"}) + res = self.dc_client.execute_graphql_read( + "query { hello }", options=options, variables_type=dict + ) + mock_execute_graphql_read.assert_called_once_with( + query="query { hello }", + options=options, + variables_type=dict + ) + assert res.data == {"key": "val"} + + +class TestDataConnectApiClientExecuteGraphql: + + def setup_method(self): + self.cred = testutils.MockCredential() + self.app = firebase_admin.initialize_app( + self.cred, options={'projectId': 'test-project'} + ) + self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_execute_graphql_invalid_query_type(self): + with pytest.raises(ValueError, match="query must be a string"): + self.api_client.execute_graphql(123) + + def test_execute_graphql_empty_query(self): + with pytest.raises(ValueError, match="query must be a non-empty string"): + self.api_client.execute_graphql(" ") + + def test_execute_graphql_read_invalid_query_type(self): + with pytest.raises(ValueError, match="query must be a string"): + self.api_client.execute_graphql_read(123) + + def test_execute_graphql_read_empty_query(self): + with pytest.raises(ValueError, match="query must be a non-empty string"): + self.api_client.execute_graphql_read(" ") + + def test_execute_graphql_invalid_options(self): + with pytest.raises(ValueError, match="options must be a GraphqlOptions instance"): + self.api_client.execute_graphql("query { foo }", options="not-graphql-options") + + def test_execute_graphql_invalid_variables_type(self): + @dataclass + class User: + name: str + + options = dataconnect.GraphqlOptions(variables={"name": "Fred"}) + with pytest.raises(ValueError, match="Expected variables 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): + mock_make_gql_request.return_value = {"data": {"foo": "bar"}} + res = self.api_client.execute_graphql("query { foo }") + assert res.data == {"foo": "bar"} + mock_make_gql_request.assert_called_once() + + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_with_dataclass_variables(self, mock_make_gql_request): + @dataclass + class User: + name: str + + mock_make_gql_request.return_value = {"data": {"user": {"name": "Fred"}}} + user_var = User(name="Fred") + options = dataconnect.GraphqlOptions(variables=user_var) + res = self.api_client.execute_graphql( + "query CreateUser { user }", options=options, variables_type=User + ) + assert res.data == {"user": {"name": "Fred"}} + mock_make_gql_request.assert_called_once_with( + url=mock.ANY, + headers=mock.ANY, + payload={ + "query": "query CreateUser { user }", + "variables": {"name": "Fred"} + } + ) + + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_read_success(self, mock_make_gql_request): + mock_make_gql_request.return_value = {"data": {"foo": "bar"}} + res = self.api_client.execute_graphql_read("query { foo }") + assert res.data == {"foo": "bar"} + mock_make_gql_request.assert_called_once() + + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_read_with_dataclass_variables(self, mock_make_gql_request): + @dataclass + class User: + name: str + + mock_make_gql_request.return_value = {"data": {"user": {"name": "Fred"}}} + user_var = User(name="Fred") + options = dataconnect.GraphqlOptions(variables=user_var) + res = self.api_client.execute_graphql_read( + "query GetUser { user }", options=options, variables_type=User + ) + assert res.data == {"user": {"name": "Fred"}} + mock_make_gql_request.assert_called_once_with( + url=mock.ANY, + headers=mock.ANY, + payload={ + "query": "query GetUser { user }", + "variables": {"name": "Fred"} + } + ) + + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_payload_omits_empty_fields(self, mock_make_gql_request): + mock_make_gql_request.return_value = {"data": {"foo": "bar"}} + self.api_client.execute_graphql("query { foo }") + mock_make_gql_request.assert_called_once_with( + url=mock.ANY, + headers=mock.ANY, + payload={"query": "query { foo }"} + ) + + @mock.patch.object(_http_client.JsonHttpClient, "body_and_response") + + def test_execute_graphql_parses_graphql_errors(self, mock_body_and_response): + mock_response = mock.Mock(spec=requests.Response) + mock_body_and_response.return_value = ( + {"errors": [{"message": "Syntax error in GraphQL query"}]}, + mock_response + ) + with pytest.raises(dataconnect.QueryError) as excinfo: + self.api_client.execute_graphql("query { invalid }") + + assert "Syntax error in GraphQL query" in str(excinfo.value) + assert excinfo.value.code == dataconnect._QUERY_ERROR_CODE + + @mock.patch.object(_http_client.JsonHttpClient, "body_and_response") + def test_execute_graphql_malformed_response_payload(self, mock_body_and_response): + mock_response = mock.Mock(spec=requests.Response) + mock_body_and_response.return_value = ("invalid-string-payload", mock_response) + with pytest.raises(exceptions.InternalError) as excinfo: + self.api_client.execute_graphql("query { foo }") + + assert excinfo.value.code == exceptions.INTERNAL + assert str(excinfo.value) == "Response payload is not a valid JSON dictionary." + + @pytest.mark.parametrize("error_class", [ + exceptions.InvalidArgumentError, + exceptions.PermissionDeniedError, + exceptions.NotFoundError, + exceptions.FirebaseError + ]) + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_bubbles_http_exceptions(self, mock_make_gql_request, error_class): + mock_make_gql_request.side_effect = error_class("Mocked API error") + with pytest.raises(error_class, match="Mocked API error"): + self.api_client.execute_graphql("query { foo }") From 03fe60566e73e23978cbc7a83cdff165de2433d0 Mon Sep 17 00:00:00 2001 From: mk2023 Date: Tue, 28 Jul 2026 15:31:12 -0700 Subject: [PATCH 6/6] 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. --- firebase_admin/dataconnect.py | 42 ++++++++++++++++++++-- tests/test_data_connect.py | 67 ++++++++++++++++++++++++++++++++++- 2 files changed, 105 insertions(+), 4 deletions(-) diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py index f26744fc..e7167993 100644 --- a/firebase_admin/dataconnect.py +++ b/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, diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py index f35eeb39..e4464018 100644 --- a/tests/test_data_connect.py +++ b/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):