feat(functions): Enable Cloud Task Queue Emulator support (#920)

* feat(functions): Enable Cloud Task Queue Emulator support

* fix: lint

* fix: Resolved issues from gemini review

* chore: Added basic integration tests for task enqueue and delete

* chore: Setup emulator testing for Functions integration tests

* fix: Re-added accidentally removed lint

* fix: integration test default apps

* fix: lint

Author: jonathanedey

.github/workflows/ci.yml

@@ -12,6 +12,17 @@ jobs:
 
     steps:
     - uses: actions/checkout@v4
+
+    - name: Set up Python 3.13 for emulator
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.13'
+    - name: Setup functions emulator environment
+      run: |
+        python -m venv integration/emulators/functions/venv
+        source integration/emulators/functions/venv/bin/activate
+        pip install -r integration/emulators/functions/requirements.txt
+        deactivate
     - name: Set up Python ${{ matrix.python }}
       uses: actions/setup-python@v5
       with:
@@ -26,11 +37,12 @@ jobs:
       uses: actions/setup-node@v4
       with:
         node-version: 20
-    - name: Run integration tests against emulator
-      run: |
-        npm install -g firebase-tools
-        firebase emulators:exec --only database --project fake-project-id 'pytest integration/test_db.py'
-
+    - name: Install firebase-tools
+      run: npm install -g firebase-tools
+    - name: Run Database emulator tests
+      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'
   lint:
     runs-on: ubuntu-latest
     steps:

CONTRIBUTING.md

@@ -252,6 +252,17 @@ to ensure that exported user records contain the password hashes of the user acc
    3. Click **ADD ANOTHER ROLE** and choose **Firebase Authentication Admin**.
    4. Click **SAVE**.
 
+9. Enable Cloud Tasks:
+   1. Search for and enable **Cloud Run**.
+   2. Search for and enable **Cloud Tasks**.
+   3. Go to [Google Cloud console | IAM & admin](https://console.cloud.google.com/iam-admin)
+   and make sure your Firebase project is selected.
+   4. Ensure your service account has the following required roles:
+      * **Cloud Tasks Enqueuer** - `cloudtasks.taskEnqueuer`
+      * **Cloud Tasks Task Deleter** - `cloudtasks.taskDeleter`
+      * **Cloud Run Invoker** - `run.invoker`
+      * **Service Account User** - `iam.serviceAccountUser`
+
 
 Now you can invoke the integration test suite as follows:
 

firebase_admin/functions.py

@@ -18,6 +18,7 @@
 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
@@ -49,6 +50,8 @@
     'https://cloudtasks.googleapis.com/v2/' + _CLOUD_TASKS_API_RESOURCE_PATH
 _FIREBASE_FUNCTION_URL_FORMAT = \
     'https://{location_id}-{project_id}.cloudfunctions.net/{resource_id}'
+_EMULATOR_HOST_ENV_VAR = 'CLOUD_TASKS_EMULATOR_HOST'
+_EMULATED_SERVICE_ACCOUNT_DEFAULT = 'emulated-service-acct@email.com'
 
 _FUNCTIONS_HEADERS = {
     'X-GOOG-API-FORMAT-VERSION': '2',
@@ -58,6 +61,17 @@
 # Default canonical location ID of the task queue.
 _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
+
+
 def _get_functions_service(app) -> _FunctionsService:
     return _utils.get_app_service(app, _FUNCTIONS_ATTRIBUTE, _FunctionsService)
 
@@ -103,13 +117,19 @@ def __init__(self, app: App):
                 'projectId option, or use service account credentials. Alternatively, set the '
                 'GOOGLE_CLOUD_PROJECT environment variable.')
 
-        self._credential = app.credential.get_credential()
+        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 task_queue(self, function_name: str, extension_id: Optional[str] = None) -> TaskQueue:
         """Creates a TaskQueue instance."""
         return TaskQueue(
-            function_name, extension_id, self._project_id, self._credential, self._http_client)
+            function_name, extension_id, self._project_id, self._credential, self._http_client,
+            self._emulator_host)
 
     @classmethod
     def handle_functions_error(cls, error: Any):
@@ -125,7 +145,8 @@ def __init__(
             extension_id: Optional[str],
             project_id,
             credential,
-            http_client
+            http_client,
+            emulator_host: Optional[str] = None
         ) -> None:
 
         # Validate function_name
@@ -134,6 +155,7 @@ def __init__(
         self._project_id = project_id
         self._credential = credential
         self._http_client = http_client
+        self._emulator_host = emulator_host
         self._function_name = function_name
         self._extension_id = extension_id
         # Parse resources from function_name
@@ -167,16 +189,26 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str:
             str: The ID of the task relative to this queue.
         """
         task = self._validate_task_options(task_data, self._resource, opts)
-        service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT)
+        emulator_url = self._get_emulator_url(self._resource)
+        service_url = emulator_url or self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT)
         task_payload = self._update_task_payload(task, self._resource, self._extension_id)
         try:
             resp = self._http_client.body(
                 'post',
                 url=service_url,
                 headers=_FUNCTIONS_HEADERS,
-                json={'task': task_payload.__dict__}
+                json={'task': task_payload.to_api_dict()}
             )
-            task_name = resp.get('name', None)
+            if self._is_emulated():
+                # Emulator returns a response with format {task: {name: <task_name>}}
+                # The task name also has an extra '/' at the start compared to prod
+                task_info = resp.get('task') or {}
+                task_name = task_info.get('name')
+                if task_name:
+                    task_name = task_name[1:]
+            else:
+                # Production returns a response with format {name: <task_name>}
+                task_name = resp.get('name')
             task_resource = \
                 self._parse_resource_name(task_name, f'queues/{self._resource.resource_id}/tasks')
             return task_resource.resource_id
@@ -197,7 +229,11 @@ def delete(self, task_id: str) -> None:
             ValueError: If the input arguments are invalid.
         """
         _Validators.check_non_empty_string('task_id', task_id)
-        service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}')
+        emulator_url = self._get_emulator_url(self._resource)
+        if emulator_url:
+            service_url = emulator_url + f'/{task_id}'
+        else:
+            service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}')
         try:
             self._http_client.body(
                 'delete',
@@ -235,8 +271,8 @@ def _validate_task_options(
         """Validate and create a Task from optional ``TaskOptions``."""
         task_http_request = {
             'url': '',
-            'oidc_token': {
-                'service_account_email': ''
+            'oidcToken': {
+                'serviceAccountEmail': ''
             },
             'body': b64encode(json.dumps(data).encode()).decode(),
             'headers': {
@@ -250,7 +286,7 @@ def _validate_task_options(
                 task.http_request['headers'] = {**task.http_request['headers'], **opts.headers}
             if opts.schedule_time is not None and opts.schedule_delay_seconds is not None:
                 raise ValueError(
-                    'Both sechdule_delay_seconds and schedule_time cannot be set at the same time.')
+                    'Both schedule_delay_seconds and schedule_time cannot be set at the same time.')
             if opts.schedule_time is not None and opts.schedule_delay_seconds is None:
                 if not isinstance(opts.schedule_time, datetime):
                     raise ValueError('schedule_time should be UTC datetime.')
@@ -288,7 +324,10 @@ def _update_task_payload(self, task: Task, resource: Resource, extension_id: str
         """Prepares task to be sent with credentials."""
         # Get function url from task or generate from resources
         if not _Validators.is_non_empty_string(task.http_request['url']):
-            task.http_request['url'] = self._get_url(resource, _FIREBASE_FUNCTION_URL_FORMAT)
+            if self._is_emulated():
+                task.http_request['url'] = ''
+            else:
+                task.http_request['url'] = self._get_url(resource, _FIREBASE_FUNCTION_URL_FORMAT)
 
         # Refresh the credential to ensure all attributes (e.g. service_account_email, id_token)
         # are populated, preventing cold start errors.
@@ -298,20 +337,40 @@ def _update_task_payload(self, task: Task, resource: Resource, extension_id: str
             except RefreshError as err:
                 raise ValueError(f'Initial task payload credential refresh failed: {err}') from err
 
-        # If extension id is provided, it emplies that it is being run from a deployed extension.
+        # If extension id is provided, it implies that it is being run from a deployed extension.
         # Meaning that it's credential should be a Compute Engine Credential.
         if _Validators.is_non_empty_string(extension_id) and \
             isinstance(self._credential, ComputeEngineCredentials):
             id_token = self._credential.token
             task.http_request['headers'] = \
                 {**task.http_request['headers'], 'Authorization': f'Bearer {id_token}'}
             # Delete oidc token
-            del task.http_request['oidc_token']
+            del task.http_request['oidcToken']
         else:
-            task.http_request['oidc_token'] = \
-                {'service_account_email': self._credential.service_account_email}
+            try:
+                task.http_request['oidcToken'] = \
+                    {'serviceAccountEmail': self._credential.service_account_email}
+            except AttributeError as error:
+                if self._is_emulated():
+                    task.http_request['oidcToken'] = \
+                        {'serviceAccountEmail': _EMULATED_SERVICE_ACCOUNT_DEFAULT}
+                else:
+                    raise ValueError(
+                        'Failed to determine service account. Initialize the SDK with service '
+                        'account credentials or set service account ID as an app option.'
+                        ) from error
         return task
 
+    def _get_emulator_url(self, resource: Resource):
+        if self._emulator_host:
+            emulator_url_format = f'http://{self._emulator_host}/' + _CLOUD_TASKS_API_RESOURCE_PATH
+            url = self._get_url(resource, emulator_url_format)
+            return url
+        return None
+
+    def _is_emulated(self):
+        return self._emulator_host is not None
+
 
 class _Validators:
     """A collection of data validation utilities."""
@@ -436,6 +495,14 @@ class Task:
     schedule_time: Optional[str] = None
     dispatch_deadline: Optional[str] = None
 
+    def to_api_dict(self) -> dict:
+        """Converts the Task object to a dictionary suitable for the Cloud Tasks API."""
+        return {
+            'httpRequest': self.http_request,
+            'name': self.name,
+            'scheduleTime': self.schedule_time,
+            'dispatchDeadline': self.dispatch_deadline,
+        }
 
 @dataclass
 class Resource:

integration/emulators/.gitignore

@@ -0,0 +1,69 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+firebase-debug.log*
+firebase-debug.*.log*
+
+# Firebase cache
+.firebase/
+
+# Firebase config
+
+# Uncomment this if you'd like others to create their own Firebase project.
+# For a team working on the same Firebase project(s), it is recommended to leave
+# it commented so all members can deploy to the same project(s) in .firebaserc.
+# .firebaserc
+
+# Runtime data
+pids
+*.pid
+*.seed
+*.pid.lock
+
+# Directory for instrumented libs generated by jscoverage/JSCover
+lib-cov
+
+# Coverage directory used by tools like istanbul
+coverage
+
+# nyc test coverage
+.nyc_output
+
+# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files)
+.grunt
+
+# Bower dependency directory (https://bower.io/)
+bower_components
+
+# node-waf configuration
+.lock-wscript
+
+# Compiled binary addons (http://nodejs.org/api/addons.html)
+build/Release
+
+# Dependency directories
+node_modules/
+
+# Optional npm cache directory
+.npm
+
+# Optional eslint cache
+.eslintcache
+
+# Optional REPL history
+.node_repl_history
+
+# Output of 'npm pack'
+*.tgz
+
+# Yarn Integrity file
+.yarn-integrity
+
+# dotenv environment variables file
+.env
+
+# dataconnect generated files
+.dataconnect

integration/emulators/firebase.json

@@ -0,0 +1,29 @@
+{
+  "emulators": {
+    "tasks": {
+      "port": 9499
+    },
+    "ui": {
+      "enabled": false
+    },
+    "singleProjectMode": true,
+    "functions": {
+      "port": 5001
+    }
+  },
+  "functions": [
+    {
+      "source": "functions",
+      "codebase": "default",
+      "disallowLegacyRuntimeConfig": true,
+      "ignore": [
+        "venv",
+        ".git",
+        "firebase-debug.log",
+        "firebase-debug.*.log",
+        "*.local"
+      ],
+      "runtime": "python313"
+    }
+  ]
+}

integration/emulators/functions/.gitignore

@@ -0,0 +1,6 @@
+# Python bytecode
+__pycache__/
+
+# Python virtual environment
+venv/
+*.local

integration/emulators/functions/main.py

@@ -0,0 +1,7 @@
+from firebase_functions import tasks_fn
+
+@tasks_fn.on_task_dispatched()
+def testTaskQueue(req: tasks_fn.CallableRequest) -> None:
+    """Handles tasks from the task queue."""
+    print(f"Received task with data: {req.data}")
+    return

integration/emulators/functions/requirements.txt

@@ -0,0 +1 @@
+firebase_functions~=0.4.1