diff --git a/.github/resources/integ-service-account.json.gpg b/.github/resources/integ-service-account.json.gpg
index 7740dccd8..5a52805c9 100644
Binary files a/.github/resources/integ-service-account.json.gpg and b/.github/resources/integ-service-account.json.gpg differ
diff --git a/.github/scripts/publish_preflight_check.sh b/.github/scripts/publish_preflight_check.sh
index 1d001c3b9..38fe49a88 100755
--- a/.github/scripts/publish_preflight_check.sh
+++ b/.github/scripts/publish_preflight_check.sh
@@ -159,8 +159,8 @@ echo_info "Generating changelog"
 echo_info "--------------------------------------------"
 echo_info ""
 
-echo_info "---< git fetch origin master --prune --unshallow >---"
-git fetch origin master --prune --unshallow
+echo_info "---< git fetch origin main --prune --unshallow >---"
+git fetch origin main --prune --unshallow
 echo ""
 
 echo_info "Generating changelog from history..."
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bfd29e2cc..5bf78a56b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -11,9 +11,20 @@ jobs:
         python: ['3.9', '3.10', '3.11', '3.12', '3.13', 'pypy3.9']
 
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
+
+    - name: Set up Python 3.13 for emulator
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
+      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
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: ${{ matrix.python }}
     - name: Install dependencies
@@ -23,20 +34,27 @@ jobs:
     - name: Test with pytest
       run: pytest
     - name: Set up Node.js 20
-      uses: actions/setup-node@v4
+      uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # 4.4.0
       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: Set up Java 21
+      uses: actions/setup-java@f2beeb24e141e01a676f977032f5a29d81c9e27e # 5.1.0
+      with:
+        distribution: 'temurin'
+        java-version: '21'
+        check-latest: true
+    - 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:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
     - name: Set up Python 3.9
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: 3.9
     - name: Install dependencies
diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml
index 3d5420537..d60b3cd0b 100644
--- a/.github/workflows/nightly.yml
+++ b/.github/workflows/nightly.yml
@@ -29,12 +29,12 @@ jobs:
 
     steps:
     - name: Checkout source for staging
-      uses: actions/checkout@v4
+      uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
       with:
         ref: ${{ github.event.client_payload.ref || github.ref }}
 
     - name: Set up Python
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: 3.9
 
@@ -63,14 +63,14 @@ jobs:
     # Attach the packaged artifacts to the workflow output. These can be manually
     # downloaded for later inspection if necessary.
     - name: Archive artifacts
-      uses: actions/upload-artifact@v4
+      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
       with:
         name: dist
         path: dist
 
     - name: Send email on failure
       if: failure()
-      uses: firebase/firebase-admin-node/.github/actions/send-email@master
+      uses: firebase/firebase-admin-node/.github/actions/send-email@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
       with:
         api-key: ${{ secrets.OSS_BOT_MAILGUN_KEY }}
         domain: ${{ secrets.OSS_BOT_MAILGUN_DOMAIN }}
@@ -85,7 +85,7 @@ jobs:
 
     - name: Send email on cancelled
       if: cancelled()
-      uses: firebase/firebase-admin-node/.github/actions/send-email@master
+      uses: firebase/firebase-admin-node/.github/actions/send-email@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
       with:
         api-key: ${{ secrets.OSS_BOT_MAILGUN_KEY }}
         domain: ${{ secrets.OSS_BOT_MAILGUN_DOMAIN }}
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 6cd1d3f07..6bbf19aab 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -15,10 +15,18 @@
 name: Release Candidate
 
 on:
-  # Only run the workflow when a PR is updated or when a developer explicitly requests
-  # a build by sending a 'firebase_build' event.
+  # Run the workflow when:
+  # 1. A PR is created or updated (staging checks).
+  # 2. A commit is pushed to main (release publication).
+  # 3. A developer explicitly requests a build via 'firebase_build' event.
   pull_request:
-    types: [opened, synchronize, closed]
+    types: [opened, synchronize]
+
+  push:
+    branches:
+      - main
+    paths:
+      - 'firebase_admin/__about__.py'
 
   repository_dispatch:
     types:
@@ -26,26 +34,22 @@ on:
 
 jobs:
   stage_release:
-    # To publish a release, merge the release PR with the label 'release:publish'.
+    # To publish a release, merge a PR with the title prefix '[chore] Release ' to main
+    # and ensure the squashed commit message also has the prefix.
     # To stage a release without publishing it, send a 'firebase_build' event or apply
     # the 'release:stage' label to a PR.
     if: github.event.action == 'firebase_build' ||
       contains(github.event.pull_request.labels.*.name, 'release:stage') ||
-      (github.event.pull_request.merged &&
-        contains(github.event.pull_request.labels.*.name, 'release:publish'))
+      (github.event_name == 'push' && startsWith(github.event.head_commit.message, '[chore] Release '))
 
     runs-on: ubuntu-latest
 
-    # When manually triggering the build, the requester can specify a target branch or a tag
-    # via the 'ref' client parameter.
     steps:
     - name: Checkout source for staging
-      uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.client_payload.ref || github.ref }}
+      uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
 
     - name: Set up Python
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # 5.6.0
       with:
         python-version: 3.9
 
@@ -74,7 +78,7 @@ jobs:
     # Attach the packaged artifacts to the workflow output. These can be manually
     # downloaded for later inspection if necessary.
     - name: Archive artifacts
-      uses: actions/upload-artifact@v4
+      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
       with:
         name: dist
         path: dist
@@ -82,17 +86,16 @@ jobs:
   publish_release:
     needs: stage_release
 
-    # Check whether the release should be published. We publish only when the trigger PR is
-    #   1. merged
-    #   2. to the master branch
-    #   3. with the label 'release:publish', and
-    #   4. the title prefix '[chore] Release '.
-    if: github.event.pull_request.merged &&
-      github.ref == 'refs/heads/master' &&
-      contains(github.event.pull_request.labels.*.name, 'release:publish') &&
-      startsWith(github.event.pull_request.title, '[chore] Release ')
+    # Check whether the release should be published. We publish only when the trigger is
+    #   1. a push (merge)
+    #   2. to the main branch
+    #   3. and the commit message has the title prefix '[chore] Release '.
+    if: github.event_name == 'push' &&
+      github.ref == 'refs/heads/main' &&
+      startsWith(github.event.head_commit.message, '[chore] Release ')
 
     runs-on: ubuntu-latest
+    environment: Release
     permissions:
       # Used to create a short-lived OIDC token which is given to PyPi to identify this workflow job
       # See: https://docs.github.com/en/actions/deployment/security-hardening-your-deployments/about-security-hardening-with-openid-connect#adding-permissions-settings
@@ -102,11 +105,11 @@ jobs:
 
     steps:
     - name: Checkout source for publish
-      uses: actions/checkout@v4
+      uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # 4.3.1
 
     # Download the artifacts created by the stage_release job.
     - name: Download release candidates
-      uses: actions/download-artifact@v4.1.7
+      uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
       with:
         name: dist
         path: dist
