fix(functions): Removed usage of deprecated datetime.utcnow()

firebase_admin/functions.py

@@ -15,7 +15,7 @@
 """Firebase Functions module."""
 
 from __future__ import annotations
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from urllib import parse
 import re
 import json
@@ -255,7 +255,8 @@ def _validate_task_options(
                 if not isinstance(opts.schedule_delay_seconds, int) \
                 or opts.schedule_delay_seconds < 0:
                     raise ValueError('schedule_delay_seconds should be positive int.')
-                schedule_time = datetime.utcnow() + timedelta(seconds=opts.schedule_delay_seconds)
+                schedule_time = (
+                    datetime.now(timezone.utc) + timedelta(seconds=opts.schedule_delay_seconds))
                 task.schedule_time = schedule_time.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
             if opts.dispatch_deadline_seconds is not None:
                 if not isinstance(opts.dispatch_deadline_seconds, int) \

tests/test_functions.py

@@ -14,7 +14,7 @@
 
 """Test cases for the firebase_admin.functions module."""
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import json
 import time
 import pytest
@@ -33,8 +33,6 @@
     _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/queues/test-function-name/tasks'
 _DEFAULT_TASK_URL = _CLOUD_TASKS_URL + _DEFAULT_TASK_PATH
 _DEFAULT_RESPONSE = json.dumps({'name': _DEFAULT_TASK_PATH})
-_ENQUEUE_TIME = datetime.utcnow()
-_SCHEDULE_TIME = _ENQUEUE_TIME + timedelta(seconds=100)
 
 class TestTaskQueue:
     @classmethod
@@ -185,47 +183,65 @@ def _instrument_functions_service(self, app=None, status=200, payload=_DEFAULT_R
             testutils.MockAdapter(payload, status, recorder))
         return functions_service, recorder
 
-
-    @pytest.mark.parametrize('task_opts_params', [
-        {
+    def test_task_options_delay_seconds(self):
+        _, recorder = self._instrument_functions_service()
+        enqueue_time = datetime.now(timezone.utc)
+        expected_schedule_time = enqueue_time + timedelta(seconds=100)
+        task_opts_params = {
             'schedule_delay_seconds': 100,
             'schedule_time': None,
             'dispatch_deadline_seconds': 200,
             'task_id': 'test-task-id',
             'headers': {'x-test-header': 'test-header-value'},
             'uri': 'https://google.com'
-        },
-        {
+        }
+        queue = functions.task_queue('test-function-name')
+        task_opts = functions.TaskOptions(**task_opts_params)
+        queue.enqueue(_DEFAULT_DATA, task_opts)
+
+        assert len(recorder) == 1
+        task = json.loads(recorder[0].body.decode())['task']
+
+        task_schedule_time = datetime.fromisoformat(task['schedule_time'].replace('Z', '+00:00'))
+        delta = abs(task_schedule_time - expected_schedule_time)
+        assert delta <= timedelta(seconds=1)
+
+        assert task['dispatch_deadline'] == '200s'
+        assert task['http_request']['headers']['x-test-header'] == 'test-header-value'
+        assert task['http_request']['url'] in ['http://google.com', 'https://google.com']
+        assert task['name'] == _DEFAULT_TASK_PATH
+
+    def test_task_options_utc_time(self):
+        _, recorder = self._instrument_functions_service()
+        enqueue_time = datetime.now(timezone.utc)
+        expected_schedule_time = enqueue_time + timedelta(seconds=100)
+        task_opts_params = {
             'schedule_delay_seconds': None,
-            'schedule_time': _SCHEDULE_TIME,
+            'schedule_time': expected_schedule_time,
             'dispatch_deadline_seconds': 200,
             'task_id': 'test-task-id',
             'headers': {'x-test-header': 'test-header-value'},
             'uri': 'http://google.com'
-        },
-    ])
-    def test_task_options(self, task_opts_params):
-        _, recorder = self._instrument_functions_service()
+        }
         queue = functions.task_queue('test-function-name')
         task_opts = functions.TaskOptions(**task_opts_params)
         queue.enqueue(_DEFAULT_DATA, task_opts)
 
         assert len(recorder) == 1
         task = json.loads(recorder[0].body.decode())['task']
 
-        schedule_time = datetime.fromisoformat(task['schedule_time'][:-1])
-        delta = abs(schedule_time - _SCHEDULE_TIME)
-        assert delta <= timedelta(seconds=15)
+        task_schedule_time = datetime.fromisoformat(task['schedule_time'].replace('Z', '+00:00'))
+        assert task_schedule_time == expected_schedule_time
 
         assert task['dispatch_deadline'] == '200s'
         assert task['http_request']['headers']['x-test-header'] == 'test-header-value'
         assert task['http_request']['url'] in ['http://google.com', 'https://google.com']
         assert task['name'] == _DEFAULT_TASK_PATH
 
-
     def test_schedule_set_twice_error(self):
         _, recorder = self._instrument_functions_service()
-        opts = functions.TaskOptions(schedule_delay_seconds=100, schedule_time=datetime.utcnow())
+        opts = functions.TaskOptions(
+            schedule_delay_seconds=100, schedule_time=datetime.now(timezone.utc))
         queue = functions.task_queue('test-function-name')
         with pytest.raises(ValueError) as excinfo:
             queue.enqueue(_DEFAULT_DATA, opts)
@@ -236,9 +252,9 @@ def test_schedule_set_twice_error(self):
 
     @pytest.mark.parametrize('schedule_time', [
         time.time(),
-        str(datetime.utcnow()),
-        datetime.utcnow().isoformat(),
-        datetime.utcnow().isoformat() + 'Z',
+        str(datetime.now(timezone.utc)),
+        datetime.now(timezone.utc).isoformat(),
+        datetime.now(timezone.utc).isoformat() + 'Z',
         '', ' '
     ])
     def test_invalid_schedule_time_error(self, schedule_time):