fix: lint

Author: jonathanedey

firebase_admin/functions.py

@@ -193,7 +193,7 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str:
                 'post',
                 url=service_url,
                 headers=_FUNCTIONS_HEADERS,
-                json={'task': task_payload._to_api_dict()}
+                json={'task': task_payload.to_api_dict()}
             )
             if self._is_emulated():
                 # Emulator returns a response with format {task: {name: <task_name>}}
@@ -339,27 +339,28 @@ def _update_task_payload(self, task: Task, resource: Resource, extension_id: str
             # Delete oidc token
             del task.http_request['oidcToken']
         else:
-            try: 
+            try:
                 task.http_request['oidcToken'] = \
                     {'serviceAccountEmail': self._credential.service_account_email}
-            except AttributeError as e:
+            except AttributeError as error:
                 if self._is_emulated():
                     task.http_request['oidcToken'] = \
                         {'serviceAccountEmail': _EMULATED_SERVICE_ACCOUNT_DEFAULT}
                 else:
                     raise ValueError(
-                        'Failed to determine service account. Initialize the SDK with service account '
-                        'credentials or set service account ID as an app option.') from e
+                        'Failed to determine service account. Initialize the SDK with service '
+                        'account credentials or set service account ID as an app option.'
+                        ) from error
         return task
-    
+
     def _get_emulator_url(self, resource: Resource):
         emulator_host = _get_emulator_host()
         if emulator_host:
             emulator_url_format = f'http://{emulator_host}/' + _CLOUD_TASKS_API_RESOURCE_PATH
             url = self._get_url(resource, emulator_url_format)
             return url
         return None
-    
+
     def _is_emulated(self):
         return _get_emulator_host() is not None
 
@@ -487,7 +488,7 @@ class Task:
     schedule_time: Optional[str] = None
     dispatch_deadline: Optional[str] = None
 
-    def _to_api_dict(self) -> dict:
+    def to_api_dict(self) -> dict:
         """Converts the Task object to a dictionary suitable for the Cloud Tasks API."""
         return {
             'httpRequest': self.http_request,

tests/test_functions.py

@@ -16,7 +16,6 @@
 
 from datetime import datetime, timedelta, timezone
 import json
-import os
 import time
 import pytest
 
@@ -45,7 +44,8 @@ def setup_class(cls):
     def teardown_class(cls):
         testutils.cleanup_apps()
 
-    def _instrument_functions_service(self, app=None, status=200, payload=_DEFAULT_RESPONSE, mounted_url=_CLOUD_TASKS_URL):
+    def _instrument_functions_service(
+        self, app=None, status=200, payload=_DEFAULT_RESPONSE, mounted_url=_CLOUD_TASKS_URL):
         if not app:
             app = firebase_admin.get_app()
         functions_service = functions._get_functions_service(app)
@@ -214,37 +214,45 @@ def test_task_enqueue_with_emulator_host(self, monkeypatch):
         emulator_host = 'localhost:8124'
         emulator_url = f'http://{emulator_host}/'
         request_url = emulator_url + _DEFAULT_TASK_PATH.replace('/tasks/test-task-id', '/tasks')
-        
+
         monkeypatch.setenv('CLOUD_TASKS_EMULATOR_HOST', emulator_host)
-        app = firebase_admin.initialize_app(_utils.EmulatorAdminCredentials(), {'projectId': 'test-project'}, name='emulator-app')
-        
-        expected_response = json.dumps({'task': {'name': '/projects/test-project/locations/us-central1/queues/test-function-name/tasks/test-task-id'}})
-        _, recorder = self._instrument_functions_service(app, payload=expected_response, mounted_url=emulator_url)
+        app = firebase_admin.initialize_app(
+            _utils.EmulatorAdminCredentials(), {'projectId': 'test-project'}, name='emulator-app')
+
+        expected_task_name = (
+            '/projects/test-project/locations/us-central1'
+            '/queues/test-function-name/tasks/test-task-id'
+        )
+        expected_response = json.dumps({'task': {'name': expected_task_name}})
+        _, recorder = self._instrument_functions_service(
+            app, payload=expected_response, mounted_url=emulator_url)
 
         queue = functions.task_queue('test-function-name', app=app)
         task_id = queue.enqueue(_DEFAULT_DATA)
-        
+
         assert len(recorder) == 1
         assert recorder[0].method == 'POST'
         assert recorder[0].url == request_url
         assert recorder[0].headers['Content-Type'] == 'application/json'
-        
+
         task = json.loads(recorder[0].body.decode())['task']
         assert task['httpRequest']['oidcToken'] == {
             'serviceAccountEmail': '[email protected]'
         }
         assert task_id == 'test-task-id'
 
     def test_task_enqueue_without_emulator_host_error(self, monkeypatch):
-        app = firebase_admin.initialize_app(_utils.EmulatorAdminCredentials(), {'projectId': 'test-project'}, name='no-emulator-app')
-        
+        app = firebase_admin.initialize_app(
+            _utils.EmulatorAdminCredentials(),
+            {'projectId': 'test-project'}, name='no-emulator-app')
+
         _, recorder = self._instrument_functions_service(app)
         monkeypatch.delenv('CLOUD_TASKS_EMULATOR_HOST', raising=False)
         queue = functions.task_queue('test-function-name', app=app)
         with pytest.raises(ValueError) as excinfo:
             queue.enqueue(_DEFAULT_DATA)
         assert "Failed to determine service account" in str(excinfo.value)
-        assert len(recorder) == 0    
+        assert len(recorder) == 0
 
     def test_get_emulator_url_invalid_format(self, monkeypatch):
         _, recorder = self._instrument_functions_service()