@@ -119,24 +122,12 @@ jobs:
     - name: Create release tag
       env:
         GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      run: gh release create ${{ steps.preflight.outputs.version }}
-            --title "Firebase Admin Python SDK ${{ steps.preflight.outputs.version }}"
-            --notes '${{ steps.preflight.outputs.changelog }}'
+        RELEASE_VER: ${{ steps.preflight.outputs.version }}
+        RELEASE_BODY: ${{ steps.preflight.outputs.changelog }}
+      run: |
+        gh release create "$RELEASE_VER" \
+          --title "Firebase Admin Python SDK $RELEASE_VER" \
+          --notes "$RELEASE_BODY"
 
     - name: Publish to Pypi
-      uses: pypa/gh-action-pypi-publish@release/v1
-
-    # Post to Twitter if explicitly opted-in by adding the label 'release:tweet'.
-    - name: Post to Twitter
-      if: success() &&
-        contains(github.event.pull_request.labels.*.name, 'release:tweet')
-      uses: firebase/firebase-admin-node/.github/actions/send-tweet@master
-      with:
-        status: >
-          ${{ steps.preflight.outputs.version }} of @Firebase Admin Python SDK is available.
-          https://github.com/firebase/firebase-admin-python/releases/tag/${{ steps.preflight.outputs.version }}
-        consumer-key: ${{ secrets.TWITTER_CONSUMER_KEY }}
-        consumer-secret: ${{ secrets.TWITTER_CONSUMER_SECRET }}
-        access-token: ${{ secrets.TWITTER_ACCESS_TOKEN }}
-        access-token-secret: ${{ secrets.TWITTER_ACCESS_TOKEN_SECRET }}
-      continue-on-error: true
+      uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 72933a24f..139e7f96c 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -47,7 +47,7 @@ Great, we love hearing how we can improve our products! Share you idea through o
 ## <a name="submit"></a>Want to submit a pull request?
 
 Sweet, we'd love to accept your contribution!
