Eliminate deprecated utcnow usage patterns

and get rid of old time mocking hacks

see Open-EO/openeo-python-driver#389

Author: soxofaan

openeogeotrellis/backend.py

@@ -64,6 +64,7 @@
     _extract_load_parameters, ENV_MAX_BUFFER
 from openeo_driver.save_result import ImageCollectionResult
 from openeo_driver.users import User
+from openeo_driver.util.date_math import now_utc
 from openeo_driver.util.geometry import BoundingBox
 from openeo_driver.util.http import requests_with_retry
 from openeo_driver.util.utm import area_in_square_meters
@@ -220,18 +221,22 @@ def _create_service(self, user_id: str, process_graph: dict, service_type: str,
             logger.info("Can not create service for: " + str(image_collection))
             raise OpenEOApiException("Can not create service for: " + str(image_collection))
 
-
-        wmts_base_url = os.getenv('WMTS_BASE_URL_PATTERN', 'http://openeo.vgt.vito.be/openeo/services/%s') % service_id
-
-        self.service_registry.persist(user_id, ServiceMetadata(
-            id=service_id,
-            process={"process_graph": process_graph},
-            url=wmts_base_url + "/service/wmts",
-            type=service_type,
-            enabled=True,
-            attributes={},
-            configuration=configuration,
-            created=dt.datetime.utcnow()), api_version)
+        wmts_base_url = os.getenv("WMTS_BASE_URL_PATTERN", "http://openeo.vgt.vito.be/openeo/services/%s") % service_id
+
+        self.service_registry.persist(
+            user_id,
+            ServiceMetadata(
+                id=service_id,
+                process={"process_graph": process_graph},
+                url=wmts_base_url + "/service/wmts",
+                type=service_type,
+                enabled=True,
+                attributes={},
+                configuration=configuration,
+                created=now_utc(),
+            ),
+            api_version,
+        )
 
         secondary_service = self._wmts_service(image_collection, configuration, wmts_base_url)
 

openeogeotrellis/config/s3_config.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
+
 import re
 from configparser import ConfigParser
 from io import StringIO
 import os
 from dataclasses import dataclass
 from typing import Optional
+import time
 
 from openeogeotrellis.config import get_backend_config
-from openeogeotrellis.utils import utcnow_epoch
 from openeogeotrellis.workspace import ObjectStorageWorkspace
 
 
@@ -117,5 +118,5 @@ def _sanitize_session_name(session_name: str) -> str:
         """
         sanitized = re.sub(r"[^\w+=,.@-]", "", session_name)
         if len(sanitized) < 2:
-            sanitized += str(f"-{utcnow_epoch()}")
+            sanitized += str(f"-{time.time()}")
         return sanitized[0:64]

openeogeotrellis/deploy/batch_job_cfg_secret.yaml.j2

@@ -6,8 +6,8 @@ metadata:
   labels:
     job_id: {{ job_id }}
   annotations:
-    created_at: "{{ utcnow_epoch() }}"
+    created_at: "{{ unix_time() }}"
 data:
   profile_file: {{ profile_file_content | b64encode }}
   token: {{ token | b64encode }}
-type: Opaque
+type: Opaque

openeogeotrellis/integrations/identity.py

@@ -10,8 +10,9 @@
 
 import jwt
 
+from openeo_driver.util.date_math import now_utc
 from openeogeotrellis.config import get_backend_config
-from openeogeotrellis.utils import FileChangeWatcher, utcnow
+from openeogeotrellis.utils import FileChangeWatcher
 
 logger = logging.getLogger(__name__)
 
@@ -99,7 +100,7 @@ def get_job_token(self, sub_id: str, user_id: str, job_id: str) -> Optional[str]
             return None
         else:
             idp_details = self._IDP_DETAILS
-            now = utcnow()
+            now = now_utc()
             return jwt.encode(
                 {
                     "sub": sub_id,

openeogeotrellis/integrations/kubernetes.py

@@ -5,12 +5,13 @@
 import logging
 import os
 import pkg_resources
+import time
 
 from jinja2 import Environment, FileSystemLoader
+
 from openeo_driver.jobregistry import JOB_STATUS
 from openeo_driver.utils import generate_unique_id
 
-from openeogeotrellis.utils import utcnow_epoch
 
 _log = logging.getLogger(__name__)
 
@@ -97,8 +98,9 @@ def base64encode(input_str: str) -> str:
             input_str = str(input_str)
         return base64.b64encode(input_str.encode("utf-8")).decode("utf-8")
 
-    jinja_env.filters['b64encode'] = base64encode
-    jinja_env.globals['utcnow_epoch'] = utcnow_epoch
+    jinja_env.filters["b64encode"] = base64encode
+    jinja_env.globals["utcnow_epoch"] = time.time  # TODO: remove this deprecated/unused utility
+    jinja_env.globals["unix_time"] = time.time
     jinja_template = jinja_env.from_string(open(jinja_path).read())
 
     rendered = jinja_template.render(**kwargs)

openeogeotrellis/job_registry.py

@@ -100,8 +100,8 @@ def register(
             'api_version': api_version,
             "specification": specification_blob,
             "application_id": None,
-            "created": rfc3339.utcnow(),
-            "updated": rfc3339.utcnow(),
+            "created": rfc3339.now_utc(),
+            "updated": rfc3339.now_utc(),
             "title": title,
             "description": description,
         }
@@ -120,7 +120,7 @@ def set_status(
         """Updates a registered batch job with its status. Additionally, updates its "updated" property."""
         kwargs = {
             "status": status,
-            "updated": rfc3339.utcnow(),
+            "updated": rfc3339.now_utc(),
         }
 
         if started:
@@ -608,7 +608,7 @@ def create_job(
         job_options: Optional[dict] = None,
     ) -> JobDict:
         assert job_id not in self.db
-        created = rfc3339.utcnow()
+        created = rfc3339.now_utc()
         self.db[job_id] = {
             "job_id": job_id,
             "user_id": user_id,
@@ -655,7 +655,7 @@ def set_status(
         self._update(
             job_id=job_id,
             status=status,
-            updated=rfc3339.datetime(updated) if updated else rfc3339.utcnow(),
+            updated=rfc3339.datetime(updated) if updated else rfc3339.now_utc(),
         )
         if started:
             self._update(job_id=job_id, started=rfc3339.datetime(started))

openeogeotrellis/utils.py

@@ -25,6 +25,8 @@
 import pytz
 from epsel import on_first_time
 from kazoo.client import KazooClient
+
+from openeo.util import rfc3339
 from openeo_driver.datacube import DriverVectorCube
 from openeo_driver.delayed_vector import DelayedVector
 from openeo_driver.util.geometry import GeometryBufferer, reproject_bounding_box
@@ -128,45 +130,10 @@ def normalize_temporal_extent(temporal_extent: Tuple[Union[str, None], Union[str
     start, end = temporal_extent
     return (
         normalize_date(start or "2000-01-01"),  # TODO: better fallback start date?
-        normalize_date(end or utcnow().isoformat())
+        normalize_date(end or rfc3339.now_utc()),
     )
 
 
-class UtcNowClock:
-    """
-    Helper class to have a mockable wrapper for datetime.datetime.utcnow
-    (which is not straightforward to mock directly).
-    """
-
-    # TODO: just start using `time_machine` module for time mocking
-
-    _utcnow = _utcnow_orig = datetime.datetime.utcnow
-
-    @classmethod
-    def utcnow(cls) -> datetime.datetime:
-        return cls._utcnow()
-
-    @classmethod
-    @contextlib.contextmanager
-    def mock(cls, now: Union[datetime.datetime, str]):
-        """Context manager to mock the return value of `utcnow()`."""
-        if isinstance(now, str):
-            now = dateutil.parser.parse(now)
-        cls._utcnow = lambda: now
-        try:
-            yield
-        finally:
-            cls._utcnow = cls._utcnow_orig
-
-
-# Alias for general usage
-utcnow = UtcNowClock.utcnow
-
-
-def utcnow_epoch() -> float:
-    return utcnow().timestamp()
-
-
 def describe_path(path: Union[Path, str]) -> dict:
     path = Path(path)
     if path.exists() or path.is_symlink():

tests/test_api_result.py

@@ -64,7 +64,6 @@
 from openeogeotrellis.job_registry import ZkJobRegistry
 from openeogeotrellis.testing import KazooClientMock, gps_config_overrides, random_name
 from openeogeotrellis.utils import (
-    UtcNowClock,
     drop_empty_from_aggregate_polygon_result,
     get_jvm,
     is_package_available,
@@ -2015,9 +2014,10 @@ def test_extra_validation_unlimited_extent(api100, lc_args):
     (("2020-01-01", None), ("2020-01-05 00:00:00", "2020-02-15 00:00:00")),
     ((None, "2019-01-10"), ("2000-01-05 00:00:00", "2019-01-05 00:00:00")),
 ])
-def test_load_collection_open_temporal_extent(api100, temporal_extent, expected):
-    with UtcNowClock.mock(now="2020-02-20"):
-        response = api100.check_result({
+def test_load_collection_open_temporal_extent(api100, temporal_extent, expected, time_machine):
+    time_machine.move_to("2020-02-20")
+    response = api100.check_result(
+        {
             "lc": {
                 "process_id": "load_collection",
                 "arguments": {

tests/test_backend.py

@@ -26,7 +26,6 @@
 from openeogeotrellis.config.s3_config import S3Config
 from openeogeotrellis.integrations.kubernetes import k8s_render_manifest_template
 from openeogeotrellis.testing import gps_config_overrides
-from openeogeotrellis.utils import UtcNowClock, utcnow
 
 
 def test_extract_application_id():
@@ -674,28 +673,31 @@ def test_k8s_s3_profiles_and_token():
     )
 
 
-def test_k8s_s3_profiles_and_token_must_be_cleanable(backend_config_path, fast_sleep):
-    # GIVEN a specific time
-    test_timestamp = utcnow()
-    test_timestamp_epoch = test_timestamp.timestamp()
-
-    # GIVEN the green infinity stone (to control the time dimension)
-    with UtcNowClock.mock(now=test_timestamp):
-        # WHEN we render the kubernetes manifest for s3 access
-        token_path = get_backend_config().batch_job_config_dir / "token"
-        app_dict = k8s_render_manifest_template(
-            "batch_job_cfg_secret.yaml.j2",
-            secret_name="my-app",
-            job_id="test_id",
-            token="test",
-            profile_file_content=S3Config.from_backend_config("j-any", str(token_path))
-        )
+def test_k8s_s3_profiles_and_token_must_be_cleanable(backend_config_path, fast_sleep, time_machine):
+    time_machine.move_to(1745410732)  # Wed 2025-04-23 12:18:52 UTC)
+    # WHEN we render the kubernetes manifest for s3 access
+    token_path = get_backend_config().batch_job_config_dir / "token"
+    app_dict = k8s_render_manifest_template(
+        "batch_job_cfg_secret.yaml.j2",
+        secret_name="my-app",
+        job_id="test_id",
+        token="test",
+        profile_file_content=S3Config.from_backend_config("j-any", str(token_path)),
+    )
     # THEN we expect it to be cleanable by having a starttime set to the time it was created.
     # We can only clean files if we know they are stale
     assert app_dict == dirty_equals.IsPartialDict(
-        metadata=dirty_equals.IsPartialDict(
-            annotations=dirty_equals.IsPartialDict(created_at=str(test_timestamp_epoch))
-        ),
+        {
+            "apiVersion": "v1",
+            "kind": "Secret",
+            "metadata": dirty_equals.IsPartialDict(
+                {
+                    "labels": {"job_id": "test_id"},
+                    "name": "my-app",
+                    "annotations": dirty_equals.IsPartialDict(created_at="1745403532.0"),
+                }
+            ),
+        }
     )
 
 

tests/test_job_tracker_v2.py

@@ -267,7 +267,7 @@ def set_state(self, state: str):
 
     def set_submitted(self):
         if not self.start_time:
-            self.start_time = rfc3339.datetime(dt.datetime.utcnow())
+            self.start_time = rfc3339.now_utc()
         self.state = K8S_SPARK_APP_STATE.SUBMITTED
 
     def set_running(self):
@@ -283,7 +283,7 @@ def set_failed(self):
 
     def set_finish_time(self, force: bool = False):
         if not self.finish_time or force:
-            self.finish_time = rfc3339.datetime(dt.datetime.utcnow())
+            self.finish_time = rfc3339.now_utc()
 
 
 class KubernetesMock:
@@ -1625,8 +1625,8 @@ def test_cpu_and_memory_usage_not_in_prometheus(self, caplog):
         k8s_mock.get_namespaced_custom_object.return_value = {
             "status": {
                 "applicationState": {"state": K8S_SPARK_APP_STATE.COMPLETED},
-                "lastSubmissionAttemptTime": rfc3339.datetime(dt.datetime.utcnow()),
-                "terminationTime": rfc3339.datetime(dt.datetime.utcnow()),
+                "lastSubmissionAttemptTime": rfc3339.now_utc(),
+                "terminationTime": rfc3339.now_utc(),
             }
         }
 

tests/test_load_stac.py

@@ -7,6 +7,7 @@
 from openeo_driver.ProcessGraphDeserializer import DEFAULT_TEMPORAL_EXTENT
 from openeo_driver.backend import BatchJobMetadata, BatchJobs, LoadParameters
 from openeo_driver.errors import OpenEOApiException
+from openeo_driver.util.date_math import now_utc
 from openeo_driver.utils import EvalEnv
 
 from openeogeotrellis.load_stac import extract_own_job_info, load_stac
@@ -25,8 +26,11 @@ def test_extract_own_job_info(url, user_id, job_info_id):
     batch_jobs = mock.Mock(spec=BatchJobs)
 
     def alices_single_job(job_id, user_id):
-        return (BatchJobMetadata(id=job_id, status='finished', created=dt.datetime.utcnow())
-                if job_id == 'j-20240201abc123' and user_id == 'alice' else None)
+        return (
+            BatchJobMetadata(id=job_id, status="finished", created=now_utc())
+            if job_id == "j-20240201abc123" and user_id == "alice"
+            else None
+        )
 
     batch_jobs.get_job_info.side_effect = alices_single_job
 

tests/test_object_storage_workspace.py

@@ -6,6 +6,7 @@
 from pystac import Collection, Extent, SpatialExtent, TemporalExtent, Item, CatalogType, Asset
 import pytest
 
+from openeo_driver.util.date_math import now_utc
 from openeogeotrellis.workspace import ObjectStorageWorkspace
 from openeogeotrellis.workspace.custom_stac_io import CustomStacIO
 
@@ -237,7 +238,7 @@ def _collection(
         asset_href_parts = urlparse(asset_href)
         asset_filename = asset_href_parts.path.split("/")[-1]
 
-        item = Item(id=asset_filename, geometry=None, bbox=None, datetime=dt.datetime.utcnow(), properties={})
+        item = Item(id=asset_filename, geometry=None, bbox=None, datetime=now_utc(), properties={})
         asset = Asset(href=asset_href)
 
         item.add_asset(key=asset_filename, asset=asset)