From 05399b588035be69130924387365b6f5a0fe13c9 Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Wed, 26 Aug 2026 14:36:11 -0400 Subject: [PATCH 1/4] feat(fcm): Migrate topic management to FCM v1 API --- firebase_admin/messaging.py | 319 +++++++++++++++++++++++++++++++- tests/test_messaging.py | 350 ++++++++++++++++++++++++++++++++++-- 2 files changed, 648 insertions(+), 21 deletions(-) diff --git a/firebase_admin/messaging.py b/firebase_admin/messaging.py index 93ecfda1..d266f6f9 100644 --- a/firebase_admin/messaging.py +++ b/firebase_admin/messaging.py @@ -15,14 +15,17 @@ """Firebase Cloud Messaging module.""" from __future__ import annotations -from typing import Any, Callable, Dict, List, Optional, cast +import asyncio import concurrent.futures import json -import asyncio import logging +import re +from typing import Any, Callable, Dict, List, Optional, cast +import urllib.parse import warnings -import requests + import httpx +import requests import firebase_admin from firebase_admin import ( @@ -73,7 +76,11 @@ 'send_each_for_multicast', 'send_each_for_multicast_async', 'subscribe_to_topic', + 'subscribe_to_topic_async', + 'subscribe_to_topic_legacy', 'unsubscribe_from_topic', + 'unsubscribe_from_topic_async', + 'unsubscribe_from_topic_legacy', ] @@ -255,6 +262,44 @@ def send_each_for_multicast(multicast_message, dry_run=False, app=None): def subscribe_to_topic(tokens, topic, app=None): """Subscribes a list of registration tokens to an FCM topic. + Args: + tokens: A non-empty list of device registration tokens. List may not have more than 1000 + elements. + topic: Name of the topic to subscribe to. May contain the ``/topics/`` prefix. + app: An App instance (optional). + + Returns: + TopicManagementResponse: A ``TopicManagementResponse`` instance. + + Raises: + FirebaseError: If an error occurs while communicating with the FCM service. + ValueError: If the input arguments are invalid. + """ + return _get_messaging_service(app).subscribe_to_topic(tokens, topic) + +async def subscribe_to_topic_async(tokens, topic, app=None): + """Subscribes a list of registration tokens to an FCM topic asynchronously. + + Args: + tokens: A non-empty list of device registration tokens. List may not have more than 1000 + elements. + topic: Name of the topic to subscribe to. May contain the ``/topics/`` prefix. + app: An App instance (optional). + + Returns: + TopicManagementResponse: A ``TopicManagementResponse`` instance. + + Raises: + FirebaseError: If an error occurs while communicating with the FCM service. + ValueError: If the input arguments are invalid. + """ + return await _get_messaging_service(app).subscribe_to_topic_async(tokens, topic) + +def subscribe_to_topic_legacy(tokens, topic, app=None): + """Subscribes a list of registration tokens to an FCM topic using the legacy Instance ID API. + + subscribe_to_topic_legacy is deprecated. Use subscribe_to_topic instead. + Args: tokens: A non-empty list of device registration tokens. List may not have more than 1000 elements. @@ -268,12 +313,55 @@ def subscribe_to_topic(tokens, topic, app=None): FirebaseError: If an error occurs while communicating with instance ID service. ValueError: If the input arguments are invalid. """ + warnings.warn( + 'subscribe_to_topic_legacy is deprecated. Use subscribe_to_topic instead.', + DeprecationWarning, + stacklevel=2) return _get_messaging_service(app).make_topic_management_request( tokens, topic, 'iid/v1:batchAdd') def unsubscribe_from_topic(tokens, topic, app=None): """Unsubscribes a list of registration tokens from an FCM topic. + Args: + tokens: A non-empty list of device registration tokens. List may not have more than 1000 + elements. + topic: Name of the topic to unsubscribe from. May contain the ``/topics/`` prefix. + app: An App instance (optional). + + Returns: + TopicManagementResponse: A ``TopicManagementResponse`` instance. + + Raises: + FirebaseError: If an error occurs while communicating with the FCM service. + ValueError: If the input arguments are invalid. + """ + return _get_messaging_service(app).unsubscribe_from_topic(tokens, topic) + +async def unsubscribe_from_topic_async(tokens, topic, app=None): + """Unsubscribes a list of registration tokens from an FCM topic asynchronously. + + Args: + tokens: A non-empty list of device registration tokens. List may not have more than 1000 + elements. + topic: Name of the topic to unsubscribe from. May contain the ``/topics/`` prefix. + app: An App instance (optional). + + Returns: + TopicManagementResponse: A ``TopicManagementResponse`` instance. + + Raises: + FirebaseError: If an error occurs while communicating with the FCM service. + ValueError: If the input arguments are invalid. + """ + return await _get_messaging_service(app).unsubscribe_from_topic_async(tokens, topic) + +def unsubscribe_from_topic_legacy(tokens, topic, app=None): + """Unsubscribes a list of registration tokens from an FCM topic using the legacy + Instance ID API. + + unsubscribe_from_topic_legacy is deprecated. Use unsubscribe_from_topic instead. + Args: tokens: A non-empty list of device registration tokens. List may not have more than 1000 elements. @@ -287,6 +375,10 @@ def unsubscribe_from_topic(tokens, topic, app=None): FirebaseError: If an error occurs while communicating with instance ID service. ValueError: If the input arguments are invalid. """ + warnings.warn( + 'unsubscribe_from_topic_legacy is deprecated. Use unsubscribe_from_topic instead.', + DeprecationWarning, + stacklevel=2) return _get_messaging_service(app).make_topic_management_request( tokens, topic, 'iid/v1:batchRemove') @@ -410,7 +502,9 @@ def __init__(self, app: App) -> None: 'Project ID is required to access Cloud Messaging service. Either set the ' 'projectId option, or use service account credentials. Alternatively, set the ' 'GOOGLE_CLOUD_PROJECT environment variable.') + self._project_id = project_id self._fcm_url = _MessagingService.FCM_URL.format(project_id) + self._fcm_topic_url = f'https://fcm.googleapis.com/v1/projects/{project_id}/registrations' self._fcm_headers = { 'X-GOOG-API-FORMAT-VERSION': '2', 'X-FIREBASE-CLIENT': f'fire-admin-python/{firebase_admin.__version__}', @@ -499,6 +593,225 @@ async def send_data(data): message=f'Unknown error while making remote service calls: {error}', cause=error) + def _validate_topic_management_args(self, tokens, topic): + """Validates and formats topic management arguments.""" + if isinstance(tokens, str): + tokens = [tokens] + if not isinstance(tokens, list) or not tokens: + raise ValueError('Tokens must be a string or a non-empty list of strings.') + invalid_str = [t for t in tokens if not isinstance(t, str) or not t] + if invalid_str: + raise ValueError('Tokens must be non-empty strings.') + if len(tokens) > 1000: + raise ValueError('tokens must not contain more than 1000 elements.') + + if not isinstance(topic, str) or not topic: + raise ValueError('Topic must be a non-empty string.') + topic_name = topic + if topic_name.startswith('/topics/'): + topic_name = topic_name[len('/topics/'):] + if not topic_name or not re.match(r'^[a-zA-Z0-9-_\.~%]+$', topic_name): + raise ValueError('Malformed topic name.') + + return tokens, topic_name + + def subscribe_to_topic(self, tokens, topic) -> TopicManagementResponse: + """Subscribes a list of registration tokens to an FCM topic via the FCM v1 API.""" + return self._make_topic_management_request_v1(tokens, topic, is_subscribe=True) + + def unsubscribe_from_topic(self, tokens, topic) -> TopicManagementResponse: + """Unsubscribes a list of registration tokens from an FCM topic via the FCM v1 API.""" + return self._make_topic_management_request_v1(tokens, topic, is_subscribe=False) + + async def subscribe_to_topic_async(self, tokens, topic) -> TopicManagementResponse: + """Subscribes a list of registration tokens to an FCM topic asynchronously + via the FCM v1 API.""" + return await self._make_topic_management_request_v1_async( + tokens, topic, is_subscribe=True) + + async def unsubscribe_from_topic_async(self, tokens, topic) -> TopicManagementResponse: + """Unsubscribes a list of registration tokens from an FCM topic asynchronously + via the FCM v1 API.""" + return await self._make_topic_management_request_v1_async( + tokens, topic, is_subscribe=False) + + def _make_topic_management_request_v1( + self, tokens, topic, is_subscribe: bool + ) -> TopicManagementResponse: + """Helper method that sends topic subscription requests via FCM v1 API.""" + tokens_list, topic_name = self._validate_topic_management_args(tokens, topic) + + def send_request(token: str): + encoded_token = urllib.parse.quote(token, safe='') + encoded_topic = urllib.parse.quote(topic_name, safe='') + base_url = f'{self._fcm_topic_url}/{encoded_token}/topicSubscriptions' + if is_subscribe: + url = f'{base_url}?topic_name={encoded_topic}' + method = 'post' + json_data = {} + else: + url = f'{base_url}/{encoded_topic}?allow_missing=true' + method = 'delete' + json_data = None + + try: + self._client.request( + method, + url=url, + headers=self._fcm_headers, + json=json_data, + ) + return {'success': True} + except requests.exceptions.RequestException as error: + return self._build_topic_subscription_result_from_requests_error( + error, is_subscribe) + + try: + with concurrent.futures.ThreadPoolExecutor( + max_workers=min(len(tokens_list), 100) + ) as executor: + results = list(executor.map(send_request, tokens_list)) + return self._parse_topic_management_results(results) + except Exception as error: + raise exceptions.UnknownError( + message=f'Unknown error while making remote service calls: {error}', + cause=error) + + async def _make_topic_management_request_v1_async( + self, tokens, topic, is_subscribe: bool + ) -> TopicManagementResponse: + """Helper method that sends topic subscription requests asynchronously via FCM v1 API.""" + tokens_list, topic_name = self._validate_topic_management_args(tokens, topic) + semaphore = asyncio.Semaphore(100) + + async def send_request_async(token: str): + encoded_token = urllib.parse.quote(token, safe='') + encoded_topic = urllib.parse.quote(topic_name, safe='') + base_url = f'{self._fcm_topic_url}/{encoded_token}/topicSubscriptions' + if is_subscribe: + url = f'{base_url}?topic_name={encoded_topic}' + method = 'post' + json_data = {} + else: + url = f'{base_url}/{encoded_topic}?allow_missing=true' + method = 'delete' + json_data = None + + async with semaphore: + try: + await self._async_client.request( + method, + url=url, + headers=self._fcm_headers, + json=json_data, + ) + return {'success': True} + except httpx.HTTPError as error: + return self._build_topic_subscription_result_from_httpx_error( + error, is_subscribe) + except requests.exceptions.RequestException as error: + return self._build_topic_subscription_result_from_requests_error( + error, is_subscribe) + + try: + results = await asyncio.gather(*[send_request_async(token) for token in tokens_list]) + return self._parse_topic_management_results(results) + except Exception as error: + raise exceptions.UnknownError( + message=f'Unknown error while making remote service calls: {error}', + cause=error) + + @classmethod + def _get_topic_error_code(cls, error_dict: dict, status_code: int) -> str: + """Extracts the error code for a topic subscription error response.""" + error_data = error_dict.get('error') + if isinstance(error_data, str) and error_data: + return error_data + if isinstance(error_data, dict): + details = error_data.get('details') + if isinstance(details, list): + fcm_error_type = 'type.googleapis.com/google.firebase.fcm.v1.FcmError' + for element in details: + if isinstance(element, dict) and element.get('@type') == fcm_error_type: + code = element.get('errorCode') + if code: + return code + status = error_data.get('status') + if status: + return status + message = error_data.get('message') + if message: + return message + + status_map = { + 400: 'INVALID_ARGUMENT', + 401: 'PERMISSION_DENIED', + 403: 'PERMISSION_DENIED', + 404: 'NOT_FOUND', + 429: 'RESOURCE_EXHAUSTED', + 500: 'INTERNAL', + 503: 'DEADLINE_EXCEEDED', + } + return status_map.get(status_code, 'UNKNOWN_ERROR') + + def _build_topic_subscription_result_from_requests_error(self, error, is_subscribe): + """Constructs a result dict from a requests error.""" + if error.response is not None: + if is_subscribe and error.response.status_code == 409: + return {'success': True} + error_dict = {} + try: + parsed = error.response.json() + if isinstance(parsed, dict): + error_dict = parsed + except ValueError: + pass + + error_data = error_dict.get('error') + if is_subscribe and isinstance(error_data, dict) and ( + error_data.get('status') == 'ALREADY_EXISTS' + ): + return {'success': True} + + error_code = self._get_topic_error_code(error_dict, error.response.status_code) + return {'success': False, 'error': error_code} + + return {'success': False, 'error': 'UNKNOWN_ERROR'} + + def _build_topic_subscription_result_from_httpx_error(self, error, is_subscribe): + """Constructs a result dict from an httpx error.""" + if isinstance(error, httpx.HTTPStatusError): + if is_subscribe and error.response.status_code == 409: + return {'success': True} + error_dict = {} + try: + parsed = error.response.json() + if isinstance(parsed, dict): + error_dict = parsed + except ValueError: + pass + + error_data = error_dict.get('error') + if is_subscribe and isinstance(error_data, dict) and ( + error_data.get('status') == 'ALREADY_EXISTS' + ): + return {'success': True} + + error_code = self._get_topic_error_code(error_dict, error.response.status_code) + return {'success': False, 'error': error_code} + + return {'success': False, 'error': 'UNKNOWN_ERROR'} + + def _parse_topic_management_results(self, results) -> TopicManagementResponse: + """Parses individual request results into a TopicManagementResponse.""" + formatted_results = [] + for result in results: + if result.get('success'): + formatted_results.append({}) + else: + formatted_results.append({'error': result.get('error', 'UNKNOWN_ERROR')}) + return TopicManagementResponse({'results': formatted_results}) + def make_topic_management_request(self, tokens, topic, operation): """Invokes the IID service for topic management functionality.""" if isinstance(tokens, str): diff --git a/tests/test_messaging.py b/tests/test_messaging.py index 749e5311..b0cf7462 100644 --- a/tests/test_messaging.py +++ b/tests/test_messaging.py @@ -14,10 +14,12 @@ """Test cases for the firebase_admin.messaging module.""" import datetime +import io from itertools import chain, repeat import json import numbers import httpx +import requests import respx import pytest @@ -1742,10 +1744,25 @@ def test_topic_management_custom_timeout(self, options, timeout): all_options.update(options) firebase_admin.initialize_app(cred, all_options) recorder = self._instrument_service( - 'https://iid.googleapis.com', {'results': [{}, {'error': 'error_reason'}]}) + 'https://fcm.googleapis.com', {}) messaging.subscribe_to_topic(['1'], 'a') self._check_timeout(recorder, timeout) + @pytest.mark.parametrize('options, timeout', [ + ({'httpTimeout': 4}, 4), + ({'httpTimeout': None}, None), + ({}, _http_client.DEFAULT_TIMEOUT_SECONDS), + ]) + def test_topic_management_legacy_custom_timeout(self, options, timeout): + cred = testutils.MockCredential() + all_options = {'projectId': 'explicit-project-id'} + all_options.update(options) + firebase_admin.initialize_app(cred, all_options) + recorder = self._instrument_service( + 'https://iid.googleapis.com', {'results': [{}, {'error': 'error_reason'}]}) + messaging.subscribe_to_topic_legacy(['1'], 'a') + self._check_timeout(recorder, timeout) + class TestSend: @@ -2448,7 +2465,7 @@ def test_send_each_for_multicast_fcm_error_code(self, status): check_exception(exception, 'test error', status) -class TestTopicManagement: +class TestTopicManagementLegacy: _DEFAULT_RESPONSE = json.dumps({'results': [{}, {'error': 'error_reason'}]}) _DEFAULT_ERROR_RESPONSE = json.dumps({'error': 'error_reason'}) @@ -2502,77 +2519,79 @@ def test_invalid_tokens(self, tokens): expected = 'Tokens must be non-empty strings.' with pytest.raises(ValueError) as excinfo: - messaging.subscribe_to_topic(tokens, 'test-topic') + messaging.subscribe_to_topic_legacy(tokens, 'test-topic') assert str(excinfo.value) == expected with pytest.raises(ValueError) as excinfo: - messaging.unsubscribe_from_topic(tokens, 'test-topic') + messaging.unsubscribe_from_topic_legacy(tokens, 'test-topic') assert str(excinfo.value) == expected @pytest.mark.parametrize('topic', NON_STRING_ARGS + [None, '']) def test_invalid_topic(self, topic): expected = 'Topic must be a non-empty string.' with pytest.raises(ValueError) as excinfo: - messaging.subscribe_to_topic('test-token', topic) + messaging.subscribe_to_topic_legacy('test-token', topic) assert str(excinfo.value) == expected with pytest.raises(ValueError) as excinfo: - messaging.unsubscribe_from_topic('test-tokens', topic) + messaging.unsubscribe_from_topic_legacy('test-tokens', topic) assert str(excinfo.value) == expected @pytest.mark.parametrize('args', _VALID_ARGS) - def test_subscribe_to_topic(self, args): + def test_subscribe_to_topic_legacy(self, args): _, recorder = self._instrument_iid_service() - resp = messaging.subscribe_to_topic(args[0], args[1]) + with pytest.deprecated_call(): + resp = messaging.subscribe_to_topic_legacy(args[0], args[1]) self._check_response(resp) assert len(recorder) == 1 self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchAdd')) assert json.loads(recorder[0].body.decode()) == args[2] @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items()) - def test_subscribe_to_topic_error(self, status, exc_type): + def test_subscribe_to_topic_legacy_error(self, status, exc_type): _, recorder = self._instrument_iid_service( status=status, payload=self._DEFAULT_ERROR_RESPONSE) with pytest.raises(exc_type) as excinfo: - messaging.subscribe_to_topic('foo', 'test-topic') + messaging.subscribe_to_topic_legacy('foo', 'test-topic') assert str(excinfo.value) == 'Error while calling the IID service: error_reason' assert len(recorder) == 1 self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchAdd')) @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items()) - def test_subscribe_to_topic_non_json_error(self, status, exc_type): + def test_subscribe_to_topic_legacy_non_json_error(self, status, exc_type): _, recorder = self._instrument_iid_service(status=status, payload='not json') with pytest.raises(exc_type) as excinfo: - messaging.subscribe_to_topic('foo', 'test-topic') + messaging.subscribe_to_topic_legacy('foo', 'test-topic') reason = f'Unexpected HTTP response with status: {status}; body: not json' assert str(excinfo.value) == reason assert len(recorder) == 1 self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchAdd')) @pytest.mark.parametrize('args', _VALID_ARGS) - def test_unsubscribe_from_topic(self, args): + def test_unsubscribe_from_topic_legacy(self, args): _, recorder = self._instrument_iid_service() - resp = messaging.unsubscribe_from_topic(args[0], args[1]) + with pytest.deprecated_call(): + resp = messaging.unsubscribe_from_topic_legacy(args[0], args[1]) self._check_response(resp) assert len(recorder) == 1 self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchRemove')) assert json.loads(recorder[0].body.decode()) == args[2] @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items()) - def test_unsubscribe_from_topic_error(self, status, exc_type): + def test_unsubscribe_from_topic_legacy_error(self, status, exc_type): _, recorder = self._instrument_iid_service( status=status, payload=self._DEFAULT_ERROR_RESPONSE) with pytest.raises(exc_type) as excinfo: - messaging.unsubscribe_from_topic('foo', 'test-topic') + messaging.unsubscribe_from_topic_legacy('foo', 'test-topic') assert str(excinfo.value) == 'Error while calling the IID service: error_reason' assert len(recorder) == 1 self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchRemove')) @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items()) - def test_unsubscribe_from_topic_non_json_error(self, status, exc_type): + def test_unsubscribe_from_topic_legacy_non_json_error(self, status, exc_type): _, recorder = self._instrument_iid_service(status=status, payload='not json') with pytest.raises(exc_type) as excinfo: - messaging.unsubscribe_from_topic('foo', 'test-topic') + messaging.unsubscribe_from_topic_legacy('foo', 'test-topic') reason = f'Unexpected HTTP response with status: {status}; body: not json' assert str(excinfo.value) == reason assert len(recorder) == 1 @@ -2584,3 +2603,298 @@ def _check_response(self, resp): assert len(resp.errors) == 1 assert resp.errors[0].index == 1 assert resp.errors[0].reason == 'error_reason' + + +class TestTopicManagement: + + _CLIENT_VERSION = f'fire-admin-python/{firebase_admin.__version__}' + + @classmethod + def setup_class(cls): + cred = testutils.MockCredential() + firebase_admin.initialize_app(cred, {'projectId': 'explicit-project-id'}) + + @classmethod + def teardown_class(cls): + testutils.cleanup_apps() + + def _instrument_messaging_service(self, app=None, status=200, payload='{}'): + if not app: + app = firebase_admin.get_app() + fcm_service = messaging._get_messaging_service(app) + recorder = [] + fcm_service._client.session.mount( + 'https://fcm.googleapis.com', + testutils.MockAdapter(payload, status, recorder)) + return fcm_service, recorder + + def _assert_request(self, request, expected_method, expected_url, expected_body=None): + assert request.method == expected_method + assert request.url == expected_url + assert request.headers['X-GOOG-API-FORMAT-VERSION'] == '2' + assert request.headers['X-FIREBASE-CLIENT'] == self._CLIENT_VERSION + expected_metrics_header = _utils.get_metrics_header() + ' mock-cred-metric-tag' + assert request.headers['x-goog-api-client'] == expected_metrics_header + if expected_body is None: + assert request.body is None + else: + assert json.loads(request.body.decode()) == expected_body + + @pytest.mark.parametrize('tokens', [None, '', [], {}, tuple()]) + def test_invalid_tokens(self, tokens): + expected = 'Tokens must be a string or a non-empty list of strings.' + if isinstance(tokens, str): + expected = 'Tokens must be non-empty strings.' + + with pytest.raises(ValueError) as excinfo: + messaging.subscribe_to_topic(tokens, 'test-topic') + assert str(excinfo.value) == expected + + with pytest.raises(ValueError) as excinfo: + messaging.unsubscribe_from_topic(tokens, 'test-topic') + assert str(excinfo.value) == expected + + @pytest.mark.parametrize('tokens', [ + ['foo', 'bar', ''], + ['foo', 123, 'bar'], + ]) + def test_invalid_tokens_in_list(self, tokens): + with pytest.raises(ValueError) as excinfo: + messaging.subscribe_to_topic(tokens, 'test-topic') + assert str(excinfo.value) == 'Tokens must be non-empty strings.' + + with pytest.raises(ValueError) as excinfo: + messaging.unsubscribe_from_topic(tokens, 'test-topic') + assert str(excinfo.value) == 'Tokens must be non-empty strings.' + + def test_tokens_over_1000(self): + tokens = [f'token{i}' for i in range(1001)] + with pytest.raises(ValueError) as excinfo: + messaging.subscribe_to_topic(tokens, 'test-topic') + assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.' + + with pytest.raises(ValueError) as excinfo: + messaging.unsubscribe_from_topic(tokens, 'test-topic') + assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.' + + @pytest.mark.parametrize('topic', NON_STRING_ARGS + [None, '']) + def test_invalid_topic(self, topic): + expected = 'Topic must be a non-empty string.' + with pytest.raises(ValueError) as excinfo: + messaging.subscribe_to_topic('test-token', topic) + assert str(excinfo.value) == expected + + with pytest.raises(ValueError) as excinfo: + messaging.unsubscribe_from_topic('test-token', topic) + assert str(excinfo.value) == expected + + @pytest.mark.parametrize('topic', ['/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&']) + def test_malformed_topic(self, topic): + with pytest.raises(ValueError) as excinfo: + messaging.subscribe_to_topic('test-token', topic) + assert str(excinfo.value) == 'Malformed topic name.' + + with pytest.raises(ValueError) as excinfo: + messaging.unsubscribe_from_topic('test-token', topic) + assert str(excinfo.value) == 'Malformed topic name.' + + def test_subscribe_to_topic_single(self): + _, recorder = self._instrument_messaging_service() + resp = messaging.subscribe_to_topic('token1', 'test-topic') + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + assert len(recorder) == 1 + expected_url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + self._assert_request(recorder[0], 'POST', expected_url, {}) + + def test_subscribe_to_topic_prefixed(self): + _, recorder = self._instrument_messaging_service() + resp = messaging.subscribe_to_topic('token1', '/topics/test-topic') + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + assert len(recorder) == 1 + expected_url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + self._assert_request(recorder[0], 'POST', expected_url, {}) + + def test_unsubscribe_from_topic_single(self): + _, recorder = self._instrument_messaging_service() + resp = messaging.unsubscribe_from_topic('token1', 'test-topic') + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + assert len(recorder) == 1 + expected_url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions/test-topic?allow_missing=true' + self._assert_request(recorder[0], 'DELETE', expected_url, None) + + def test_subscribe_to_topic_already_exists_409(self): + payload = json.dumps({'error': {'status': 'ALREADY_EXISTS', 'message': 'Already exists'}}) + _, recorder = self._instrument_messaging_service(status=409, payload=payload) + resp = messaging.subscribe_to_topic('token1', 'test-topic') + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + assert len(recorder) == 1 + + def test_unsubscribe_from_topic_not_found_404(self): + payload = json.dumps({'error': {'status': 'NOT_FOUND', 'message': 'Not found'}}) + _, recorder = self._instrument_messaging_service(status=404, payload=payload) + resp = messaging.unsubscribe_from_topic('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'NOT_FOUND' + assert len(recorder) == 1 + + def test_topic_management_multiple_tokens(self): + fcm_service = messaging._get_messaging_service(firebase_admin.get_app()) + recorder = [] + + class MultiTokenMockAdapter(requests.adapters.HTTPAdapter): + def send(self, request, **kwargs): + recorder.append(request) + resp = requests.models.Response() + resp.url = request.url + if 'token2' in request.url: + resp.status_code = 404 + content = json.dumps({'error': {'status': 'NOT_FOUND'}}).encode() + else: + resp.status_code = 200 + content = b'{}' + resp.raw = io.BytesIO(content) + return resp + + fcm_service._client.session.mount( + 'https://fcm.googleapis.com', + MultiTokenMockAdapter()) + + resp = messaging.subscribe_to_topic(['token1', 'token2', 'token3'], 'test-topic') + assert resp.success_count == 2 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 1 + assert resp.errors[0].reason == 'NOT_FOUND' + assert len(recorder) == 3 + + def test_topic_management_fcm_error_details(self): + payload = json.dumps({ + 'error': { + 'status': 'NOT_FOUND', + 'details': [ + { + '@type': 'type.googleapis.com/google.firebase.fcm.v1.FcmError', + 'errorCode': 'UNREGISTERED', + }, + ], + } + }) + _, recorder = self._instrument_messaging_service(status=404, payload=payload) + resp = messaging.subscribe_to_topic('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'UNREGISTERED' + assert len(recorder) == 1 + + def test_topic_management_500_error(self): + _, recorder = self._instrument_messaging_service(status=500, payload='{"error": null}') + resp = messaging.subscribe_to_topic('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'INTERNAL' + assert len(recorder) == 1 + + def test_topic_management_non_json_error(self): + _, recorder = self._instrument_messaging_service(status=400, payload='not json') + resp = messaging.subscribe_to_topic('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'INVALID_ARGUMENT' + assert len(recorder) == 1 + + +class TestTopicManagementAsync: + + @classmethod + def setup_class(cls): + cred = testutils.MockCredential() + firebase_admin.initialize_app(cred, {'projectId': 'explicit-project-id'}) + + @classmethod + def teardown_class(cls): + testutils.cleanup_apps() + + @pytest.mark.asyncio + @respx.mock + async def test_subscribe_to_topic_async_single(self): + url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + route = respx.post(url).mock(return_value=respx.MockResponse(200, json={})) + resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') + assert route.call_count == 1 + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + + @pytest.mark.asyncio + @respx.mock + async def test_subscribe_to_topic_async_already_exists(self): + url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + payload = {'error': {'status': 'ALREADY_EXISTS', 'message': 'Already exists'}} + route = respx.post(url).mock(return_value=respx.MockResponse(409, json=payload)) + resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') + assert route.call_count == 1 + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + + @pytest.mark.asyncio + @respx.mock + async def test_unsubscribe_from_topic_async_single(self): + url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions/test-topic?allow_missing=true' + route = respx.delete(url).mock(return_value=respx.MockResponse(200, json={})) + resp = await messaging.unsubscribe_from_topic_async('token1', 'test-topic') + assert route.call_count == 1 + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + + @pytest.mark.asyncio + @respx.mock + async def test_unsubscribe_from_topic_async_not_found(self): + url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions/test-topic?allow_missing=true' + payload = {'error': {'status': 'NOT_FOUND', 'message': 'Not found'}} + route = respx.delete(url).mock(return_value=respx.MockResponse(404, json=payload)) + resp = await messaging.unsubscribe_from_topic_async('token1', 'test-topic') + assert route.call_count == 1 + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'NOT_FOUND' + + @pytest.mark.asyncio + @respx.mock + async def test_subscribe_to_topic_async_multiple(self): + url1 = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + url2 = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token2/topicSubscriptions?topic_name=test-topic' + url3 = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token3/topicSubscriptions?topic_name=test-topic' + route1 = respx.post(url1).mock(return_value=respx.MockResponse(200, json={})) + route2 = respx.post(url2).mock(return_value=respx.MockResponse(404, json={'error': {'status': 'NOT_FOUND'}})) + route3 = respx.post(url3).mock(return_value=respx.MockResponse(200, json={})) + + resp = await messaging.subscribe_to_topic_async(['token1', 'token2', 'token3'], 'test-topic') + assert route1.call_count == 1 + assert route2.call_count == 1 + assert route3.call_count == 1 + assert resp.success_count == 2 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 1 + assert resp.errors[0].reason == 'NOT_FOUND' From 9c9b63736c18d97641936b02dbc0201c6878d713 Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Tue, 15 Sep 2026 10:18:57 -0400 Subject: [PATCH 2/4] fix(fcm): Prevent headers mutation race condition and deduplicate topic error parsing - Pass a copy of self._fcm_headers in topic management requests to prevent concurrent mutation race conditions in ThreadPoolExecutor. - Extract common error parsing logic from _build_topic_subscription_result_from_requests_error and _build_topic_subscription_result_from_httpx_error into _build_topic_subscription_result. --- firebase_admin/messaging.py | 43 ++++++++++++------------------------- 1 file changed, 14 insertions(+), 29 deletions(-) diff --git a/firebase_admin/messaging.py b/firebase_admin/messaging.py index d266f6f9..ac2fe89a 100644 --- a/firebase_admin/messaging.py +++ b/firebase_admin/messaging.py @@ -658,7 +658,7 @@ def send_request(token: str): self._client.request( method, url=url, - headers=self._fcm_headers, + headers=dict(self._fcm_headers), json=json_data, ) return {'success': True} @@ -702,7 +702,7 @@ async def send_request_async(token: str): await self._async_client.request( method, url=url, - headers=self._fcm_headers, + headers=dict(self._fcm_headers), json=json_data, ) return {'success': True} @@ -754,14 +754,14 @@ def _get_topic_error_code(cls, error_dict: dict, status_code: int) -> str: } return status_map.get(status_code, 'UNKNOWN_ERROR') - def _build_topic_subscription_result_from_requests_error(self, error, is_subscribe): - """Constructs a result dict from a requests error.""" - if error.response is not None: - if is_subscribe and error.response.status_code == 409: + def _build_topic_subscription_result(self, response, is_subscribe): + """Constructs a result dict from a response object.""" + if response is not None: + if is_subscribe and response.status_code == 409: return {'success': True} error_dict = {} try: - parsed = error.response.json() + parsed = response.json() if isinstance(parsed, dict): error_dict = parsed except ValueError: @@ -773,34 +773,19 @@ def _build_topic_subscription_result_from_requests_error(self, error, is_subscri ): return {'success': True} - error_code = self._get_topic_error_code(error_dict, error.response.status_code) + error_code = self._get_topic_error_code(error_dict, response.status_code) return {'success': False, 'error': error_code} return {'success': False, 'error': 'UNKNOWN_ERROR'} + def _build_topic_subscription_result_from_requests_error(self, error, is_subscribe): + """Constructs a result dict from a requests error.""" + return self._build_topic_subscription_result(error.response, is_subscribe) + def _build_topic_subscription_result_from_httpx_error(self, error, is_subscribe): """Constructs a result dict from an httpx error.""" - if isinstance(error, httpx.HTTPStatusError): - if is_subscribe and error.response.status_code == 409: - return {'success': True} - error_dict = {} - try: - parsed = error.response.json() - if isinstance(parsed, dict): - error_dict = parsed - except ValueError: - pass - - error_data = error_dict.get('error') - if is_subscribe and isinstance(error_data, dict) and ( - error_data.get('status') == 'ALREADY_EXISTS' - ): - return {'success': True} - - error_code = self._get_topic_error_code(error_dict, error.response.status_code) - return {'success': False, 'error': error_code} - - return {'success': False, 'error': 'UNKNOWN_ERROR'} + response = error.response if isinstance(error, httpx.HTTPStatusError) else None + return self._build_topic_subscription_result(response, is_subscribe) def _parse_topic_management_results(self, results) -> TopicManagementResponse: """Parses individual request results into a TopicManagementResponse.""" From 02948a248af335b87221ec32d3a5a8f47616322b Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 12:51:29 -0400 Subject: [PATCH 3/4] fix(fcm): Improve topic management performance and fix test lint issues - Hoist URL-encoded topic computation out of the per-token request loops. - Mount an HTTPAdapter with a connection pool size of 100 on the FCM client session. - Fix line length and method signature override lint warnings in test_messaging.py. --- firebase_admin/messaging.py | 10 +++++-- tests/test_messaging.py | 54 +++++++++++++++++++++++++++---------- 2 files changed, 48 insertions(+), 16 deletions(-) diff --git a/firebase_admin/messaging.py b/firebase_admin/messaging.py index ac2fe89a..372124f6 100644 --- a/firebase_admin/messaging.py +++ b/firebase_admin/messaging.py @@ -512,6 +512,12 @@ def __init__(self, app: App) -> None: timeout = app.options.get('httpTimeout', _http_client.DEFAULT_TIMEOUT_SECONDS) self._credential = app.credential.get_credential() self._client = _http_client.JsonHttpClient(credential=self._credential, timeout=timeout) + fcm_adapter = requests.adapters.HTTPAdapter( + pool_connections=100, + pool_maxsize=100, + max_retries=_http_client.DEFAULT_RETRY_CONFIG + ) + self._client.session.mount('https://fcm.googleapis.com', fcm_adapter) self._async_client = _http_client.HttpxAsyncClient( credential=self._credential, timeout=timeout) @@ -640,10 +646,10 @@ def _make_topic_management_request_v1( ) -> TopicManagementResponse: """Helper method that sends topic subscription requests via FCM v1 API.""" tokens_list, topic_name = self._validate_topic_management_args(tokens, topic) + encoded_topic = urllib.parse.quote(topic_name, safe='') def send_request(token: str): encoded_token = urllib.parse.quote(token, safe='') - encoded_topic = urllib.parse.quote(topic_name, safe='') base_url = f'{self._fcm_topic_url}/{encoded_token}/topicSubscriptions' if is_subscribe: url = f'{base_url}?topic_name={encoded_topic}' @@ -682,11 +688,11 @@ async def _make_topic_management_request_v1_async( ) -> TopicManagementResponse: """Helper method that sends topic subscription requests asynchronously via FCM v1 API.""" tokens_list, topic_name = self._validate_topic_management_args(tokens, topic) + encoded_topic = urllib.parse.quote(topic_name, safe='') semaphore = asyncio.Semaphore(100) async def send_request_async(token: str): encoded_token = urllib.parse.quote(token, safe='') - encoded_topic = urllib.parse.quote(topic_name, safe='') base_url = f'{self._fcm_topic_url}/{encoded_token}/topicSubscriptions' if is_subscribe: url = f'{base_url}?topic_name={encoded_topic}' diff --git a/tests/test_messaging.py b/tests/test_messaging.py index b0cf7462..dfaeaf66 100644 --- a/tests/test_messaging.py +++ b/tests/test_messaging.py @@ -2688,7 +2688,9 @@ def test_invalid_topic(self, topic): messaging.unsubscribe_from_topic('test-token', topic) assert str(excinfo.value) == expected - @pytest.mark.parametrize('topic', ['/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&']) + @pytest.mark.parametrize('topic', [ + '/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&' + ]) def test_malformed_topic(self, topic): with pytest.raises(ValueError) as excinfo: messaging.subscribe_to_topic('test-token', topic) @@ -2705,7 +2707,10 @@ def test_subscribe_to_topic_single(self): assert resp.failure_count == 0 assert resp.errors == [] assert len(recorder) == 1 - expected_url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + expected_url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) self._assert_request(recorder[0], 'POST', expected_url, {}) def test_subscribe_to_topic_prefixed(self): @@ -2715,7 +2720,10 @@ def test_subscribe_to_topic_prefixed(self): assert resp.failure_count == 0 assert resp.errors == [] assert len(recorder) == 1 - expected_url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + expected_url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) self._assert_request(recorder[0], 'POST', expected_url, {}) def test_unsubscribe_from_topic_single(self): @@ -2725,7 +2733,10 @@ def test_unsubscribe_from_topic_single(self): assert resp.failure_count == 0 assert resp.errors == [] assert len(recorder) == 1 - expected_url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions/test-topic?allow_missing=true' + expected_url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions/test-topic?allow_missing=true' + ) self._assert_request(recorder[0], 'DELETE', expected_url, None) def test_subscribe_to_topic_already_exists_409(self): @@ -2753,7 +2764,7 @@ def test_topic_management_multiple_tokens(self): recorder = [] class MultiTokenMockAdapter(requests.adapters.HTTPAdapter): - def send(self, request, **kwargs): + def send(self, request, **kwargs): # pylint: disable=arguments-differ,unused-argument recorder.append(request) resp = requests.models.Response() resp.url = request.url @@ -2834,7 +2845,10 @@ def teardown_class(cls): @pytest.mark.asyncio @respx.mock async def test_subscribe_to_topic_async_single(self): - url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) route = respx.post(url).mock(return_value=respx.MockResponse(200, json={})) resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') assert route.call_count == 1 @@ -2845,7 +2859,10 @@ async def test_subscribe_to_topic_async_single(self): @pytest.mark.asyncio @respx.mock async def test_subscribe_to_topic_async_already_exists(self): - url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) payload = {'error': {'status': 'ALREADY_EXISTS', 'message': 'Already exists'}} route = respx.post(url).mock(return_value=respx.MockResponse(409, json=payload)) resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') @@ -2857,7 +2874,10 @@ async def test_subscribe_to_topic_async_already_exists(self): @pytest.mark.asyncio @respx.mock async def test_unsubscribe_from_topic_async_single(self): - url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions/test-topic?allow_missing=true' + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions/test-topic?allow_missing=true' + ) route = respx.delete(url).mock(return_value=respx.MockResponse(200, json={})) resp = await messaging.unsubscribe_from_topic_async('token1', 'test-topic') assert route.call_count == 1 @@ -2868,7 +2888,10 @@ async def test_unsubscribe_from_topic_async_single(self): @pytest.mark.asyncio @respx.mock async def test_unsubscribe_from_topic_async_not_found(self): - url = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions/test-topic?allow_missing=true' + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions/test-topic?allow_missing=true' + ) payload = {'error': {'status': 'NOT_FOUND', 'message': 'Not found'}} route = respx.delete(url).mock(return_value=respx.MockResponse(404, json=payload)) resp = await messaging.unsubscribe_from_topic_async('token1', 'test-topic') @@ -2882,14 +2905,17 @@ async def test_unsubscribe_from_topic_async_not_found(self): @pytest.mark.asyncio @respx.mock async def test_subscribe_to_topic_async_multiple(self): - url1 = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token1/topicSubscriptions?topic_name=test-topic' - url2 = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token2/topicSubscriptions?topic_name=test-topic' - url3 = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/token3/topicSubscriptions?topic_name=test-topic' + base = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations' + url1 = f'{base}/token1/topicSubscriptions?topic_name=test-topic' + url2 = f'{base}/token2/topicSubscriptions?topic_name=test-topic' + url3 = f'{base}/token3/topicSubscriptions?topic_name=test-topic' route1 = respx.post(url1).mock(return_value=respx.MockResponse(200, json={})) - route2 = respx.post(url2).mock(return_value=respx.MockResponse(404, json={'error': {'status': 'NOT_FOUND'}})) + route2 = respx.post(url2).mock( + return_value=respx.MockResponse(404, json={'error': {'status': 'NOT_FOUND'}})) route3 = respx.post(url3).mock(return_value=respx.MockResponse(200, json={})) - resp = await messaging.subscribe_to_topic_async(['token1', 'token2', 'token3'], 'test-topic') + resp = await messaging.subscribe_to_topic_async( + ['token1', 'token2', 'token3'], 'test-topic') assert route1.call_count == 1 assert route2.call_count == 1 assert route3.call_count == 1 From 3c654a0b213d739a4e225c2d5c5d1bfbbd82412c Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 13:11:22 -0400 Subject: [PATCH 4/4] fix(fcm): Addressed review improvements for FCM v1 topic management - Added type annotations to public topic management functions and internal methods - Updated HTTP status code mapping (401->UNAUTHENTICATED, 503->UNAVAILABLE, 408/504->DEADLINE_EXCEEDED) and prioritized status codes over free-form message strings - Updated topic regex with \Z to reject trailing newlines - Passed header copies in send, send_each, and send_each_async to prevent concurrency mutations - Added test coverage for status code mapping, async argument validation, prefixed topics, and async batch unsubscribe --- firebase_admin/messaging.py | 69 ++++++++---- tests/test_messaging.py | 208 +++++++++++++++++++++++++++++++++++- 2 files changed, 255 insertions(+), 22 deletions(-) diff --git a/firebase_admin/messaging.py b/firebase_admin/messaging.py index 372124f6..a108444a 100644 --- a/firebase_admin/messaging.py +++ b/firebase_admin/messaging.py @@ -20,7 +20,7 @@ import json import logging import re -from typing import Any, Callable, Dict, List, Optional, cast +from typing import Any, Callable, Dict, List, Optional, Union, cast import urllib.parse import warnings @@ -259,7 +259,9 @@ def send_each_for_multicast(multicast_message, dry_run=False, app=None): messages = _get_messages_from_multicast(multicast_message) return _get_messaging_service(app).send_each(messages, dry_run) -def subscribe_to_topic(tokens, topic, app=None): +def subscribe_to_topic( + tokens: Union[str, List[str]], topic: str, app: Optional[App] = None +) -> TopicManagementResponse: """Subscribes a list of registration tokens to an FCM topic. Args: @@ -277,7 +279,9 @@ def subscribe_to_topic(tokens, topic, app=None): """ return _get_messaging_service(app).subscribe_to_topic(tokens, topic) -async def subscribe_to_topic_async(tokens, topic, app=None): +async def subscribe_to_topic_async( + tokens: Union[str, List[str]], topic: str, app: Optional[App] = None +) -> TopicManagementResponse: """Subscribes a list of registration tokens to an FCM topic asynchronously. Args: @@ -295,7 +299,9 @@ async def subscribe_to_topic_async(tokens, topic, app=None): """ return await _get_messaging_service(app).subscribe_to_topic_async(tokens, topic) -def subscribe_to_topic_legacy(tokens, topic, app=None): +def subscribe_to_topic_legacy( + tokens: Union[str, List[str]], topic: str, app: Optional[App] = None +) -> TopicManagementResponse: """Subscribes a list of registration tokens to an FCM topic using the legacy Instance ID API. subscribe_to_topic_legacy is deprecated. Use subscribe_to_topic instead. @@ -320,7 +326,9 @@ def subscribe_to_topic_legacy(tokens, topic, app=None): return _get_messaging_service(app).make_topic_management_request( tokens, topic, 'iid/v1:batchAdd') -def unsubscribe_from_topic(tokens, topic, app=None): +def unsubscribe_from_topic( + tokens: Union[str, List[str]], topic: str, app: Optional[App] = None +) -> TopicManagementResponse: """Unsubscribes a list of registration tokens from an FCM topic. Args: @@ -338,7 +346,9 @@ def unsubscribe_from_topic(tokens, topic, app=None): """ return _get_messaging_service(app).unsubscribe_from_topic(tokens, topic) -async def unsubscribe_from_topic_async(tokens, topic, app=None): +async def unsubscribe_from_topic_async( + tokens: Union[str, List[str]], topic: str, app: Optional[App] = None +) -> TopicManagementResponse: """Unsubscribes a list of registration tokens from an FCM topic asynchronously. Args: @@ -356,7 +366,9 @@ async def unsubscribe_from_topic_async(tokens, topic, app=None): """ return await _get_messaging_service(app).unsubscribe_from_topic_async(tokens, topic) -def unsubscribe_from_topic_legacy(tokens, topic, app=None): +def unsubscribe_from_topic_legacy( + tokens: Union[str, List[str]], topic: str, app: Optional[App] = None +) -> TopicManagementResponse: """Unsubscribes a list of registration tokens from an FCM topic using the legacy Instance ID API. @@ -534,7 +546,7 @@ def send(self, message: Message, dry_run: bool = False) -> str: resp = self._client.body( 'post', url=self._fcm_url, - headers=self._fcm_headers, + headers=dict(self._fcm_headers), json=data ) except requests.exceptions.RequestException as error: @@ -553,7 +565,7 @@ def send_data(data): resp = self._client.body( 'post', url=self._fcm_url, - headers=self._fcm_headers, + headers=dict(self._fcm_headers), json=data) except requests.exceptions.RequestException as exception: return SendResponse(resp=None, exception=self._handle_fcm_error(exception)) @@ -581,7 +593,7 @@ async def send_data(data): resp = await self._async_client.request( 'post', url=self._fcm_url, - headers=self._fcm_headers, + headers=dict(self._fcm_headers), json=data) except httpx.HTTPError as exception: return SendResponse(resp=None, exception=self._handle_fcm_httpx_error(exception)) @@ -616,26 +628,34 @@ def _validate_topic_management_args(self, tokens, topic): topic_name = topic if topic_name.startswith('/topics/'): topic_name = topic_name[len('/topics/'):] - if not topic_name or not re.match(r'^[a-zA-Z0-9-_\.~%]+$', topic_name): + if not topic_name or not re.match(r'^[a-zA-Z0-9-_\.~%]+\Z', topic_name): raise ValueError('Malformed topic name.') return tokens, topic_name - def subscribe_to_topic(self, tokens, topic) -> TopicManagementResponse: + def subscribe_to_topic( + self, tokens: Union[str, List[str]], topic: str + ) -> TopicManagementResponse: """Subscribes a list of registration tokens to an FCM topic via the FCM v1 API.""" return self._make_topic_management_request_v1(tokens, topic, is_subscribe=True) - def unsubscribe_from_topic(self, tokens, topic) -> TopicManagementResponse: + def unsubscribe_from_topic( + self, tokens: Union[str, List[str]], topic: str + ) -> TopicManagementResponse: """Unsubscribes a list of registration tokens from an FCM topic via the FCM v1 API.""" return self._make_topic_management_request_v1(tokens, topic, is_subscribe=False) - async def subscribe_to_topic_async(self, tokens, topic) -> TopicManagementResponse: + async def subscribe_to_topic_async( + self, tokens: Union[str, List[str]], topic: str + ) -> TopicManagementResponse: """Subscribes a list of registration tokens to an FCM topic asynchronously via the FCM v1 API.""" return await self._make_topic_management_request_v1_async( tokens, topic, is_subscribe=True) - async def unsubscribe_from_topic_async(self, tokens, topic) -> TopicManagementResponse: + async def unsubscribe_from_topic_async( + self, tokens: Union[str, List[str]], topic: str + ) -> TopicManagementResponse: """Unsubscribes a list of registration tokens from an FCM topic asynchronously via the FCM v1 API.""" return await self._make_topic_management_request_v1_async( @@ -745,20 +765,27 @@ def _get_topic_error_code(cls, error_dict: dict, status_code: int) -> str: status = error_data.get('status') if status: return status - message = error_data.get('message') - if message: - return message status_map = { 400: 'INVALID_ARGUMENT', - 401: 'PERMISSION_DENIED', + 401: 'UNAUTHENTICATED', 403: 'PERMISSION_DENIED', 404: 'NOT_FOUND', + 408: 'DEADLINE_EXCEEDED', 429: 'RESOURCE_EXHAUSTED', 500: 'INTERNAL', - 503: 'DEADLINE_EXCEEDED', + 503: 'UNAVAILABLE', + 504: 'DEADLINE_EXCEEDED', } - return status_map.get(status_code, 'UNKNOWN_ERROR') + if status_code in status_map: + return status_map[status_code] + + if isinstance(error_data, dict): + message = error_data.get('message') + if message: + return message + + return 'UNKNOWN_ERROR' def _build_topic_subscription_result(self, response, is_subscribe): """Constructs a result dict from a response object.""" diff --git a/tests/test_messaging.py b/tests/test_messaging.py index dfaeaf66..ae2e4a4d 100644 --- a/tests/test_messaging.py +++ b/tests/test_messaging.py @@ -2689,7 +2689,8 @@ def test_invalid_topic(self, topic): assert str(excinfo.value) == expected @pytest.mark.parametrize('topic', [ - '/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&' + '/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&', + '/topics/foo\n', 'foo\n', ]) def test_malformed_topic(self, topic): with pytest.raises(ValueError) as excinfo: @@ -2739,6 +2740,19 @@ def test_unsubscribe_from_topic_single(self): ) self._assert_request(recorder[0], 'DELETE', expected_url, None) + def test_unsubscribe_from_topic_prefixed(self): + _, recorder = self._instrument_messaging_service() + resp = messaging.unsubscribe_from_topic('token1', '/topics/test-topic') + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + assert len(recorder) == 1 + expected_url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions/test-topic?allow_missing=true' + ) + self._assert_request(recorder[0], 'DELETE', expected_url, None) + def test_subscribe_to_topic_already_exists_409(self): payload = json.dumps({'error': {'status': 'ALREADY_EXISTS', 'message': 'Already exists'}}) _, recorder = self._instrument_messaging_service(status=409, payload=payload) @@ -2830,6 +2844,25 @@ def test_topic_management_non_json_error(self): assert resp.errors[0].reason == 'INVALID_ARGUMENT' assert len(recorder) == 1 + @pytest.mark.parametrize('status_code, expected_reason', [ + (401, 'UNAUTHENTICATED'), + (403, 'PERMISSION_DENIED'), + (408, 'DEADLINE_EXCEEDED'), + (429, 'RESOURCE_EXHAUSTED'), + (503, 'UNAVAILABLE'), + (504, 'DEADLINE_EXCEEDED'), + ]) + def test_topic_management_status_code_mapping(self, status_code, expected_reason): + _, recorder = self._instrument_messaging_service( + status=status_code, payload=json.dumps({'error': {'message': 'Some message'}})) + resp = messaging.subscribe_to_topic('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == expected_reason + assert len(recorder) == (3 if status_code == 401 else 1) + class TestTopicManagementAsync: @@ -2924,3 +2957,176 @@ async def test_subscribe_to_topic_async_multiple(self): assert len(resp.errors) == 1 assert resp.errors[0].index == 1 assert resp.errors[0].reason == 'NOT_FOUND' + + @pytest.mark.asyncio + @respx.mock + async def test_subscribe_to_topic_async_prefixed(self): + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) + route = respx.post(url).mock(return_value=respx.MockResponse(200, json={})) + resp = await messaging.subscribe_to_topic_async('token1', '/topics/test-topic') + assert route.call_count == 1 + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + + @pytest.mark.asyncio + @respx.mock + async def test_unsubscribe_from_topic_async_prefixed(self): + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions/test-topic?allow_missing=true' + ) + route = respx.delete(url).mock(return_value=respx.MockResponse(200, json={})) + resp = await messaging.unsubscribe_from_topic_async('token1', '/topics/test-topic') + assert route.call_count == 1 + assert resp.success_count == 1 + assert resp.failure_count == 0 + assert resp.errors == [] + + @pytest.mark.asyncio + @respx.mock + async def test_unsubscribe_from_topic_async_multiple(self): + base = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations' + url1 = f'{base}/token1/topicSubscriptions/test-topic?allow_missing=true' + url2 = f'{base}/token2/topicSubscriptions/test-topic?allow_missing=true' + url3 = f'{base}/token3/topicSubscriptions/test-topic?allow_missing=true' + route1 = respx.delete(url1).mock(return_value=respx.MockResponse(200, json={})) + route2 = respx.delete(url2).mock( + return_value=respx.MockResponse(404, json={'error': {'status': 'NOT_FOUND'}})) + route3 = respx.delete(url3).mock(return_value=respx.MockResponse(200, json={})) + + resp = await messaging.unsubscribe_from_topic_async( + ['token1', 'token2', 'token3'], 'test-topic') + assert route1.call_count == 1 + assert route2.call_count == 1 + assert route3.call_count == 1 + assert resp.success_count == 2 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 1 + assert resp.errors[0].reason == 'NOT_FOUND' + + @pytest.mark.asyncio + @pytest.mark.parametrize('tokens', [None, '', [], {}, tuple()]) + async def test_invalid_tokens(self, tokens): + expected = 'Tokens must be a string or a non-empty list of strings.' + if isinstance(tokens, str): + expected = 'Tokens must be non-empty strings.' + + with pytest.raises(ValueError) as excinfo: + await messaging.subscribe_to_topic_async(tokens, 'test-topic') + assert str(excinfo.value) == expected + + with pytest.raises(ValueError) as excinfo: + await messaging.unsubscribe_from_topic_async(tokens, 'test-topic') + assert str(excinfo.value) == expected + + @pytest.mark.asyncio + @pytest.mark.parametrize('tokens', [ + ['foo', 'bar', ''], + ['foo', 123, 'bar'], + ]) + async def test_invalid_tokens_in_list(self, tokens): + with pytest.raises(ValueError) as excinfo: + await messaging.subscribe_to_topic_async(tokens, 'test-topic') + assert str(excinfo.value) == 'Tokens must be non-empty strings.' + + with pytest.raises(ValueError) as excinfo: + await messaging.unsubscribe_from_topic_async(tokens, 'test-topic') + assert str(excinfo.value) == 'Tokens must be non-empty strings.' + + @pytest.mark.asyncio + async def test_tokens_over_1000(self): + tokens = [f'token{i}' for i in range(1001)] + with pytest.raises(ValueError) as excinfo: + await messaging.subscribe_to_topic_async(tokens, 'test-topic') + assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.' + + with pytest.raises(ValueError) as excinfo: + await messaging.unsubscribe_from_topic_async(tokens, 'test-topic') + assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.' + + @pytest.mark.asyncio + @pytest.mark.parametrize('topic', NON_STRING_ARGS + [None, '']) + async def test_invalid_topic(self, topic): + expected = 'Topic must be a non-empty string.' + with pytest.raises(ValueError) as excinfo: + await messaging.subscribe_to_topic_async('test-token', topic) + assert str(excinfo.value) == expected + + with pytest.raises(ValueError) as excinfo: + await messaging.unsubscribe_from_topic_async('test-token', topic) + assert str(excinfo.value) == expected + + @pytest.mark.asyncio + @pytest.mark.parametrize('topic', [ + '/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&', + '/topics/foo\n', 'foo\n', + ]) + async def test_malformed_topic(self, topic): + with pytest.raises(ValueError) as excinfo: + await messaging.subscribe_to_topic_async('test-token', topic) + assert str(excinfo.value) == 'Malformed topic name.' + + with pytest.raises(ValueError) as excinfo: + await messaging.unsubscribe_from_topic_async('test-token', topic) + assert str(excinfo.value) == 'Malformed topic name.' + + @pytest.mark.asyncio + @respx.mock + async def test_topic_management_async_fcm_error_details(self): + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) + payload = { + 'error': { + 'status': 'NOT_FOUND', + 'details': [ + { + '@type': 'type.googleapis.com/google.firebase.fcm.v1.FcmError', + 'errorCode': 'UNREGISTERED', + }, + ], + } + } + respx.post(url).mock(return_value=respx.MockResponse(404, json=payload)) + resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'UNREGISTERED' + + @pytest.mark.asyncio + @respx.mock + async def test_topic_management_async_500_error(self): + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) + respx.post(url).mock(return_value=respx.MockResponse(500, json={'error': None})) + resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'INTERNAL' + + @pytest.mark.asyncio + @respx.mock + async def test_topic_management_async_non_json_error(self): + url = ( + 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/' + 'token1/topicSubscriptions?topic_name=test-topic' + ) + respx.post(url).mock(return_value=respx.MockResponse(400, text='not json')) + resp = await messaging.subscribe_to_topic_async('token1', 'test-topic') + assert resp.success_count == 0 + assert resp.failure_count == 1 + assert len(resp.errors) == 1 + assert resp.errors[0].index == 0 + assert resp.errors[0].reason == 'INVALID_ARGUMENT'