add more unit tests per gemini review.

Author: taeold

tests/test_scheduler_fn.py

@@ -15,7 +15,7 @@
 
 import unittest
 from datetime import datetime, timezone
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 from flask import Flask, Request
 from werkzeug.test import EnvironBuilder
@@ -146,6 +146,90 @@ def test_on_schedule_call_with_exception(self):
             self.assertEqual(response.status_code, 500)
             self.assertEqual(response.data, b"Test exception")
 
+    def test_on_schedule_call_with_fractional_seconds(self):
+        """
+        Tests to ensure that timestamps with fractional seconds are parsed correctly.
+        """
+        with Flask(__name__).test_request_context("/"):
+            environ = EnvironBuilder(
+                headers={
+                    "X-CloudScheduler-JobName": "example-job",
+                    "X-CloudScheduler-ScheduleTime": "2023-04-13T19:00:00.123456Z",
+                }
+            ).get_environ()
+            mock_request = Request(environ)
+            example_func = Mock(__name__="example_func")
+            decorated_func = scheduler_fn.on_schedule(schedule="* * * * *")(example_func)
+            response = decorated_func(mock_request)
+
+            self.assertEqual(response.status_code, 200)
+            example_func.assert_called_once_with(
+                scheduler_fn.ScheduledEvent(
+                    job_name="example-job",
+                    schedule_time=datetime(2023, 4, 13, 19, 0, 0, 123456, tzinfo=timezone.utc),
+                )
+            )
+
+    def test_on_schedule_call_fallback_parsing(self):
+        """
+        Tests fallback parsing for formats that might fail fromisoformat
+        but pass strptime (e.g., offset without colon).
+        """
+        with Flask(__name__).test_request_context("/"):
+            environ = EnvironBuilder(
+                headers={
+                    "X-CloudScheduler-JobName": "example-job",
+                    # Offset without colon might fail fromisoformat in some versions
+                    # but should pass strptime("%Y-%m-%dT%H:%M:%S%z")
+                    "X-CloudScheduler-ScheduleTime": "2023-04-13T12:00:00-0700",
+                }
+            ).get_environ()
+            mock_request = Request(environ)
+            example_func = Mock(__name__="example_func")
+            decorated_func = scheduler_fn.on_schedule(schedule="* * * * *")(example_func)
+            response = decorated_func(mock_request)
+
+            self.assertEqual(response.status_code, 200)
+
+            # Create expected datetime with fixed offset -07:00
+            tz = datetime.strptime("-0700", "%z").tzinfo
+            expected_dt = datetime(2023, 4, 13, 12, 0, 0, tzinfo=tz)
+
+            example_func.assert_called_once_with(
+                scheduler_fn.ScheduledEvent(
+                    job_name="example-job",
+                    schedule_time=expected_dt,
+                )
+            )
+
+    def test_on_schedule_call_invalid_timestamp(self):
+        """
+        Tests that invalid timestamps log an error and fallback to current time.
+        """
+        with Flask(__name__).test_request_context("/"):
+            environ = EnvironBuilder(
+                headers={
+                    "X-CloudScheduler-JobName": "example-job",
+                    "X-CloudScheduler-ScheduleTime": "invalid-timestamp",
+                }
+            ).get_environ()
+            mock_request = Request(environ)
+            example_func = Mock(__name__="example_func")
+
+            with patch("firebase_functions.scheduler_fn._logging") as mock_logging:
+                decorated_func = scheduler_fn.on_schedule(schedule="* * * * *")(example_func)
+                response = decorated_func(mock_request)
+
+                self.assertEqual(response.status_code, 200)
+                mock_logging.exception.assert_called_once()
+
+                # Should have called with *some* time (current time), so we just check it's not None
+                self.assertEqual(example_func.call_count, 1)
+                called_event = example_func.call_args[0][0]
+                self.assertEqual(called_event.job_name, "example-job")
+                self.assertIsNotNone(called_event.schedule_time)
+                self.assertIsNotNone(called_event.schedule_time.tzinfo)
+
     def test_calls_init(self):
         hello = None