diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 32ef50ae..bbcb1d4a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -54,6 +54,10 @@ jobs: run: firebase emulators:exec --only database --project fake-project-id 'pytest integration/test_db.py' - name: Run Functions emulator tests run: firebase emulators:exec --config integration/emulators/firebase.json --only tasks,functions --project fake-project-id 'CLOUD_TASKS_EMULATOR_HOST=localhost:9499 pytest integration/test_functions.py' + - name: Run Data Connect emulator tests + run: firebase emulators:exec --config integration/emulators/firebase.json --only dataconnect --project fake-project-id 'DATA_CONNECT_EMULATOR_HOST=localhost:9399 pytest integration/test_data_connect.py' + + lint: runs-on: ubuntu-latest steps: diff --git a/firebase_admin/_utils.py b/firebase_admin/_utils.py index 0277b9e5..1c7c3337 100644 --- a/firebase_admin/_utils.py +++ b/firebase_admin/_utils.py @@ -15,6 +15,8 @@ """Internal utilities common to all modules.""" import json +import os +import re from platform import python_version from typing import Callable, Optional @@ -345,3 +347,16 @@ def __init__(self): def refresh(self, request): pass + + +def get_emulator_host(env_var_name: str) -> Optional[str]: + """Retrieves and validates the host from the specified emulator environment variable.""" + emulator_host = os.environ.get(env_var_name) + if emulator_host: + if not re.match(r'^(?:\[[a-fA-F0-9:]+\]|[a-zA-Z0-9._-]+):[0-9]+$', emulator_host): + raise ValueError( + f'Invalid {env_var_name}: "{emulator_host}". It must follow format ' + '"host:port".' + ) + return emulator_host + return None diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py new file mode 100644 index 00000000..4a852b31 --- /dev/null +++ b/firebase_admin/dataconnect.py @@ -0,0 +1,574 @@ +# 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. + +"""Firebase Data Connect module. + +This module contains utilities for accessing Firebase Data Connect services associated with +Firebase apps. +""" + +from __future__ import annotations + +from collections.abc import Mapping +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, exceptions + +__all__ = [ + 'ConnectorConfig', + 'DataConnect', + 'client', + 'GraphqlOptions', + 'Impersonation', + 'ExecuteGraphqlResponse', + 'QueryError', +] + +_DATA_CONNECT_ATTRIBUTE = '_data_connect' +_DATA_CONNECT_PROD_URL = 'https://firebasedataconnect.googleapis.com' +_API_VERSION = 'v1' + +_SERVICES_URL_FORMAT = ( + '{host}/{version}/projects/{project_id}/locations/{location_id}' + '/services/{service_id}:{endpoint_id}' +) + +_EMULATOR_SERVICES_URL_FORMAT = ( + 'http://{host}/{version}/projects/{project_id}/locations/{location_id}' + '/services/{service_id}:{endpoint_id}' +) + +_EXECUTE_GRAPHQL_ENDPOINT = 'executeGraphql' +_EXECUTE_GRAPHQL_READ_ENDPOINT = 'executeGraphqlRead' + +# Generic Type Parameters +_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. + + Attributes: + service_id: A string representing the Google Cloud project ID of the service. + location: A string representing the region of the service. + connector: A string representing the name of the connector. + """ + + service_id: str + location: str + connector: str + + def __post_init__(self): + if not isinstance(self.service_id, str): + raise ValueError("service_id must be a string") + if not self.service_id: + raise ValueError("service_id cannot be empty") + if not isinstance(self.location, str): + raise ValueError("location must be a string") + if not self.location: + raise ValueError("location cannot be empty") + if not isinstance(self.connector, str): + raise ValueError("connector must be a string") + if not self.connector: + raise ValueError("connector cannot be empty") + + +class Impersonation(dict): + """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 + ) -> None: + if unauthenticated is None and auth_claims is None: + raise ValueError( + "Impersonation requires either 'unauthenticated=True' or 'auth_claims'." + ) + if unauthenticated is not None and auth_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(auth_claims, dict): + raise ValueError("'auth_claims' must be a dictionary.") + super().__init__(auth_claims=auth_claims) + + @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(auth_claims=auth_claims) + + +@dataclass +class GraphqlOptions(Generic[_Variables]): + variables: Optional[_Variables] = None + operation_name: Optional[str] = None + 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. + + 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 + for a specific Firebase app and connector configuration. + + Attributes: + app: The Firebase App instance for this client. + config: The ConnectorConfig object specifying the service ID, location, and connector name. + """ + + 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: + return self._app + + @property + 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. + """ + return self._client.execute_graphql( + query=query, options=options, variables_type=variables_type + ) + + 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. + """ + return self._client.execute_graphql_read( + query=query, options=options, variables_type=variables_type + ) + + +class _DataConnectService: + """Service that maintains a collection of DataConnect clients.""" + + def __init__(self, app: App) -> None: + self._app: App = app + self._clients: Dict[ConnectorConfig, DataConnect] = {} + + def get_client(self, config: ConnectorConfig) -> DataConnect: + """Creates a client based on the ConnectorConfig. These clients are cached.""" + if not isinstance(config, ConnectorConfig): + raise ValueError("Config must be of type firebase_admin.dataconnect.ConnectorConfig") + if config not in self._clients: + self._clients[config] = DataConnect(app=self._app, config=config) + return self._clients[config] + + +def client(config: ConnectorConfig, app: Optional[App] = None) -> DataConnect: + """Returns a DataConnect client for the specified configuration. + + This function does not make any RPC calls. + + Args: + config: A ConnectorConfig instance specifying the service ID, location, + and connector name. + app: An App instance (optional). Defaults to the default Firebase App. + + Returns: + DataConnect: A handle to the specified DataConnect client instance. + + Raises: + ValueError: If config argument is not an instance of ConnectorConfig, or if + app is an invalid instance of App. + """ + + if not isinstance(config, ConnectorConfig): + raise ValueError("Config must be of type firebase_admin.dataconnect.ConnectorConfig") + + # must check whether app has a _DataConnectService attached to it yet + dc_service = _utils.get_app_service(app, _DATA_CONNECT_ATTRIBUTE, _DataConnectService) + + return dc_service.get_client(config) + + +def _get_emulator_host() -> Optional[str]: + return _utils.get_emulator_host("DATA_CONNECT_EMULATOR_HOST") + + +class _DataConnectApiClient: + """Internal client for sending requests to the Firebase Data Connect backend. + + Attributes: + connector_config: The connector configuration specifying the service, + location, and connector name. + app: The Firebase App instance associated with this client. + """ + + def __init__(self, connector_config: ConnectorConfig, app: App) -> None: + if not isinstance(app, App): + raise ValueError( + 'Second argument passed to DataConnectApiClient must be a valid ' + 'Firebase app instance.' + ) + self._connector_config = connector_config + self._app = app + + self._project_id = app.project_id + if not self._project_id: + raise ValueError( + 'Failed to determine project ID. Initialize the SDK with service ' + 'account credentials or set project ID as an app option. Alternatively, set the ' + 'GOOGLE_CLOUD_PROJECT environment variable.') + + self._emulator_host = _get_emulator_host() + if self._emulator_host: + self._credential = _utils.EmulatorAdminCredentials() + else: + self._credential = app.credential.get_credential() + + self._http_client = _http_client.JsonHttpClient(credential=self._credential) + + def _validate_variables_type( + self, + variables: Any, + variable_type: Optional[Type[Any]] = None + ) -> None: + """Validates variables against expected type.""" + if variables is not None: + if not (isinstance(variables, Mapping) or is_dataclass(variables)): + raise ValueError("variables must be a collections.abc.Mapping or a dataclass") + if ( + variable_type is not None + and variable_type is not Any + and variable_type is not typing.Any + ): + expected_type = typing.get_origin(variable_type) or variable_type + if not isinstance(variables, expected_type): + type_name = getattr(expected_type, '__name__', str(expected_type)) + raise ValueError(f"variables must be of type {type_name}") + + def _validate_impersonation_options(self, impersonate: Any) -> None: + """Validates impersonation dictionary options.""" + if impersonate is not None: + if not isinstance(impersonate, dict): + raise ValueError('impersonate option must be a dictionary') + if 'unauthenticated' not in impersonate and 'auth_claims' not in impersonate: + raise ValueError( + "impersonate option must contain either " + "'unauthenticated' or 'auth_claims'" + ) + if 'unauthenticated' in impersonate and 'auth_claims' in impersonate: + raise ValueError( + "impersonate option cannot contain both " + "'unauthenticated' and 'auth_claims'" + ) + if 'unauthenticated' in impersonate: + if not isinstance(impersonate['unauthenticated'], bool): + raise ValueError("'unauthenticated' claim must be a boolean") + if 'auth_claims' in impersonate: + if not isinstance(impersonate['auth_claims'], dict): + raise ValueError("'auth_claims' claim must be a dictionary") + + def _validate_graphql_options( + self, + graphql_options: Optional[GraphqlOptions[Any]], + variable_type: Optional[Type[Any]] = None + ) -> None: + """Validates GraphqlOptions inputs at runtime.""" + if graphql_options is not None: + if not isinstance(graphql_options, GraphqlOptions): + raise ValueError('options must be a GraphqlOptions instance') + + # Validate Variables against expected variable_type + self._validate_variables_type(graphql_options.variables, variable_type) + + # Validate Operation Name (if it exists) + operation_name = graphql_options.operation_name + if operation_name is not None: + if not isinstance(operation_name, str): + raise ValueError('operation_name must be a string') + if not operation_name.strip(): + raise ValueError('operation_name must be a non-empty string') + + # Validate Impersonation (if it exists) + self._validate_impersonation_options(graphql_options.impersonate) + + def _prepare_graphql_payload( + self, + graphql_query: str, + graphql_options: Optional[GraphqlOptions[_Variables]] + ) -> Dict[str, Any]: + """Serializes input query and options to JSON-compatible dictionary.""" + payload = { + "query": graphql_query + } + + if graphql_options is not None: + if graphql_options.variables is not None: + if is_dataclass(graphql_options.variables): + payload["variables"] = asdict(graphql_options.variables) + else: + payload["variables"] = graphql_options.variables + + if graphql_options.operation_name is not None: + payload["operationName"] = graphql_options.operation_name.strip() + + if graphql_options.impersonate is not None: + impersonate_payload = dict(graphql_options.impersonate) + if "auth_claims" in impersonate_payload: + impersonate_payload["authClaims"] = impersonate_payload.pop("auth_claims") + payload["extensions"] = { + "impersonate": impersonate_payload + } + + return payload + + def _get_firebase_dataconnect_service_url(self, method_name: str) -> str: + """Build and return the URL for a Firebase Data Connect API method.""" + project_id = self._project_id + location = self._connector_config.location + service_id = self._connector_config.service_id + + if self._emulator_host: + return _EMULATOR_SERVICES_URL_FORMAT.format( + host=self._emulator_host, + version=_API_VERSION, + project_id=project_id, + location_id=location, + service_id=service_id, + endpoint_id=method_name + ) + return _SERVICES_URL_FORMAT.format( + host=_DATA_CONNECT_PROD_URL, + version=_API_VERSION, + project_id=project_id, + location_id=location, + service_id=service_id, + endpoint_id=method_name + ) + + def _get_headers(self) -> Dict[str, str]: + """Build and return the headers for a Firebase Data Connect API call.""" + return { + "X-Firebase-Client": f"fire-admin-python/{firebase_admin.__version__}", + "x-goog-api-client": _utils.get_metrics_header(), + "X-Client-Version": f"Python/Admin/{firebase_admin.__version__}", + "X-Firebase-Sqlconnect-Affinity": ( + f"{self._project_id}{self._connector_config.service_id}" + ), + } + + @staticmethod + def _check_graphql_errors(resp_dict: Any, resp: Any) -> None: + """Raises QueryError if the GraphQL response payload contains non-empty errors.""" + if isinstance(resp_dict, dict) and resp_dict.get("errors"): + 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, + 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) + + _DataConnectApiClient._check_graphql_errors(resp_dict, resp) + return resp_dict + + @staticmethod + def _parse_graphql_response( + 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=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, + 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.""" + if not isinstance(query, str): + raise ValueError("query must be a string") + query = query.strip() + if not query: + raise ValueError("query must be a non-empty string") + + self._validate_graphql_options(options, variable_type=variables_type) + + url = self._get_firebase_dataconnect_service_url(endpoint) + headers = self._get_headers() + payload = self._prepare_graphql_payload(query, options) + + resp_dict = self._make_gql_request(url=url, headers=headers, payload=payload) + return self._parse_graphql_response(resp_dict) + + 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.""" + return self._execute_graphql_helper( + query, _EXECUTE_GRAPHQL_ENDPOINT, options, variables_type + ) + + 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.""" + return self._execute_graphql_helper( + query, _EXECUTE_GRAPHQL_READ_ENDPOINT, options, variables_type + ) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index 66ba700b..b111d88f 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -18,7 +18,6 @@ from datetime import datetime, timedelta, timezone from urllib import parse import re -import os import json from base64 import b64encode from typing import Any, Optional, Dict @@ -62,14 +61,7 @@ _DEFAULT_LOCATION = 'us-central1' def _get_emulator_host() -> Optional[str]: - emulator_host = os.environ.get(_EMULATOR_HOST_ENV_VAR) - if emulator_host: - if '//' in emulator_host: - raise ValueError( - f'Invalid {_EMULATOR_HOST_ENV_VAR}: "{emulator_host}". It must follow format ' - '"host:port".') - return emulator_host - return None + return _utils.get_emulator_host(_EMULATOR_HOST_ENV_VAR) def _get_functions_service(app) -> _FunctionsService: diff --git a/integration/emulators/dataconnect/connector/connector.yaml b/integration/emulators/dataconnect/connector/connector.yaml new file mode 100644 index 00000000..3b1bcdcc --- /dev/null +++ b/integration/emulators/dataconnect/connector/connector.yaml @@ -0,0 +1 @@ +connectorId: "my-connector" diff --git a/integration/emulators/dataconnect/connector/mutations.gql b/integration/emulators/dataconnect/connector/mutations.gql new file mode 100644 index 00000000..f3bfbeeb --- /dev/null +++ b/integration/emulators/dataconnect/connector/mutations.gql @@ -0,0 +1,102 @@ +mutation upsertFredUser @auth(level: NO_ACCESS) { + user_upsert(data: { id: "fred_id", address: "32 Elm St.", name: "Fred" }) +} +mutation updateFredrickUserImpersonation +@auth(level: USER, insecureReason: "test") { + user_update( + key: { id_expr: "auth.uid" } + data: { address: "64 Elm St. North", name: "Fredrick" } + ) +} +mutation upsertJeffUser @auth(level: NO_ACCESS) { + user_upsert(data: { id: "jeff_id", address: "99 Oak St.", name: "Jeff" }) +} + +mutation upsertJeffEmail @auth(level: NO_ACCESS) { + email_upsert( + data: { + id: "jeff_email_id" + subject: "free bitcoin inside" + date: "1999-12-31" + text: "get pranked! LOL!" + fromId: "jeff_id" + } + ) +} + +mutation InsertUser($id: String!, $name: String!, $address: String!) +@auth(level: PUBLIC, insecureReason: "test") { + user_insert(data: { id: $id, name: $name, address: $address }) +} + +mutation InsertEmailPublic($id: String!) +@auth(level: PUBLIC, insecureReason: "test") { + email_insert( + data: { + id: $id + subject: "PublicEmail" + date: "1999-12-31" + text: "PublicEmail" + fromId: "jeff_id" + } + ) +} +mutation InsertEmailUserAnon($id: String!) +@auth(level: USER_ANON, insecureReason: "test") { + email_insert( + data: { + id: $id + subject: "UserAnonEmail" + date: "1999-12-31" + text: "UserAnonEmail" + fromId: "jeff_id" + } + ) +} +mutation InsertEmailUser($id: String!) +@auth(level: USER, insecureReason: "test") { + email_insert( + data: { + id: $id + subject: "UserEmail" + date: "1999-12-31" + text: "UserEmail" + fromId: "jeff_id" + } + ) +} +mutation InsertEmailUserEmailVerified($id: String!) +@auth(level: USER_EMAIL_VERIFIED, insecureReason: "test") { + email_insert( + data: { + id: $id + subject: "UserEmailVerifiedEmail" + date: "1999-12-31" + text: "UserEmailVerifiedEmail" + fromId: "jeff_id" + } + ) +} +mutation InsertEmailNoAccess($id: String!) @auth(level: NO_ACCESS) { + email_insert( + data: { + id: $id + subject: "NoAccessEmail" + date: "1999-12-31" + text: "NoAccessEmail" + fromId: "jeff_id" + } + ) +} +mutation InsertEmailImpersonation($id: String!) +@auth(level: USER_ANON, insecureReason: "test") { + email_insert( + data: { + id: $id + subject: "ImpersonatedEmail" + date: "1999-12-31" + text: "ImpersonatedEmail" + fromId_expr: "auth.uid" + } + ) +} diff --git a/integration/emulators/dataconnect/connector/queries.gql b/integration/emulators/dataconnect/connector/queries.gql new file mode 100644 index 00000000..b4c3c922 --- /dev/null +++ b/integration/emulators/dataconnect/connector/queries.gql @@ -0,0 +1,75 @@ +query ListUsersPublic @auth(level: PUBLIC, insecureReason: "test") { + users { + id + name + address + } +} +query ListUsersUserAnon @auth(level: USER_ANON, insecureReason: "test") { + users { + id + name + address + } +} +query ListUsersUser @auth(level: USER, insecureReason: "test") { + users { + id + name + address + } +} +query ListUsersUserEmailVerified +@auth(level: USER_EMAIL_VERIFIED, insecureReason: "test") { + users { + id + name + address + } +} +query ListUsersNoAccess @auth(level: NO_ACCESS) { + users { + id + name + address + } +} +query ListUsersImpersonationAnon @auth(level: USER_ANON) { + users(where: { id: { eq_expr: "auth.uid" } }) { + id + name + address + } +} +query GetUser($id: User_Key!) @auth(level: PUBLIC, insecureReason: "test") { + user(key: $id) { + id + name + address + } +} + +query ListEmails @auth(level: NO_ACCESS) { + emails { + id + subject + text + date + from { + name + } + } +} +query GetEmail($id: String!) @auth(level: USER_ANON, insecureReason: "test") { + email(id: $id) { + id + subject + date + text + from { + id + name + address + } + } +} diff --git a/integration/emulators/dataconnect/dataconnect.yaml b/integration/emulators/dataconnect/dataconnect.yaml new file mode 100644 index 00000000..562964d8 --- /dev/null +++ b/integration/emulators/dataconnect/dataconnect.yaml @@ -0,0 +1,12 @@ +specVersion: "v1" +serviceId: "my-service" +location: "us-west2" +schema: + source: "./schema" + datasource: + postgresql: + database: "my-database" + cloudSql: + instanceId: "my-instance" +connectorDirs: + - "./connector" \ No newline at end of file diff --git a/integration/emulators/dataconnect/schema/schema.gql b/integration/emulators/dataconnect/schema/schema.gql new file mode 100644 index 00000000..405d593f --- /dev/null +++ b/integration/emulators/dataconnect/schema/schema.gql @@ -0,0 +1,13 @@ +type User @table(key: ["id"]) { + id: String! + name: String! + address: String! +} + +type Email @table { + id: String! + subject: String! + date: Date! + text: String! + from: User! +} diff --git a/integration/emulators/firebase.json b/integration/emulators/firebase.json index a7b727c4..7c2bff8a 100644 --- a/integration/emulators/firebase.json +++ b/integration/emulators/firebase.json @@ -1,5 +1,8 @@ { "emulators": { + "dataconnect": { + "port": 9399 + }, "tasks": { "port": 9499 }, @@ -11,6 +14,9 @@ "port": 5001 } }, + "dataconnect": { + "source": "dataconnect" + }, "functions": [ { "source": "functions", diff --git a/integration/test_data_connect.py b/integration/test_data_connect.py new file mode 100644 index 00000000..914ee910 --- /dev/null +++ b/integration/test_data_connect.py @@ -0,0 +1,358 @@ +# 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 os +import pytest + +import firebase_admin +from firebase_admin import _utils, dataconnect, exceptions +from integration import conftest + +def integration_conf(request): + host_override = os.environ.get('DATA_CONNECT_EMULATOR_HOST') + if host_override: + return _utils.EmulatorAdminCredentials(), 'fake-project-id' + + return conftest.integration_conf(request) + +@pytest.fixture(scope='module') +def app(request): + cred, project_id = integration_conf(request) + return firebase_admin.initialize_app( + cred, options={'projectId': project_id}, name='integration-dataconnect') + +@pytest.fixture(scope='module', autouse=True) +def default_app(): + # Overwrites the default_app fixture in conftest.py. + # This test suite should not use the default app. Use the app fixture instead. + pass + +@pytest.fixture +def dc_client(app): + return dataconnect.client(CONNECTOR_CONFIG, app=app) + +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']} +} + +UPDATED_FRED_EMAIL = { + 'id': FRED_EMAIL['id'], + 'subject': 'updated subject', + 'date': '2026-07-31', + 'text': 'updated body text!', + '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']}" + }}) + }}""" + +UPSERT_UPDATED_FRED_EMAIL = f""" + mutation email {{ + email_upsert(data: {{ + id:"{UPDATED_FRED_EMAIL['id']}", + subject: "{UPDATED_FRED_EMAIL['subject']}", + date: "{UPDATED_FRED_EMAIL['date']}", + text: "{UPDATED_FRED_EMAIL['text']}", + fromId: "{UPDATED_FRED_EMAIL['from']['id']}" + }}) + }}""" + +DELETE_ALL = """ + mutation delete { + email_deleteMany(all: true) + user_deleteMany(all: true) + }""" + +@pytest.fixture(autouse=True) +def setup_and_cleanup_database(dc_client): + """Initializes database via DataConnect client before each test and wipes it afterwards.""" + 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) + +# 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 + }) +) + +class TestExecuteGraphql: + """Integration tests for execute_graphql method.""" + + def test_execute_graphql_query(self, dc_client): + """Tests executing a query via execute_graphql.""" + resp = dc_client.execute_graphql(QUERY_LIST_USERS) + assert sorted(resp.data['users'], key=lambda user: user['id']) == sorted( + INITIAL_STATE['users'], key=lambda user: user['id'] + ) + + def test_execute_graphql_query_with_variables(self, dc_client): + """Tests query execution with variables.""" + 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] + + def test_execute_graphql_operation_name_multiple_queries(self, dc_client): + """Tests operation_name with multi-query document.""" + 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, dc_client): + """Tests query error when required variables are missing.""" + 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_mutation(self, dc_client): + """Tests executing mutations via execute_graphql.""" + 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_UPDATED_FRED_EMAIL) + email_id = upsert_email_resp.data['email_upsert']['id'] + assert email_id == UPDATED_FRED_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'] == UPDATED_FRED_EMAIL + + +class TestExecuteGraphqlRead: + """Integration tests for execute_graphql_read method.""" + + def test_execute_graphql_read_query(self, dc_client): + """Tests read-only query execution.""" + resp = dc_client.execute_graphql_read(QUERY_LIST_USERS) + assert sorted(resp.data['users'], key=lambda user: user['id']) == sorted( + INITIAL_STATE['users'], key=lambda user: user['id'] + ) + + def test_execute_graphql_read_mutation_fails(self, dc_client): + """Tests that execute_graphql_read rejects mutation queries.""" + 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, dc_client): + """Tests read query with authenticated impersonation.""" + 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_impersonated_authenticated(self, dc_client): + """Tests query with authenticated impersonation.""" + 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, dc_client): + """Tests query with unauthenticated impersonation fails.""" + 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, dc_client): + """Tests query with non-existing user claims returns empty list.""" + 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, dc_client): + """Tests mutation with authenticated impersonation.""" + 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, dc_client): + """Tests mutation with unauthenticated impersonation fails.""" + 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, dc_client): + """Tests mutation with non-existing claims returns None.""" + 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, dc_client): + """Tests public query with authenticated claims.""" + resp = dc_client.execute_graphql( + QUERY_LIST_USERS, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + assert sorted(resp.data['users'], key=lambda user: user['id']) == sorted( + INITIAL_STATE['users'], key=lambda user: user['id'] + ) + + def test_impersonated_unauthenticated(self, dc_client): + """Tests public query with unauthenticated claims.""" + resp = dc_client.execute_graphql( + QUERY_LIST_USERS, options=OPTS_UNAUTHORIZED_CLAIMS + ) + assert sorted(resp.data['users'], key=lambda user: user['id']) == sorted( + INITIAL_STATE['users'], key=lambda user: user['id'] + ) + + def test_impersonated_non_existing_claims(self, dc_client): + """Tests public query with non-existing user claims.""" + resp = dc_client.execute_graphql( + QUERY_LIST_USERS, options=OPTS_NON_EXISTING_CLAIMS + ) + assert sorted(resp.data['users'], key=lambda user: user['id']) == sorted( + INITIAL_STATE['users'], key=lambda user: user['id'] + ) + + + class TestNoAccessAuthPolicy: + """Integration tests for @auth(level: NO_ACCESS) policy.""" + + def test_impersonated_authenticated_fails(self, dc_client): + """Tests no-access query with authenticated claims fails.""" + with pytest.raises(exceptions.PermissionDeniedError): + dc_client.execute_graphql( + QUERY_LIST_EMAILS, options=OPTS_AUTHORIZED_FRED_CLAIMS + ) + + def test_impersonated_unauthenticated_fails(self, dc_client): + """Tests no-access query with unauthenticated claims fails.""" + with pytest.raises(exceptions.PermissionDeniedError): + dc_client.execute_graphql( + QUERY_LIST_EMAILS, options=OPTS_UNAUTHORIZED_CLAIMS + ) + + def test_impersonated_non_existing_claims_fails(self, dc_client): + """Tests no-access query with non-existing user claims fails.""" + 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 new file mode 100644 index 00000000..3facb50b --- /dev/null +++ b/tests/test_data_connect.py @@ -0,0 +1,1013 @@ +# 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. + +"""Test cases for the firebase_admin.dataconnect module.""" + +from dataclasses import dataclass +from typing import Any, Dict, Mapping +from unittest import mock + +from google.auth import credentials as google_auth_credentials +import pytest +import requests + +import firebase_admin +from firebase_admin import _utils, _http_client, exceptions +from firebase_admin import dataconnect +from tests import testutils + +BASE_CONFIG = dataconnect.ConnectorConfig( + service_id="starterproject", + location="us-east4", + connector="my_connector", +) + +TEST_QUERY = "query { hello }" +TEST_RESPONSE_DATA = {"foo": "bar"} +TEST_URL = "https://example.com/endpoint" +TEST_HEADERS = {"key": "val"} +TEST_PAYLOAD = {"query": TEST_QUERY} +TEST_AUTH_CLAIMS = {"sub": "user_123"} +TEST_VARIABLES = {"var_key": "var_val"} + +@dataclass +class UserProfile: + address: str + phone: str + +@dataclass +class CreateUserVariables: + user_id: str + name: str + profile: UserProfile + +TEST_PROFILE = UserProfile(address="123 Road", phone="332-3233-0199") +TEST_DATACLASS_VARIABLES = CreateUserVariables( + user_id="1", name="Fred", profile=TEST_PROFILE +) + +@dataclass +class User: + name: str + +class TestConnectorConfig: + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_connector_config_initialization(self): + assert BASE_CONFIG.service_id == "starterproject" + assert BASE_CONFIG.location == "us-east4" + assert BASE_CONFIG.connector == "my_connector" + + def test_connector_config_is_frozen(self): + with pytest.raises(AttributeError, match="cannot assign to field 'service_id'"): + BASE_CONFIG.service_id = "changed_id" + with pytest.raises(AttributeError, match="cannot assign to field 'location'"): + BASE_CONFIG.location = "us-central1" + with pytest.raises(AttributeError, match="cannot assign to field 'connector'"): + BASE_CONFIG.connector = "changed_connector" + + def test_connector_config_string_written(self): + repr_str = repr(BASE_CONFIG) + assert "service_id='starterproject'" in repr_str + assert "location='us-east4'" in repr_str + assert "connector='my_connector'" in repr_str + + def test_connector_config_empty_strings(self): + with pytest.raises(ValueError, match="service_id cannot be empty"): + dataconnect.ConnectorConfig( + service_id="", location="us-east4", connector="my_connector" + ) + + with pytest.raises(ValueError, match="location cannot be empty"): + dataconnect.ConnectorConfig( + service_id="starterproject", location="", connector="my_connector" + ) + + with pytest.raises(ValueError, match="connector cannot be empty"): + dataconnect.ConnectorConfig( + service_id="starterproject", location="us-east4", connector="" + ) + + def test_connector_config_invalid_types(self): + with pytest.raises(ValueError, match="service_id must be a string"): + dataconnect.ConnectorConfig( + service_id=None, location="us-east4", connector="my_connector" + ) + with pytest.raises(ValueError, match="location must be a string"): + dataconnect.ConnectorConfig( + service_id="starterproject", location=123, connector="my_connector" + ) + with pytest.raises(ValueError, match="connector must be a string"): + dataconnect.ConnectorConfig( + service_id="starterproject", location="us-east4", connector=456 + ) + + +class TestDataConnect: + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_init_property_assignment(self): + cred = testutils.MockCredential() + try: + app = firebase_admin.initialize_app( + cred, options={'projectId': 'test-project'}, name="starter_app" + ) + except ValueError: + pytest.fail("initialize app has an error") + + try: + data_connect_instance = dataconnect.DataConnect(app, BASE_CONFIG) + except ValueError: + pytest.fail("DataConnect initialization failed.") + + assert data_connect_instance._app is app # pylint: disable=protected-access + 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 + + +class TestDataConnectClientFactory: + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def setup_method(self): + self.cred = testutils.MockCredential() + 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" + ) + + @mock.patch.object(dataconnect._DataConnectService, "get_client", autospec=True) + def test_client_successful(self, mock_get_client): + mock_get_client.side_effect = lambda service, config: dataconnect.DataConnect( + service._app, config # pylint: disable=protected-access + ) + client1 = dataconnect.client(self.config1, app=self.app) + client2 = dataconnect.client(self.config2, app=self.app) + assert mock_get_client.call_count == 2 + mock_get_client.assert_any_call(mock.ANY, self.config1) + mock_get_client.assert_any_call(mock.ANY, self.config2) + assert isinstance(client1, dataconnect.DataConnect) + assert client1.config is self.config1 + assert client1.app is self.app + assert client2.config is self.config2 + + def test_client_retrieval_different_apps_same_config(self): + 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) + + assert client1 is not client2 + assert client1.app is self.app + assert client1.app is not client2.app + + def test_invalid_config_type(self): + err_msg = "Config must be of type firebase_admin.dataconnect.ConnectorConfig" + with pytest.raises(ValueError, match=err_msg): + dataconnect.client("not-a-config", app=self.app) + + def test_invalid_app_type(self): + with pytest.raises(ValueError, match="Illegal app argument"): + dataconnect.client(self.config1, "not-a-app") + + def test_client_default_app(self): + 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 + + def test_client_none_config(self): + err_msg = "Config must be of type firebase_admin.dataconnect.ConnectorConfig" + with pytest.raises(ValueError, match=err_msg): + dataconnect.client(None, app=self.app) + + @mock.patch.object(_utils, "get_app_service", wraps=_utils.get_app_service) + def test_uses_app_service_mechanism(self, mock_get_app_service): + """Ensures dataconnect.client uses the standard app service loader.""" + dataconnect.client(self.config1, app=self.app) + mock_get_app_service.assert_called_once() + args, _ = mock_get_app_service.call_args + assert args[0] is self.app + assert args[1] == dataconnect._DATA_CONNECT_ATTRIBUTE # pylint: disable=protected-access + assert args[2] == dataconnect._DataConnectService # pylint: disable=protected-access + + +class TestDataConnectService: + + def setup_method(self): + self.cred = testutils.MockCredential() + 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() + + def test_cache_hit(self): + config = dataconnect.ConnectorConfig("s1", "l1", "c1") + client1 = self.service.get_client(config) + client2 = self.service.get_client(config) + assert client1 is client2 + + assert isinstance(client1, dataconnect.DataConnect) + assert client1.config is config + + def test_cache_miss_on_different_config(self): + config1 = dataconnect.ConnectorConfig("s1", "l1", "c1") + config2 = dataconnect.ConnectorConfig("s2", "l2", "c2") + client1 = self.service.get_client(config1) + client2 = self.service.get_client(config2) + assert client1 is not client2 + + @pytest.mark.parametrize( + "config_a, config_b, expect_same", + [ + ( + dataconnect.ConnectorConfig("s", "l", "c"), + dataconnect.ConnectorConfig("s", "l", "c_diff"), + False, + ), + ( + dataconnect.ConnectorConfig("s", "l", "c"), + dataconnect.ConnectorConfig("s", "l_diff", "c"), + False, + ), + ( + dataconnect.ConnectorConfig("s", "l", "c"), + dataconnect.ConnectorConfig("s_diff", "l", "c"), + False, + ), + ( + dataconnect.ConnectorConfig("s", "l", "c"), + dataconnect.ConnectorConfig("s", "l", "c"), + True, + ), + ], + ) + def test_complex_cache_key(self, config_a, config_b, expect_same): + client_a = self.service.get_client(config_a) + client_b = self.service.get_client(config_b) + if expect_same: + assert client_a is client_b + else: + assert client_a is not client_b + + def test_config_equivalence(self): + config1 = dataconnect.ConnectorConfig("s1", "l1", "c1") + config2 = dataconnect.ConnectorConfig("s1", "l1", "c1") + client1 = self.service.get_client(config1) + client2 = self.service.get_client(config2) + assert client1 is client2 + + @mock.patch("firebase_admin.dataconnect.DataConnect", autospec=True) + def test_client_creation_mocking(self, mock_data_connect): + config1 = dataconnect.ConnectorConfig("s_mock", "l_mock", "c_mock1") + config2 = dataconnect.ConnectorConfig("s_mock", "l_mock", "c_mock2") + + self.service.get_client(config1) + mock_data_connect.assert_called_once_with(app=self.app, config=config1) + + mock_data_connect.reset_mock() + + self.service.get_client(config1) + mock_data_connect.assert_not_called() + + mock_data_connect.reset_mock() + + # first call using config2 + self.service.get_client(config2) + mock_data_connect.assert_called_once_with(app=self.app, config=config2) + + @mock.patch("firebase_admin.dataconnect.DataConnect", autospec=True) + def test_error_handling_in_creation(self, mock_data_connect): + config = dataconnect.ConnectorConfig("s_err", "l_err", "c_err") + test_error = RuntimeError("Failed to create client") + mock_data_connect.side_effect = test_error + + with pytest.raises(RuntimeError, match="Failed to create client"): + self.service.get_client(config) + + # Ensure the failed creation wasn't cached + mock_data_connect.side_effect = None + self.service.get_client(config) + assert mock_data_connect.call_count == 2 + + def test_invalid_config_in_service(self): + err_msg = "Config must be of type firebase_admin.dataconnect.ConnectorConfig" + with pytest.raises(ValueError, match=err_msg): + self.service.get_client(None) + + +class TestDataConnectServiceWorkflow: + + def setup_method(self): + self.cred = testutils.MockCredential() + 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( + service_id="service2", location="us-east4", connector="conn2" + ) + self.config1_copy = dataconnect.ConnectorConfig( + service_id="starterproject", location="us-east4", connector="my_connector" + ) + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_overall_client_retrieval_and_caching(self): + client1a = dataconnect.client(self.config1, app=self.app1) + client1b = dataconnect.client(self.config1_copy, app=self.app1) + client2 = dataconnect.client(self.config2, app=self.app1) + + assert isinstance(client1a, dataconnect.DataConnect) + assert client1a.app is self.app1 + assert client1a.config is self.config1 + + # Same config + assert client1b is client1a + + # Different config + assert isinstance(client2, dataconnect.DataConnect) + assert client2.app is self.app1 + assert client2.config is self.config2 + assert client2 is not client1a + + # Different app + client1_app2 = dataconnect.client(self.config1, app=self.app2) + + assert isinstance(client1_app2, dataconnect.DataConnect) + assert client1_app2.app is self.app2 + assert client1_app2.config is self.config1 + assert client1_app2 is not client1a + + +class TestDataConnectApiClientConstructor: + + def setup_method(self): + self.cred = testutils.MockCredential() + + def teardown_method(self, method): + del method + testutils.cleanup_apps() + + def test_constructor_invalid_app(self): + msg = ( + "Second argument passed to DataConnectApiClient must be a valid " + "Firebase app instance." + ) + with pytest.raises(ValueError, match=msg): + dataconnect._DataConnectApiClient(BASE_CONFIG, None) + + def test_constructor_missing_project_id(self): + class CredentialWithoutProjectId(firebase_admin.credentials.Base): + def get_credential(self): + class DummyGoogleCred(google_auth_credentials.Credentials): + def refresh(self, request): + pass + return DummyGoogleCred() + + app_no_project_id = firebase_admin.initialize_app( + CredentialWithoutProjectId(), + name="no-project-id-app" + ) + try: + with pytest.raises(ValueError, match="Failed to determine project ID"): + dataconnect._DataConnectApiClient(BASE_CONFIG, app_no_project_id) + finally: + firebase_admin.delete_app(app_no_project_id) + + def test_constructor_connector_config(self): + app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'}) + api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, app) + assert api_client._connector_config is BASE_CONFIG + + def test_constructor_emulator_host_invalid(self, monkeypatch): + monkeypatch.setenv("DATA_CONNECT_EMULATOR_HOST", "http://localhost:9399") + app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'}) + with pytest.raises(ValueError, match="Invalid DATA_CONNECT_EMULATOR_HOST"): + dataconnect._DataConnectApiClient(BASE_CONFIG, app) + + +class TestDataConnectApiClientValidateGraphqlOptions: + + 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_validate_graphql_options_valid(self): + # Valid with no options + self.api_client._validate_graphql_options(None) + + # Valid with default options (no arguments) + options = dataconnect.GraphqlOptions() + self.api_client._validate_graphql_options(options) + + def test_validate_graphql_options_valid_impersonate(self): + # Valid unauthenticated impersonation + imp_unauth = dataconnect.Impersonation.unauthenticated() + options = dataconnect.GraphqlOptions(impersonate=imp_unauth) + self.api_client._validate_graphql_options(options) + + # Valid authenticated impersonation + imp_auth = dataconnect.Impersonation.authenticated( + {"sub": "authenticated-UUID"} + ) + options = dataconnect.GraphqlOptions(impersonate=imp_auth) + self.api_client._validate_graphql_options(options) + + def test_validate_graphql_options_valid_dataclass_variables(self): + options = dataconnect.GraphqlOptions(variables=TEST_DATACLASS_VARIABLES) + self.api_client._validate_graphql_options(options, CreateUserVariables) + + def test_validate_graphql_options_valid_mapping_variables(self): + options = dataconnect.GraphqlOptions(variables={"user_id": "1", "name": "Fred"}) + self.api_client._validate_graphql_options(options) + + def test_validate_graphql_options_valid_generic_variables(self): + options = dataconnect.GraphqlOptions(variables={"user_id": "1", "name": "Fred"}) + self.api_client._validate_graphql_options(options, Dict[str, Any]) + self.api_client._validate_graphql_options(options, Mapping[str, Any]) + + def test_validate_graphql_options_invalid_options(self): + with pytest.raises(ValueError, match="options must be a GraphqlOptions instance"): + self.api_client._validate_graphql_options("invalid-options") + + def test_validate_graphql_options_invalid_impersonate(self): + # impersonate must be dict + options = dataconnect.GraphqlOptions(impersonate="invalid") + with pytest.raises(ValueError, match="impersonate option must be a dictionary"): + self.api_client._validate_graphql_options(options) + + # impersonate must have either unauthenticated or auth_claims + options = dataconnect.GraphqlOptions(impersonate={"invalid_key": True}) + msg = ( + "impersonate option must contain either " + "'unauthenticated' or 'auth_claims'" + ) + with pytest.raises(ValueError, match=msg): + self.api_client._validate_graphql_options(options) + + # unauthenticated must be boolean + options = dataconnect.GraphqlOptions(impersonate={"unauthenticated": "not-bool"}) + with pytest.raises(ValueError, match="'unauthenticated' claim must be a boolean"): + self.api_client._validate_graphql_options(options) + + # auth_claims must be a dict + options = dataconnect.GraphqlOptions(impersonate={"auth_claims": "not-dict"}) + with pytest.raises(ValueError, match="'auth_claims' claim must be a dictionary"): + self.api_client._validate_graphql_options(options) + + # impersonate cannot contain both unauthenticated and auth_claims + options = dataconnect.GraphqlOptions( + impersonate={"unauthenticated": True, "auth_claims": {"uid": "123"}} + ) + msg = ( + "impersonate option cannot contain both " + "'unauthenticated' and 'auth_claims'" + ) + with pytest.raises(ValueError, match=msg): + self.api_client._validate_graphql_options(options) + + def test_validate_graphql_options_invalid_operation_name(self): + # Test type validation + options = dataconnect.GraphqlOptions(operation_name=123) + with pytest.raises(ValueError, match="operation_name must be a string"): + self.api_client._validate_graphql_options(options) + + # Test empty string validation + options = dataconnect.GraphqlOptions(operation_name="") + with pytest.raises(ValueError, match="operation_name must be a non-empty string"): + self.api_client._validate_graphql_options(options) + + # Test stripped whitespace validation + options = dataconnect.GraphqlOptions(operation_name=" ") + with pytest.raises(ValueError, match="operation_name must be a non-empty string"): + self.api_client._validate_graphql_options(options) + + def test_validate_graphql_options_invalid_variables(self): + # Test invalid variable format (not Mapping or dataclass) + options = dataconnect.GraphqlOptions(variables="invalid-string-format") + msg = "variables must be a collections.abc.Mapping or a dataclass" + with pytest.raises(ValueError, match=msg): + self.api_client._validate_graphql_options(options) + + # Test valid Mapping format but type mismatch against expected dataclass type + options = dataconnect.GraphqlOptions(variables=TEST_VARIABLES) + with pytest.raises(ValueError, match="variables must be of type CreateUserVariables"): + self.api_client._validate_graphql_options(options, CreateUserVariables) + + # Test type mismatch when variable_type is a tuple of classes (no __name__ attribute) + options = dataconnect.GraphqlOptions(variables=TEST_VARIABLES) + msg = r"variables must be of type \(\, \\)" + with pytest.raises(ValueError, match=msg): + self.api_client._validate_graphql_options(options, (list, tuple)) + + # Test type mismatch when a dataclass is passed but a Dict is expected + options = dataconnect.GraphqlOptions(variables=TEST_DATACLASS_VARIABLES) + with pytest.raises(ValueError, match="variables must be of type dict"): + self.api_client._validate_graphql_options(options, Dict[str, Any]) + + +class TestDataConnectApiClientPrepareGraphqlPayload: + + 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_prepare_graphql_payload_only_query(self): + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, None) + assert payload == TEST_PAYLOAD + + def test_prepare_graphql_payload_with_variables(self): + options = dataconnect.GraphqlOptions(variables=TEST_VARIABLES) + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, options) + assert payload == { + "query": TEST_QUERY, + "variables": TEST_VARIABLES + } + + def test_prepare_graphql_payload_with_dataclass_variables(self): + options = dataconnect.GraphqlOptions(variables=TEST_DATACLASS_VARIABLES) + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, options) + assert payload == { + "query": TEST_QUERY, + "variables": { + "user_id": "1", + "name": "Fred", + "profile": { + "address": "123 Road", + "phone": "332-3233-0199" + } + } + } + + def test_prepare_graphql_payload_with_operation_name(self): + options = dataconnect.GraphqlOptions(operation_name=" myOp ") + self.api_client._validate_graphql_options(options) + assert options.operation_name == " myOp " + + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, options) + assert payload == { + "query": TEST_QUERY, + "operationName": "myOp" + } + assert options.operation_name == " myOp " + + def test_prepare_graphql_payload_with_impersonate_unauthenticated(self): + imp_unauth = dataconnect.Impersonation.unauthenticated() + options = dataconnect.GraphqlOptions(impersonate=imp_unauth) + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, options) + assert payload == { + "query": TEST_QUERY, + "extensions": { + "impersonate": {"unauthenticated": True} + } + } + + def test_prepare_graphql_payload_with_impersonate_authenticated(self): + imp_auth = dataconnect.Impersonation.authenticated( + {"sub": "authenticated-UUID"} + ) + options = dataconnect.GraphqlOptions(impersonate=imp_auth) + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, options) + assert payload == { + "query": TEST_QUERY, + "extensions": { + "impersonate": {"authClaims": {"sub": "authenticated-UUID"}} + } + } + + def test_prepare_graphql_payload_with_all_fields(self): + imp_auth = dataconnect.Impersonation.authenticated( + {"sub": "authenticated-UUID"} + ) + options = dataconnect.GraphqlOptions( + variables=TEST_DATACLASS_VARIABLES, + operation_name="getUsers", + impersonate=imp_auth + ) + payload = self.api_client._prepare_graphql_payload(TEST_QUERY, options) + assert payload == { + "query": TEST_QUERY, + "operationName": "getUsers", + "variables": { + "user_id": "1", + "name": "Fred", + "profile": { + "address": "123 Road", + "phone": "332-3233-0199" + } + }, + "extensions": { + "impersonate": {"authClaims": {"sub": "authenticated-UUID"}} + } + } + + +class TestDataConnectApiClientServiceUrl: + + 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_get_firebase_dataconnect_service_url_production(self): + url = self.api_client._get_firebase_dataconnect_service_url("executeGraphql") + expected = ( + "https://firebasedataconnect.googleapis.com/v1" + "/projects/test-project/locations/us-east4" + "/services/starterproject:executeGraphql" + ) + assert url == expected + + def test_get_firebase_dataconnect_service_url_emulator(self, monkeypatch): + monkeypatch.setenv("DATA_CONNECT_EMULATOR_HOST", "localhost:9399") + api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app) + url = api_client._get_firebase_dataconnect_service_url("executeGraphql") + expected = ( + "http://localhost:9399/v1" + "/projects/test-project/locations/us-east4" + "/services/starterproject:executeGraphql" + ) + assert url == expected + + +class TestDataConnectApiClientGetHeaders: + + 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_get_headers(self): + headers = self.api_client._get_headers() + 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() + assert headers.get("X-Client-Version") == f"Python/Admin/{firebase_admin.__version__}" + assert headers.get("X-Firebase-Sqlconnect-Affinity") == "test-projectstarterproject" + + +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) + + res = self.api_client._make_gql_request(TEST_URL, TEST_HEADERS, TEST_PAYLOAD) + mock_body_and_response.assert_called_once_with( + "post", url=TEST_URL, headers=TEST_HEADERS, json=TEST_PAYLOAD + ) + assert res == {"data": "val"} + + def test_make_gql_request_missing_url(self): + with pytest.raises(ValueError, match="url, headers, and payload must all be specified."): + self.api_client._make_gql_request(None, TEST_HEADERS, TEST_PAYLOAD) + + def test_make_gql_request_missing_headers(self): + with pytest.raises(ValueError, match="url, headers, and payload must all be specified."): + self.api_client._make_gql_request(TEST_URL, None, TEST_PAYLOAD) + + def test_make_gql_request_missing_payload(self): + with pytest.raises(ValueError, match="url, headers, and payload must all be specified."): + self.api_client._make_gql_request(TEST_URL, TEST_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() + + with pytest.raises(exceptions.FirebaseError): + self.api_client._make_gql_request(TEST_URL, TEST_HEADERS, TEST_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 + ) + + with pytest.raises(exceptions.FirebaseError) as excinfo: + self.api_client._make_gql_request(TEST_URL, TEST_HEADERS, TEST_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 + ) + + with pytest.raises(exceptions.FirebaseError) as excinfo: + self.api_client._make_gql_request(TEST_URL, TEST_HEADERS, TEST_PAYLOAD) + + assert excinfo.value.code == "query-error" + assert str(excinfo.value) == "GraphQL execution failed: String error message" + + mock_body_and_response.return_value = ( + {"data": TEST_RESPONSE_DATA, "errors": []}, + mock_response + ) + res = self.api_client._make_gql_request(TEST_URL, TEST_HEADERS, TEST_PAYLOAD) + assert res == {"data": TEST_RESPONSE_DATA, "errors": []} + + +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) + assert isinstance(res, dataconnect.ExecuteGraphqlResponse) + assert res.data == {"name": "Fred", "age": 20} + + + 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") + + assert str(excinfo.value) == ( + "Response payload is not a valid JSON dictionary: not-a-dict" + ) + + +class TestImpersonation: + + def test_impersonation_unauthenticated_factory(self): + """Tests factory method for unauthenticated impersonation.""" + imp = dataconnect.Impersonation.unauthenticated() + assert imp == {"unauthenticated": True} + + def test_impersonation_authenticated_factory(self): + """Tests factory method for authenticated impersonation.""" + imp = dataconnect.Impersonation.authenticated(TEST_AUTH_CLAIMS) + assert imp == {"auth_claims": TEST_AUTH_CLAIMS} + + def test_impersonation_constructor_unauthenticated(self): + """Tests direct constructor with unauthenticated=True.""" + imp = dataconnect.Impersonation(unauthenticated=True) + assert imp == {"unauthenticated": True} + + def test_impersonation_constructor_auth_claims(self): + """Tests direct constructor with auth_claims dict.""" + imp = dataconnect.Impersonation(auth_claims=TEST_AUTH_CLAIMS) + assert imp == {"auth_claims": TEST_AUTH_CLAIMS} + + def test_impersonation_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_impersonation_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_impersonation_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_impersonation_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 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(TEST_QUERY, options="not-graphql-options") + + def test_execute_graphql_invalid_variables_type(self): + options = dataconnect.GraphqlOptions(variables={"name": "Fred"}) + with pytest.raises(ValueError, match="variables must be of type User"): + self.api_client.execute_graphql(TEST_QUERY, 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": TEST_RESPONSE_DATA} + res = self.api_client.execute_graphql(TEST_QUERY) + mock_make_gql_request.assert_called_once() + assert res.data == TEST_RESPONSE_DATA + + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_with_dataclass_variables(self, mock_make_gql_request): + 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 + ) + mock_make_gql_request.assert_called_once_with( + url=mock.ANY, + headers=mock.ANY, + payload={ + "query": "query CreateUser { user }", + "variables": {"name": "Fred"} + } + ) + assert res.data == {"user": {"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": TEST_RESPONSE_DATA} + res = self.api_client.execute_graphql_read(TEST_QUERY) + mock_make_gql_request.assert_called_once() + assert res.data == TEST_RESPONSE_DATA + + @mock.patch.object(dataconnect._DataConnectApiClient, "_make_gql_request") + def test_execute_graphql_read_with_dataclass_variables(self, mock_make_gql_request): + 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 + ) + mock_make_gql_request.assert_called_once_with( + url=mock.ANY, + headers=mock.ANY, + payload={ + "query": "query GetUser { user }", + "variables": {"name": "Fred"} + } + ) + assert res.data == {"user": {"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": TEST_RESPONSE_DATA} + self.api_client.execute_graphql(TEST_QUERY) + mock_make_gql_request.assert_called_once_with( + url=mock.ANY, + headers=mock.ANY, + payload=TEST_PAYLOAD + ) + + @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(TEST_QUERY) + + assert excinfo.value.code == exceptions.INTERNAL + assert str(excinfo.value) == ( + "Response payload is not a valid JSON dictionary: invalid-string-payload" + ) + + @pytest.mark.parametrize("error_class", [ + exceptions.InvalidArgumentError, + exceptions.PermissionDeniedError, + exceptions.NotFoundError, + exceptions.UnknownError + ]) + @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(TEST_QUERY) diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 00000000..4d57e44a --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,55 @@ +# 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. + +"""Test cases for the firebase_admin._utils module.""" + +import pytest +from firebase_admin import _utils + +class TestGetEmulatorHost: + + @pytest.mark.parametrize('host', [ + 'localhost:8080', + '127.0.0.1:8080', + '[::1]:8080', + '[2001:db8::1]:8080', + 'my-host:9000', + 'my_host:9000', + 'my.host.name:12345', + 'host_with_underscores:8080', + ]) + def test_get_emulator_host_valid(self, monkeypatch, host): + monkeypatch.setenv('TEST_EMULATOR_HOST', host) + assert _utils.get_emulator_host('TEST_EMULATOR_HOST') == host + + @pytest.mark.parametrize('host', [ + 'http://localhost:8080', + 'localhost', + '127.0.0.1', + '[::1]', + 'my_host', + 'localhost:abc', + 'localhost:', + ':8080', + 'invalid_host_name_with_chars$:8080', + 'host@name:8080', + ]) + def test_get_emulator_host_invalid(self, monkeypatch, host): + monkeypatch.setenv('TEST_EMULATOR_HOST', host) + with pytest.raises(ValueError, match='Invalid TEST_EMULATOR_HOST'): + _utils.get_emulator_host('TEST_EMULATOR_HOST') + + def test_get_emulator_host_not_set(self, monkeypatch): + monkeypatch.delenv('TEST_EMULATOR_HOST', raising=False) + assert _utils.get_emulator_host('TEST_EMULATOR_HOST') is None