-[Open a new pull request](https://github.com/firebase/firebase-admin-python/pull/new/master) and fill
+[Open a new pull request](https://github.com/firebase/firebase-admin-python/pull/new) and fill
 out the provided template.
 
 **If you want to implement a new feature, please open an issue with a proposal first so that we can
@@ -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:
 
diff --git a/firebase_admin/__about__.py b/firebase_admin/__about__.py
index 9fb40b11c..e8ae3bb46 100644
--- a/firebase_admin/__about__.py
+++ b/firebase_admin/__about__.py
@@ -14,7 +14,7 @@
 
 """About information (version, etc) for Firebase Admin SDK."""
 
-__version__ = '7.1.0'
+__version__ = '7.4.0'
 __title__ = 'firebase_admin'
 __author__ = 'Firebase'
 __license__ = 'Apache License 2.0'
diff --git a/firebase_admin/_auth_utils.py b/firebase_admin/_auth_utils.py
index a514442c4..8f3c419a7 100644
--- a/firebase_admin/_auth_utils.py
+++ b/firebase_admin/_auth_utils.py
@@ -479,7 +479,7 @@ def _parse_error_body(response):
         separator = code.find(':')
         if separator != -1:
             custom_message = code[separator + 1:].strip()
-            code = code[:separator]
+            code = code[:separator].strip()
 
     return code, custom_message
 
diff --git a/firebase_admin/_messaging_encoder.py b/firebase_admin/_messaging_encoder.py
index 960a6d742..4c0c6daa4 100644
--- a/firebase_admin/_messaging_encoder.py
+++ b/firebase_admin/_messaging_encoder.py
@@ -207,6 +207,10 @@ def encode_android(cls, android):
             'fcm_options': cls.encode_android_fcm_options(android.fcm_options),
             'direct_boot_ok': _Validators.check_boolean(
                 'AndroidConfig.direct_boot_ok', android.direct_boot_ok),
+            'bandwidth_constrained_ok': _Validators.check_boolean(
+                'AndroidConfig.bandwidth_constrained_ok', android.bandwidth_constrained_ok),
+            'restricted_satellite_ok': _Validators.check_boolean(
+                'AndroidConfig.restricted_satellite_ok', android.restricted_satellite_ok),
         }
         result = cls.remove_null_values(result)
         priority = result.get('priority')
diff --git a/firebase_admin/_messaging_utils.py b/firebase_admin/_messaging_utils.py
index 8fd720701..773ed6057 100644
--- a/firebase_admin/_messaging_utils.py
+++ b/firebase_admin/_messaging_utils.py
@@ -13,6 +13,9 @@
 # limitations under the License.
 
 """Types and utilities used by the messaging (FCM) module."""
+from __future__ import annotations
+import datetime
+from typing import Dict, Optional, Union
 
 from firebase_admin import exceptions
 
@@ -51,10 +54,25 @@ class AndroidConfig:
         fcm_options: A ``messaging.AndroidFCMOptions`` to be included in the message (optional).
         direct_boot_ok: A boolean indicating whether messages will be allowed to be delivered to
             the app while the device is in direct boot mode (optional).
+        bandwidth_constrained_ok: A boolean indicating whether messages will be allowed to be
+            delivered to the app while the device is on a bandwidth constrained network (optional).
+        restricted_satellite_ok: A boolean indicating whether messages will be allowed to be
+            delivered to the app while the device is on a restricted satellite network (optional).
     """
 
-    def __init__(self, collapse_key=None, priority=None, ttl=None, restricted_package_name=None,
-                 data=None, notification=None, fcm_options=None, direct_boot_ok=None):
+    def __init__(
+        self,
+        collapse_key: Optional[str] = None,
+        priority: Optional[str] = None,
+        ttl: Optional[Union[int, float, datetime.timedelta]] = None,
+        restricted_package_name: Optional[str] = None,
+        data: Optional[Dict[str, str]] = None,
+        notification: Optional[AndroidNotification] = None,
+        fcm_options: Optional[AndroidFCMOptions] = None,
+        direct_boot_ok: Optional[bool] = None,
+        bandwidth_constrained_ok: Optional[bool] = None,
+        restricted_satellite_ok: Optional[bool] = None
+    ):
         self.collapse_key = collapse_key
         self.priority = priority
         self.ttl = ttl
@@ -63,6 +81,8 @@ def __init__(self, collapse_key=None, priority=None, ttl=None, restricted_packag
         self.notification = notification
         self.fcm_options = fcm_options
         self.direct_boot_ok = direct_boot_ok
+        self.bandwidth_constrained_ok = bandwidth_constrained_ok
+        self.restricted_satellite_ok = restricted_satellite_ok
 
 
 class AndroidNotification:
diff --git a/firebase_admin/_utils.py b/firebase_admin/_utils.py
index d0aca884b..0277b9e5f 100644
--- a/firebase_admin/_utils.py
+++ b/firebase_admin/_utils.py
@@ -279,7 +279,6 @@ def handle_httpx_error(error: httpx.HTTPError, message=None, code=None) -> excep
             message=f'Failed to establish a connection: {error}',
             cause=error)
     if isinstance(error, httpx.HTTPStatusError):
-        print("printing status error", error)
         if not code:
             code = _http_status_to_error_code(error.response.status_code)
         if not message:
diff --git a/firebase_admin/credentials.py b/firebase_admin/credentials.py
index 7117b71a9..0edbecaae 100644
--- a/firebase_admin/credentials.py
+++ b/firebase_admin/credentials.py
@@ -37,7 +37,7 @@
 AccessTokenInfo = collections.namedtuple('AccessTokenInfo', ['access_token', 'expiry'])
 """Data included in an OAuth2 access token.
 
-Contains the access token string and the expiry time. The expirty time is exposed as a
+Contains the access token string and the expiry time. The expiry time is exposed as a
 ``datetime`` value.
 """
 
diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py
index 6db0fbb42..66ba700b3 100644
--- a/firebase_admin/functions.py
+++ b/firebase_admin/functions.py
@@ -18,11 +18,16 @@
 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
 from dataclasses import dataclass
+
 from google.auth.compute_engine import Credentials as ComputeEngineCredentials
+from google.auth.credentials import TokenState
+from google.auth.exceptions import RefreshError
+from google.auth.transport import requests as google_auth_requests
 
 import requests
 import firebase_admin
@@ -45,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',
@@ -54,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)
 
@@ -99,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):
@@ -121,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
@@ -130,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
@@ -163,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
@@ -193,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',
@@ -231,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': {
@@ -246,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.')
@@ -284,22 +324,53 @@ 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 extension id is provided, it emplies that it is being run from a deployed extension.
+            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.
+        if self._credential.token_state != TokenState.FRESH:
+            try:
+                self._credential.refresh(google_auth_requests.Request())
+            except RefreshError as err:
+                raise ValueError(f'Initial task payload credential refresh failed: {err}') from err
+
+        # 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}'}
+                {**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."""
@@ -424,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:
diff --git a/firebase_admin/phone_number_verification.py b/firebase_admin/phone_number_verification.py
new file mode 100644
index 000000000..ea8f0cdab
--- /dev/null
+++ b/firebase_admin/phone_number_verification.py
@@ -0,0 +1,250 @@
+# 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 Phone Number Verification module.
+
+This module contains functions for verifying JWTs related to the Firebase
+Phone Number Verification service.
+"""
+from __future__ import annotations
+from typing import Any, Dict, Optional
+
+import jwt
+from jwt import (
+    PyJWKClient, InvalidSignatureError,
+    PyJWKClientError, InvalidAudienceError, InvalidIssuerError, ExpiredSignatureError
+)
+
+from firebase_admin import App, _utils, exceptions
+
+_FPNV_ATTRIBUTE = '_phone_number_verification'
+_FPNV_JWKS_URL = 'https://fpnv.googleapis.com/v1beta/jwks'
+_FPNV_ISSUER = 'https://fpnv.googleapis.com/projects/'
+_ALGORITHM_ES256 = 'ES256'
+
+
+def _get_fpnv_service(app):
+    return _utils.get_app_service(app, _FPNV_ATTRIBUTE, _FpnvService)
+
+def verify_token(token: str, app: Optional[App] = None) -> PhoneNumberVerificationToken:
+    """Verifies a Firebase Phone Number Verification token.
+
+    Args:
+        token: A string containing the Firebase Phone Number Verification JWT.
+        app: An App instance (optional).
+
+    Returns:
+        PhoneNumberVerificationToken: The verified token claims.
+
+    Raises:
+        ValueError: If the token is not a string or is empty.
+        InvalidTokenError: If the token is invalid or malformed.
+        ExpiredTokenError: If the token has expired.
+    """
+    return _get_fpnv_service(app).verify_token(token)
+
+
+class PhoneNumberVerificationToken(dict):
+    """Represents a verified Firebase Phone Number Verification token.
+
+    This class behaves like a dictionary, allowing access to the decoded claims.
+    It also provides convenience properties for common claims.
+    """
+
+    def __init__(self, claims):
+        super().__init__(claims)
+        self['phone_number'] = claims.get('sub')
+
+    @property
+    def phone_number(self) -> str:
+        """Returns the phone number of the user.
+        This corresponds to the 'sub' claim in the JWT.
+        """
+        return self.get('sub')
+
+    @property
+    def issuer(self) -> str:
+        """Returns the issuer identifier for the issuer of the response."""
+        return self.get('iss')
+
+    @property
+    def audience(self) -> str:
+        """Returns the audience for which this token is intended."""
+        return self.get('aud')
+
+    @property
+    def exp(self) -> int:
+        """Returns the expiration time since the Unix epoch."""
+        return self.get('exp')
+
+    @property
+    def iat(self) -> int:
+        """Returns the issued-at time since the Unix epoch."""
+        return self.get('iat')
+
+    @property
+    def sub(self) -> str:
+        """Returns the sub (subject) of the token, which is the phone number."""
+        return self.get('sub')
+
+    @property
+    def claims(self):
+        """Returns the entire map of claims."""
+        return self
+
+
+class _FpnvService:
+    """Service class that implements Firebase Phone Number Verification functionality."""
+    _project_id = None
+
+    def __init__(self, app):
+        self._project_id = app.project_id
+        if not self._project_id:
+            raise ValueError(
+                'Project ID is required for Firebase Phone Number Verification. Please ensure the '
+                'app is initialized with a credential that contains a project ID.'
+            )
+
+        self._verifier = _FpnvTokenVerifier(self._project_id)
+
+    def verify_token(self, token) -> PhoneNumberVerificationToken:
+        """Verifies a Firebase Phone Number Verification token.
+
+        Verifies the signature, expiration, and claims of the token.
+
+        Args:
+            token: A string containing the Firebase Phone Number Verification JWT.
+
+        Returns:
+            PhoneNumberVerificationToken: The verified token claims.
+
+        Raises:
+            ValueError: If the token is not a string or is empty.
+            InvalidTokenError: If the token is invalid or malformed.
+            ExpiredTokenError: If the token has expired.
+        """
+        return PhoneNumberVerificationToken(self._verifier.verify(token))
+
+
+class _FpnvTokenVerifier:
+    """Internal class for verifying Firebase Phone Number Verification JWTs signed with ES256."""
+    _jwks_client = None
+    _project_id = None
+
+    def __init__(self, project_id):
+        self._project_id = project_id
+        self._jwks_client = PyJWKClient(_FPNV_JWKS_URL, lifespan=21600)
+
+    def verify(self, token) -> Dict[str, Any]:
+        """Verifies the given Firebase Phone Number Verification token."""
+        _Validators.check_string("Firebase Phone Number Verification check token", token)
+        try:
+            self._validate_headers(jwt.get_unverified_header(token))
+            signing_key = self._jwks_client.get_signing_key_from_jwt(token)
+            claims = self._decode_and_verify(token, signing_key.key)
+        except (jwt.InvalidTokenError, PyJWKClientError) as exception:
+            raise InvalidTokenError(
+                'Verifying phone number verification token failed.',
+                cause=exception,
+                http_response=getattr(exception, 'http_response', None)
+            ) from exception
+
+        return claims
+
+    def _validate_headers(self, headers: Any) -> None:
+        """Validates the headers."""
+        if headers.get('kid') is None:
+            raise InvalidTokenError("Token has no 'kid' claim.")
+
+        if headers.get('typ') != 'JWT':
+            raise InvalidTokenError(
+                'The provided token has an incorrect type header. ' \
+                f"Expected 'JWT' but got {headers.get('typ')!r}."
+            )
+
+        algorithm = headers.get('alg')
+        if algorithm != _ALGORITHM_ES256:
+            raise InvalidTokenError(
+                'The provided token has an incorrect alg header. '
+                f'Expected {_ALGORITHM_ES256} but got {algorithm}.'
+            )
+
+    def _decode_and_verify(self, token, signing_key) -> Dict[str, Any]:
+        """Decodes and verifies the token."""
+        expected_issuer = f'{_FPNV_ISSUER}{self._project_id}'
+        try:
+            payload = jwt.decode(
+                token,
+                signing_key,
+                algorithms=[_ALGORITHM_ES256],
+                audience=expected_issuer,
+                issuer=expected_issuer
+            )
+        except InvalidSignatureError as exception:
+            raise InvalidTokenError(
+                'The provided token has an invalid signature.'
+            ) from exception
+        except InvalidAudienceError as exception:
+            raise InvalidTokenError(
+                'The provided token has an incorrect "aud" (audience) claim. '
+                f'Expected {expected_issuer}.'
+            ) from exception
+        except InvalidIssuerError as exception:
+            raise InvalidTokenError(
+                'The provided token has an incorrect "iss" (issuer) claim. '
+                f'Expected {expected_issuer}.'
+            ) from exception
+        except ExpiredSignatureError as exception:
+            raise ExpiredTokenError(
+                'The provided token has expired.'
+            ) from exception
+        except jwt.InvalidTokenError as exception:
+            raise InvalidTokenError(
+                f'Decoding token failed. Error: {exception}'
+            ) from exception
+
+        sub_claim = payload.get('sub')
+        if not isinstance(sub_claim, str) or not sub_claim:
+            raise InvalidTokenError(
+                'The provided token has an incorrect "sub" (subject) claim. '
+                'Expected a non-empty string.'
+            )
+
+        return payload
+
+
+class _Validators:
+    """A collection of data validation utilities.
+
+    Methods provided in this class raise ``ValueErrors`` if any validations fail.
+    """
+
+    @classmethod
+    def check_string(cls, label: str, value: Any):
+        """Checks if the given value is a string."""
+        if not isinstance(value, str) or not value:
+            raise ValueError(f'{label} must be a non-empty string.')
+
+# Firebase Phone Number Verification Errors
+class InvalidTokenError(exceptions.InvalidArgumentError):
+    """Raised when a Firebase Phone Number Verification token is invalid."""
+
+    def __init__(self, message, cause=None, http_response=None):
+        exceptions.InvalidArgumentError.__init__(self, message, cause, http_response)
+
+class ExpiredTokenError(InvalidTokenError):
+    """Raised when a Firebase Phone Number Verification token is expired."""
+
+    def __init__(self, message, cause=None, http_response=None):
+        InvalidTokenError.__init__(self, message, cause, http_response)
diff --git a/integration/emulators/.gitignore b/integration/emulators/.gitignore
new file mode 100644
index 000000000..b17f63107
--- /dev/null
+++ b/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
diff --git a/integration/emulators/firebase.json b/integration/emulators/firebase.json
new file mode 100644
index 000000000..a7b727c4d
--- /dev/null
+++ b/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"
+    }
+  ]
+}
diff --git a/integration/emulators/functions/.gitignore b/integration/emulators/functions/.gitignore
new file mode 100644
index 000000000..1609bab70
--- /dev/null
+++ b/integration/emulators/functions/.gitignore
@@ -0,0 +1,6 @@
+# Python bytecode
+__pycache__/
+
+# Python virtual environment
+venv/
+*.local
diff --git a/integration/emulators/functions/main.py b/integration/emulators/functions/main.py
new file mode 100644
index 000000000..6cd2c5766
--- /dev/null
+++ b/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
diff --git a/integration/emulators/functions/requirements.txt b/integration/emulators/functions/requirements.txt
new file mode 100644
index 000000000..6bbab42f8
--- /dev/null
+++ b/integration/emulators/functions/requirements.txt
@@ -0,0 +1 @@
+firebase_functions~=0.4.1
diff --git a/integration/test_auth.py b/integration/test_auth.py
index 7f4725dfe..b36063d19 100644
--- a/integration/test_auth.py
+++ b/integration/test_auth.py
@@ -724,6 +724,19 @@ def test_email_sign_in_with_settings(new_user_email_unverified, api_key):
     assert id_token is not None and len(id_token) > 0
     assert auth.get_user(new_user_email_unverified.uid).email_verified
 
+def test_auth_error_parse(new_user_email_unverified):
+    action_code_settings = auth.ActionCodeSettings(
+        ACTION_LINK_CONTINUE_URL, handle_code_in_app=True, link_domain="cool.link")
+    with pytest.raises(auth.InvalidHostingLinkDomainError) as excinfo:
+        auth.generate_sign_in_with_email_link(new_user_email_unverified.email,
+                                            action_code_settings=action_code_settings)
+    assert str(excinfo.value) == ('The provided hosting link domain is not configured in Firebase '
+                                  'Hosting or is not owned by the current project '
+                                  '(INVALID_HOSTING_LINK_DOMAIN). The provided hosting link '
+                                  'domain is not configured in Firebase Hosting or is not owned '
+                                  'by the current project. This cannot be a default hosting domain '
+                                  '(web.app or firebaseapp.com).')
+
 
 @pytest.fixture(scope='module')
 def oidc_provider():
diff --git a/integration/test_functions.py b/integration/test_functions.py
index 606798436..fc972f9e5 100644
--- a/integration/test_functions.py
+++ b/integration/test_functions.py
@@ -14,17 +14,34 @@
 
 """Integration tests for firebase_admin.functions module."""
 
+import os
 import pytest
 
 import firebase_admin
 from firebase_admin import functions
+from firebase_admin import _utils
 from integration import conftest
 
 
+_DEFAULT_DATA = {'data': {'city': 'Seattle'}}
+def integration_conf(request):
+    host_override = os.environ.get('CLOUD_TASKS_EMULATOR_HOST')
+    if host_override:
+        return _utils.EmulatorAdminCredentials(), 'fake-project-id'
+
+    return conftest.integration_conf(request)
+
 @pytest.fixture(scope='module')
 def app(request):
-    cred, _ = conftest.integration_conf(request)
-    return firebase_admin.initialize_app(cred, name='integration-functions')
+    cred, project_id = integration_conf(request)
+    return firebase_admin.initialize_app(
+        cred, options={'projectId': project_id}, name='integration-functions')
+
+@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
 
 
 class TestFunctions:
@@ -41,16 +58,31 @@ class TestFunctions:
     ]
 
     @pytest.mark.parametrize('task_queue_params', _TEST_FUNCTIONS_PARAMS)
-    def test_task_queue(self, task_queue_params):
-        queue = functions.task_queue(**task_queue_params)
-        assert queue is not None
-        assert callable(queue.enqueue)
-        assert callable(queue.delete)
-
-    @pytest.mark.parametrize('task_queue_params', _TEST_FUNCTIONS_PARAMS)
-    def test_task_queue_app(self, task_queue_params, app):
+    def test_task_queue(self, task_queue_params, app):
         assert app.name == 'integration-functions'
         queue = functions.task_queue(**task_queue_params, app=app)
         assert queue is not None
         assert callable(queue.enqueue)
         assert callable(queue.delete)
+
+    def test_task_enqueue(self, app):
+        queue = functions.task_queue('testTaskQueue', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+        assert task_id is not None
+
+    @pytest.mark.skipif(
+        os.environ.get('CLOUD_TASKS_EMULATOR_HOST') is not None,
+        reason="Skipping test_task_delete against emulator due to bug in firebase-tools"
+    )
+    def test_task_delete(self, app):
+        # Skip this test against the emulator since tasks can't be delayed there to verify deletion
+        # See: https://github.com/firebase/firebase-tools/issues/8254
+        task_options = functions.TaskOptions(schedule_delay_seconds=60)
+        queue = functions.task_queue('testTaskQueue', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA, task_options)
+        assert task_id is not None
+        queue.delete(task_id)
+        # We don't have a way to check the contents of the queue so we check that the deleted
+        # task is not found using the delete method again.
+        with pytest.raises(firebase_admin.exceptions.NotFoundError):
+            queue.delete(task_id)
diff --git a/requirements.txt b/requirements.txt
index c68d71a0f..0bbfbe183 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,11 +1,11 @@
 astroid == 3.3.11
-pylint == 3.3.7
+pylint == 3.3.9
 pytest >= 8.2.2
 pytest-cov >= 2.4.0
 pytest-localserver >= 0.4.1
 pytest-asyncio >= 0.26.0
 pytest-mock >= 3.6.1
-respx == 0.22.0
+respx == 0.23.1
 
 cachecontrol >= 0.14.3
 google-api-core[grpc] >= 2.25.1, < 3.0.0dev; platform.python_implementation != 'PyPy'
diff --git a/setup.cfg b/setup.cfg
index 32e00676b..4c6cf8d8f 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,4 +1,3 @@
 [tool:pytest]
 testpaths = tests
 asyncio_default_test_loop_scope = class
-asyncio_default_fixture_loop_scope = None
diff --git a/snippets/auth/index.py b/snippets/auth/index.py
index 6a509b8f5..656137dba 100644
--- a/snippets/auth/index.py
+++ b/snippets/auth/index.py
@@ -770,7 +770,7 @@ def get_tenant(tenant_id):
 	# [START get_tenant]
     tenant = tenant_mgt.get_tenant(tenant_id)
 
-    print('Retreieved tenant:', tenant.tenant_id)
+    print('Retrieved tenant:', tenant.tenant_id)
 	# [END get_tenant]
 
 def create_tenant():
diff --git a/tests/test_functions.py b/tests/test_functions.py
index 52e92c1b2..0f766767a 100644
--- a/tests/test_functions.py
+++ b/tests/test_functions.py
@@ -44,13 +44,14 @@ def setup_class(cls):
     def teardown_class(cls):
         testutils.cleanup_apps()
 
-    def _instrument_functions_service(self, app=None, status=200, payload=_DEFAULT_RESPONSE):
+    def _instrument_functions_service(
+        self, app=None, status=200, payload=_DEFAULT_RESPONSE, mounted_url=_CLOUD_TASKS_URL):
         if not app:
             app = firebase_admin.get_app()
         functions_service = functions._get_functions_service(app)
         recorder = []
         functions_service._http_client.session.mount(
-            _CLOUD_TASKS_URL,
+            mounted_url,
             testutils.MockAdapter(payload, status, recorder))
         return functions_service, recorder
 
@@ -124,6 +125,10 @@ def test_task_enqueue(self):
         assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
         assert task_id == 'test-task-id'
 
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {'serviceAccountEmail': 'mock-email'}
+        assert task['httpRequest']['headers'] == {'Content-Type': 'application/json'}
+
     def test_task_enqueue_with_extension(self):
         resource_name = (
             'projects/test-project/locations/us-central1/queues/'
@@ -142,6 +147,59 @@ def test_task_enqueue_with_extension(self):
         assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
         assert task_id == 'test-task-id'
 
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {'serviceAccountEmail': 'mock-email'}
+        assert task['httpRequest']['headers'] == {'Content-Type': 'application/json'}
+
+    def test_task_enqueue_compute_engine(self):
+        app = firebase_admin.initialize_app(
+            testutils.MockComputeEngineCredential(),
+            options={'projectId': 'test-project'},
+            name='test-project-gce')
+        _, recorder = self._instrument_functions_service(app)
+        queue = functions.task_queue('test-function-name', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+        assert len(recorder) == 1
+        assert recorder[0].method == 'POST'
+        assert recorder[0].url == _DEFAULT_REQUEST_URL
+        assert recorder[0].headers['Content-Type'] == 'application/json'
+        assert recorder[0].headers['Authorization'] == 'Bearer mock-compute-engine-token'
+        expected_metrics_header = _utils.get_metrics_header() + ' mock-gce-cred-metric-tag'
+        assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
+        assert task_id == 'test-task-id'
+
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {'serviceAccountEmail': 'mock-gce-email'}
+        assert task['httpRequest']['headers'] == {'Content-Type': 'application/json'}
+
+    def test_task_enqueue_with_extension_compute_engine(self):
+        resource_name = (
+            'projects/test-project/locations/us-central1/queues/'
+            'ext-test-extension-id-test-function-name/tasks'
+        )
+        extension_response = json.dumps({'name': resource_name + '/test-task-id'})
+        app = firebase_admin.initialize_app(
+            testutils.MockComputeEngineCredential(),
+            options={'projectId': 'test-project'},
+            name='test-project-gce-extensions')
+        _, recorder = self._instrument_functions_service(app, payload=extension_response)
+        queue = functions.task_queue('test-function-name', 'test-extension-id', app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+        assert len(recorder) == 1
+        assert recorder[0].method == 'POST'
+        assert recorder[0].url == _CLOUD_TASKS_URL + resource_name
+        assert recorder[0].headers['Content-Type'] == 'application/json'
+        assert recorder[0].headers['Authorization'] == 'Bearer mock-compute-engine-token'
+        expected_metrics_header = _utils.get_metrics_header() + ' mock-gce-cred-metric-tag'
+        assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
+        assert task_id == 'test-task-id'
+
+        task = json.loads(recorder[0].body.decode())['task']
+        assert 'oidcToken' not in task['httpRequest']
+        assert task['httpRequest']['headers'] == {
+            'Content-Type': 'application/json',
+            'Authorization': 'Bearer mock-compute-engine-token'}
+
     def test_task_delete(self):
         _, recorder = self._instrument_functions_service()
         queue = functions.task_queue('test-function-name')
@@ -152,6 +210,58 @@ def test_task_delete(self):
         expected_metrics_header = _utils.get_metrics_header() + ' mock-cred-metric-tag'
         assert recorder[0].headers['x-goog-api-client'] == expected_metrics_header
 
+    def test_task_enqueue_with_emulator_host(self, monkeypatch):
+        emulator_host = 'localhost:8124'
+        emulator_url = f'http://{emulator_host}/'
+        request_url = emulator_url + _DEFAULT_TASK_PATH.replace('/tasks/test-task-id', '/tasks')
+
+        monkeypatch.setenv('CLOUD_TASKS_EMULATOR_HOST', emulator_host)
+        app = firebase_admin.initialize_app(
+            _utils.EmulatorAdminCredentials(), {'projectId': 'test-project'}, name='emulator-app')
+
+        expected_task_name = (
+            '/projects/test-project/locations/us-central1'
+            '/queues/test-function-name/tasks/test-task-id'
+        )
+        expected_response = json.dumps({'task': {'name': expected_task_name}})
+        _, recorder = self._instrument_functions_service(
+            app, payload=expected_response, mounted_url=emulator_url)
+
+        queue = functions.task_queue('test-function-name', app=app)
+        task_id = queue.enqueue(_DEFAULT_DATA)
+
+        assert len(recorder) == 1
+        assert recorder[0].method == 'POST'
+        assert recorder[0].url == request_url
+        assert recorder[0].headers['Content-Type'] == 'application/json'
+
+        task = json.loads(recorder[0].body.decode())['task']
+        assert task['httpRequest']['oidcToken'] == {
+            'serviceAccountEmail': 'emulated-service-acct@email.com'
+        }
+        assert task_id == 'test-task-id'
+
+    def test_task_enqueue_without_emulator_host_error(self, monkeypatch):
+        app = firebase_admin.initialize_app(
+            _utils.EmulatorAdminCredentials(),
+            {'projectId': 'test-project'}, name='no-emulator-app')
+
+        _, recorder = self._instrument_functions_service(app)
+        monkeypatch.delenv('CLOUD_TASKS_EMULATOR_HOST', raising=False)
+        queue = functions.task_queue('test-function-name', app=app)
+        with pytest.raises(ValueError) as excinfo:
+            queue.enqueue(_DEFAULT_DATA)
+        assert "Failed to determine service account" in str(excinfo.value)
+        assert len(recorder) == 0
+
+    def test_get_emulator_url_invalid_format(self, monkeypatch):
+        monkeypatch.setenv('CLOUD_TASKS_EMULATOR_HOST', 'http://localhost:8124')
+        app = firebase_admin.initialize_app(
+            testutils.MockCredential(), {'projectId': 'test-project'}, name='invalid-host-app')
+        with pytest.raises(ValueError) as excinfo:
+            functions.task_queue('test-function-name', app=app)
+        assert 'Invalid CLOUD_TASKS_EMULATOR_HOST' in str(excinfo.value)
+
 class TestTaskQueueOptions:
 
     _DEFAULT_TASK_OPTS = {'schedule_delay_seconds': None, 'schedule_time': None, \
@@ -202,13 +312,13 @@ def test_task_options_delay_seconds(self):
         assert len(recorder) == 1
         task = json.loads(recorder[0].body.decode())['task']
 
-        task_schedule_time = datetime.fromisoformat(task['schedule_time'].replace('Z', '+00:00'))
+        task_schedule_time = datetime.fromisoformat(task['scheduleTime'].replace('Z', '+00:00'))
         delta = abs(task_schedule_time - expected_schedule_time)
         assert delta <= timedelta(seconds=1)
 
-        assert task['dispatch_deadline'] == '200s'
-        assert task['http_request']['headers']['x-test-header'] == 'test-header-value'
-        assert task['http_request']['url'] in ['http://google.com', 'https://google.com']
+        assert task['dispatchDeadline'] == '200s'
+        assert task['httpRequest']['headers']['x-test-header'] == 'test-header-value'
+        assert task['httpRequest']['url'] in ['http://google.com', 'https://google.com']
         assert task['name'] == _DEFAULT_TASK_PATH
 
     def test_task_options_utc_time(self):
@@ -230,12 +340,12 @@ def test_task_options_utc_time(self):
         assert len(recorder) == 1
         task = json.loads(recorder[0].body.decode())['task']
 
-        task_schedule_time = datetime.fromisoformat(task['schedule_time'].replace('Z', '+00:00'))
+        task_schedule_time = datetime.fromisoformat(task['scheduleTime'].replace('Z', '+00:00'))
         assert task_schedule_time == expected_schedule_time
 
-        assert task['dispatch_deadline'] == '200s'
-        assert task['http_request']['headers']['x-test-header'] == 'test-header-value'
-        assert task['http_request']['url'] in ['http://google.com', 'https://google.com']
+        assert task['dispatchDeadline'] == '200s'
+        assert task['httpRequest']['headers']['x-test-header'] == 'test-header-value'
+        assert task['httpRequest']['url'] in ['http://google.com', 'https://google.com']
         assert task['name'] == _DEFAULT_TASK_PATH
 
     def test_schedule_set_twice_error(self):
@@ -247,7 +357,7 @@ def test_schedule_set_twice_error(self):
             queue.enqueue(_DEFAULT_DATA, opts)
         assert len(recorder) == 0
         assert str(excinfo.value) == \
-            '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.'
 
 
     @pytest.mark.parametrize('schedule_time', [
diff --git a/tests/test_messaging.py b/tests/test_messaging.py
index 9fa30fef9..b30790f14 100644
--- a/tests/test_messaging.py
+++ b/tests/test_messaging.py
@@ -335,6 +335,18 @@ def test_invalid_direct_boot_ok(self, data):
             check_encoding(messaging.Message(
                 topic='topic', android=messaging.AndroidConfig(direct_boot_ok=data)))
 
+    @pytest.mark.parametrize('data', NON_BOOL_ARGS)
+    def test_invalid_bandwidth_constrained_ok(self, data):
+        with pytest.raises(ValueError):
+            check_encoding(messaging.Message(
+                topic='topic', android=messaging.AndroidConfig(bandwidth_constrained_ok=data)))
+
+    @pytest.mark.parametrize('data', NON_BOOL_ARGS)
+    def test_invalid_restricted_satellite_ok(self, data):
+        with pytest.raises(ValueError):
+            check_encoding(messaging.Message(
+                topic='topic', android=messaging.AndroidConfig(restricted_satellite_ok=data)))
+
 
     def test_android_config(self):
         msg = messaging.Message(
@@ -347,6 +359,8 @@ def test_android_config(self):
                 data={'k1': 'v1', 'k2': 'v2'},
                 fcm_options=messaging.AndroidFCMOptions('analytics_label_v1'),
                 direct_boot_ok=True,
+                bandwidth_constrained_ok=True,
+                restricted_satellite_ok=True,
             )
         )
         expected = {
@@ -364,6 +378,8 @@ def test_android_config(self):
                     'analytics_label': 'analytics_label_v1',
                 },
                 'direct_boot_ok': True,
+                'bandwidth_constrained_ok': True,
+                'restricted_satellite_ok': True,
             },
         }
         check_encoding(msg, expected)
diff --git a/tests/test_phone_number_verification.py b/tests/test_phone_number_verification.py
new file mode 100644
index 000000000..71f8c2ded
--- /dev/null
+++ b/tests/test_phone_number_verification.py
@@ -0,0 +1,311 @@
+# 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.fpnv module."""
+
+import base64
+import time
+from unittest import mock
+from unittest.mock import patch
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric import ec
+
+import firebase_admin
+from firebase_admin import phone_number_verification as fpnv
+from tests import testutils
+
+# Mock Data
+_PROJECT_ID = 'mock-project-id'
+_EXP_TIMESTAMP = 2000000000
+_ISSUER = f'https://fpnv.googleapis.com/projects/{_PROJECT_ID}'
+_PHONE_NUMBER = '+1234567890'
+_PUBLIC_KEY = 'test-public-key'  # In real tests, use the corresponding public key
+_ALGORITHM = 'ES256'
+_KEY_ID = 'test-key-id'
+_TYPE = 'JWT'
+
+_MOCK_PAYLOAD = {
+    'iss': _ISSUER,
+    'sub': _PHONE_NUMBER,
+    'aud': [_ISSUER],
+    'exp': _EXP_TIMESTAMP,
+    'iat': _EXP_TIMESTAMP - 3600,
+    "other": 'other'
+}
+
+
+
+
+class TestCommon:
+    @classmethod
+    def setup_class(cls):
+        cred = testutils.MockCredential()
+        firebase_admin.initialize_app(cred, {'projectId': _PROJECT_ID})
+
+    @classmethod
+    def teardown_class(cls):
+        testutils.cleanup_apps()
+
+
+class TestFpnvToken:
+    def test_properties(self):
+        token = fpnv.PhoneNumberVerificationToken(_MOCK_PAYLOAD)
+
+        assert token.phone_number == _PHONE_NUMBER
+        assert token.sub == _PHONE_NUMBER
+        assert token.issuer == _ISSUER
+        assert token.audience == [_ISSUER]
+        expected_claims = _MOCK_PAYLOAD.copy()
+        expected_claims['phone_number'] = _PHONE_NUMBER
+        assert token.claims == expected_claims
+        assert token['other'] == _MOCK_PAYLOAD['other']
+
+
+class TestVerifyToken(TestCommon):
+
+    def test_no_project_id(self):
+        app = firebase_admin.initialize_app(testutils.MockCredential(), name='no_project_id')
+        app.credential.get_credential().project_id = None
+        with pytest.raises(
+            ValueError,
+            match='Project ID is required for Firebase Phone Number Verification'
+        ):
+            fpnv.verify_token('token', app=app)
+
+    def test_verify_token_with_real_crypto(self):
+        """Verifies a token signed with a real ES256 key pair.
+
+        Mocking only the JWKS endpoint.
+        This ensures the cryptographic verification logic is functioning correctly.
+        """
+        # Generate a real ES256 key pair
+        private_key = ec.generate_private_key(ec.SECP256R1())
+        public_key = private_key.public_key()
+
+        # Create the JWK representation of the public key (for the mock endpoint)
+        # Note: Retrieving numbers from the key involves cryptography primitives
+        public_numbers = public_key.public_numbers()
+
+        def to_b64url(b_data):
+            return base64.urlsafe_b64encode(b_data).rstrip(b'=').decode('utf-8')
+
+        jwk = {
+            "kty": "EC",
+            "use": "sig",
+            "alg": _ALGORITHM,
+            "kid": _KEY_ID,
+            "crv": "P-256",
+            "x": to_b64url(public_numbers.x.to_bytes(32, 'big')),
+            "y": to_b64url(public_numbers.y.to_bytes(32, 'big')),
+        }
+        now = int(time.time())
+        payload = {
+            'iss': _ISSUER,
+            'aud': [_ISSUER],
+            'iat': now,
+            'exp': now + 3600,
+            'sub': _PHONE_NUMBER
+        }
+
+        # Sign using the private key object directly (PyJWT supports this)
+        token = jwt.encode(
+            payload,
+            private_key,
+            algorithm=_ALGORITHM,
+            headers={'alg': _ALGORITHM, 'typ': _TYPE, 'kid': _KEY_ID},
+        )
+
+        # Mock PyJWKClient fetch_data
+        with patch('jwt.PyJWKClient.fetch_data') as mock_fetch:
+            mock_fetch.return_value = {'keys': [jwk]}
+
+            app = firebase_admin.get_app()
+            decoded_token = fpnv.verify_token(token, app)
+
+            assert decoded_token['sub'] == _PHONE_NUMBER
+            assert _ISSUER in decoded_token['aud']
+            assert decoded_token.phone_number == decoded_token['sub']
+            # Test convenience dictionary lookup
+            assert decoded_token['phone_number'] == _PHONE_NUMBER
+
+    def test_verify_token_module_level_delegation(self):
+        """Verifies module-level verify_token delegates correctly."""
+        with patch(
+            'firebase_admin.phone_number_verification._FpnvService.verify_token'
+        ) as mock_verify:
+            mock_verify.return_value = 'mock-result'
+            res = fpnv.verify_token('some-token')
+            assert res == 'mock-result'
+            mock_verify.assert_called_once_with('some-token')
+
+    @mock.patch('jwt.PyJWKClient')
+    @mock.patch('jwt.decode')
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_success(self, mock_header, mock_decode, mock_jwks_cls):
+        token_str = 'valid.token.string'
+        # Mock Header
+        mock_header.return_value = {'kid': 'key1', 'typ': 'JWT', 'alg': 'ES256'}
+
+        # Mock Signing Key
+        mock_jwks_instance = mock_jwks_cls.return_value
+        mock_signing_key = mock.Mock()
+        mock_signing_key.key = _PUBLIC_KEY
+        mock_jwks_instance.get_signing_key_from_jwt.return_value = mock_signing_key
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        service._verifier._jwks_client = mock_jwks_instance
+
+        mock_decode.return_value = _MOCK_PAYLOAD
+
+        # Execute
+        token = fpnv.verify_token(token_str)
+
+        # Verify
+        assert isinstance(token, fpnv.PhoneNumberVerificationToken)
+        assert token.phone_number == _PHONE_NUMBER
+
+        mock_header.assert_called_with(token_str)
+        mock_jwks_instance.get_signing_key_from_jwt.assert_called_with(token_str)
+        mock_decode.assert_called_with(
+            token_str,
+            _PUBLIC_KEY,
+            algorithms=['ES256'],
+            audience=_ISSUER,
+            issuer=_ISSUER
+        )
+
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_no_name(self, mock_header):
+        app = firebase_admin.get_app()
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+        with pytest.raises(ValueError, match="must be a non-empty string"):
+            fpnv.verify_token('', app=app)
+
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_no_kid(self, mock_header):
+        app = firebase_admin.get_app()
+        mock_header.return_value = {'typ': 'JWT', 'alg': 'ES256'}  # Missing kid
+        with pytest.raises(fpnv.InvalidTokenError, match="Token has no 'kid' claim."):
+            fpnv.verify_token('token', app=app)
+
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_wrong_alg(self, mock_header):
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'RS256'}  # Wrong alg
+        with pytest.raises(fpnv.InvalidTokenError, match="incorrect alg"):
+            fpnv.verify_token('token')
+
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_wrong_typ(self, mock_header):
+        mock_header.return_value = {'kid': 'k', 'typ': 'WRONG', 'alg': 'ES256'} # wrong typ
+        with pytest.raises(fpnv.InvalidTokenError, match="incorrect type header"):
+            fpnv.verify_token('token')
+
+    def test_verify_token_jwk_error(self):
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        jwks_client = service._verifier._jwks_client
+
+        # Mock the method on the existing instance
+        with mock.patch.object(jwks_client, 'get_signing_key_from_jwt') as mock_method:
+            mock_method.side_effect = jwt.PyJWKClientError("Key not found")
+
+            # Mock header is still needed if _get_signing_key calls it before the client
+            with mock.patch('jwt.get_unverified_header') as mock_header:
+                mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+
+                with pytest.raises(
+                    fpnv.InvalidTokenError,
+                    match="Verifying phone number verification token failed"
+                ):
+                    fpnv.verify_token('token')
+
+    @mock.patch('jwt.PyJWKClient')
+    @mock.patch('jwt.decode')
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_expired(self, mock_header, mock_decode, mock_jwks_cls):
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+        mock_jwks_instance = mock_jwks_cls.return_value
+        mock_jwks_instance.get_signing_key_from_jwt.return_value.key = _PUBLIC_KEY
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        service._verifier._jwks_client = mock_jwks_instance
+
+        # Simulate ExpiredSignatureError
+        mock_decode.side_effect = jwt.ExpiredSignatureError("Expired")
+
+        with pytest.raises(fpnv.ExpiredTokenError, match="token has expired"):
+            fpnv.verify_token('token')
+
+    @mock.patch('jwt.PyJWKClient')
+    @mock.patch('jwt.decode')
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_invalid_signature(self, mock_header, mock_decode, mock_jwks_cls):
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+        mock_jwks_instance = mock_jwks_cls.return_value
+        mock_jwks_instance.get_signing_key_from_jwt.return_value.key = _PUBLIC_KEY
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        service._verifier._jwks_client = mock_jwks_instance
+
+        # Simulate InvalidSignatureError
+        mock_decode.side_effect = jwt.InvalidSignatureError("Wrong Signature")
+
+        with pytest.raises(fpnv.InvalidTokenError, match="invalid signature"):
+            fpnv.verify_token('token')
+
+    @mock.patch('jwt.PyJWKClient')
+    @mock.patch('jwt.decode')
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_invalid_audience(self, mock_header, mock_decode, mock_jwks_cls):
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+        mock_jwks_instance = mock_jwks_cls.return_value
+        mock_jwks_instance.get_signing_key_from_jwt.return_value.key = _PUBLIC_KEY
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        service._verifier._jwks_client = mock_jwks_instance
+
+        # Simulate InvalidAudienceError
+        mock_decode.side_effect = jwt.InvalidAudienceError("Wrong Aud")
+
+        with pytest.raises(fpnv.InvalidTokenError, match="incorrect \"aud\""):
+            fpnv.verify_token('token')
+
+    @mock.patch('jwt.PyJWKClient')
+    @mock.patch('jwt.decode')
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_invalid_issuer(self, mock_header, mock_decode, mock_jwks_cls):
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+        mock_jwks_instance = mock_jwks_cls.return_value
+        mock_jwks_instance.get_signing_key_from_jwt.return_value.key = _PUBLIC_KEY
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        service._verifier._jwks_client = mock_jwks_instance
+
+        # Simulate InvalidIssuerError
+        mock_decode.side_effect = jwt.InvalidIssuerError("Wrong Iss")
+
+        with pytest.raises(fpnv.InvalidTokenError, match="incorrect \"iss\""):
+            fpnv.verify_token('token')
+
+    @mock.patch('jwt.PyJWKClient')
+    @mock.patch('jwt.decode')
+    @mock.patch('jwt.get_unverified_header')
+    def test_verify_token_invalid_token(self, mock_header, mock_decode, mock_jwks_cls):
+        mock_header.return_value = {'kid': 'k', 'typ': 'JWT', 'alg': 'ES256'}
+        mock_jwks_instance = mock_jwks_cls.return_value
+        mock_jwks_instance.get_signing_key_from_jwt.return_value.key = _PUBLIC_KEY
+        service = fpnv._get_fpnv_service(firebase_admin.get_app())
+        service._verifier._jwks_client = mock_jwks_instance
+
+        # Simulate InvalidTokenError
+        mock_decode.side_effect = jwt.InvalidTokenError("Decoding FPNV token failed")
+
+        with pytest.raises(fpnv.InvalidTokenError, match="Decoding FPNV token failed"):
+            fpnv.verify_token('token')
diff --git a/tests/testutils.py b/tests/testutils.py
index 598a929b4..7546595af 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -116,12 +116,25 @@ def __call__(self, *args, **kwargs): # pylint: disable=arguments-differ
 # pylint: disable=abstract-method
 class MockGoogleCredential(credentials.Credentials):
     """A mock Google authentication credential."""
