Use fast_sleep to accelerate tests that intentionally have missing fi…

…les. Open-EO/openeo-geotrellis-extensions#329

tests/conftest.py

@@ -1,16 +1,18 @@
-import os
-from typing import Union
-
-import sys
-from pathlib import Path
+from unittest import mock
 
 import boto3
+import contextlib
 import flask
-import pytest
 import moto
 import moto.server
+import os
+import pytest
+import sys
+import time_machine
+import typing
 import requests_mock
 from _pytest.terminal import TerminalReporter
+from pathlib import Path
 
 from openeo_driver.backend import OpenEoBackendImplementation, UserDefinedProcesses
 from openeo_driver.jobregistry import ElasticJobRegistry, JobRegistryInterface
@@ -180,6 +182,34 @@ def api_version(request):
     return request.param
 
 
+class _Sleeper:
+    def __init__(self):
+        self.history = []
+
+    @contextlib.contextmanager
+    def patch(self, time_machine: time_machine.TimeMachineFixture) -> typing.Iterator["_Sleeper"]:
+        def sleep(seconds):
+            # Note: this requires that `time_machine.move_to()` has been called before
+            # also see https://github.com/adamchainz/time-machine/issues/247
+            time_machine.coordinates.shift(seconds)
+            self.history.append(seconds)
+
+        with mock.patch("time.sleep", new=sleep):
+            yield self
+
+    def did_sleep(self) -> bool:
+        return len(self.history) > 0
+
+
+@pytest.fixture
+def fast_sleep(time_machine) -> typing.Iterator[_Sleeper]:
+    """
+    Fixture using `time_machine` to make `sleep` instant and update the current time.
+    """
+    with _Sleeper().patch(time_machine=time_machine) as sleeper:
+        yield sleeper
+
+
 @pytest.fixture
 def udf_noop():
     file_name = get_test_data_file("udf_noop.py")

tests/deploy/test_batch_job.py

@@ -391,7 +391,7 @@ def test_run_job(evaluate, tmp_path):
 
 
 @mock.patch("openeo_driver.ProcessGraphDeserializer.evaluate")
-def test_run_job_get_projection_extension_metadata(evaluate, tmp_path):
+def test_run_job_get_projection_extension_metadata(evaluate, tmp_path, fast_sleep):
     cube_mock = MagicMock()
 
     job_dir = tmp_path / "job-402"
@@ -517,9 +517,7 @@ def test_run_job_get_projection_extension_metadata(evaluate, tmp_path):
 
 
 @mock.patch("openeo_driver.ProcessGraphDeserializer.evaluate")
-def test_run_job_get_projection_extension_metadata_all_assets_same_epsg_and_bbox(
-    evaluate, tmp_path
-):
+def test_run_job_get_projection_extension_metadata_all_assets_same_epsg_and_bbox(evaluate, tmp_path, fast_sleep):
     """When there are two raster assets with the same projection metadata, it should put
     those metadata at the level of the item instead of the individual bands.
     """
@@ -959,7 +957,7 @@ def test_run_job_get_projection_extension_metadata_assets_with_different_epsg(
 
 
 @mock.patch("openeo_driver.ProcessGraphDeserializer.evaluate")
-def test_run_job_get_projection_extension_metadata_job_dir_is_relative_path(evaluate):
+def test_run_job_get_projection_extension_metadata_job_dir_is_relative_path(evaluate, fast_sleep):
     cube_mock = MagicMock()
     # job dir should be a relative path,
     # We still want the test data to be cleaned up though, so we need to use