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'