+
+    def __init__(self):
+        super().__init__()
+        self.token = None
+        self._service_account_email = None
+        self._token_state = credentials.TokenState.INVALID
+
     def refresh(self, request):
         self.token = 'mock-token'
+        self._service_account_email = 'mock-email'
+        self._token_state = credentials.TokenState.FRESH
+
+    @property
+    def token_state(self):
+        return self._token_state
 
     @property
     def service_account_email(self):
-        return 'mock-email'
+        return self._service_account_email
 
     # Simulate x-goog-api-client modification in credential refresh
     def _metric_header_for_usage(self):
@@ -139,8 +152,24 @@ def get_credential(self):
 
 class MockGoogleComputeEngineCredential(compute_engine.Credentials):
     """A mock Compute Engine credential"""
+
+    def __init__(self):
+        super().__init__()
+        self.token = None
+        self._service_account_email = None
+        self._token_state = credentials.TokenState.INVALID
+
     def refresh(self, request):
         self.token = 'mock-compute-engine-token'
+        self._service_account_email = 'mock-gce-email'
+        self._token_state = credentials.TokenState.FRESH
+
+    @property
+    def token_state(self):
+        return self._token_state
+
+    def _metric_header_for_usage(self):
+        return 'mock-gce-cred-metric-tag'
 
 class MockComputeEngineCredential(firebase_admin.credentials.Base):
     """A mock Firebase credential implementation."""