[System Tests] Fix timings in scheduled spark test (#1541)

mlrun · Dec 5, 2021 · 3ebb0cb · 3ebb0cb
1 parent 00ad602
commit 3ebb0cb
Showing 1 changed file with 9 additions and 6 deletions.
diff --git a/tests/system/feature_store/test_spark_engine.py b/tests/system/feature_store/test_spark_engine.py
@@ -1,5 +1,5 @@
 import os
-from datetime import datetime, timedelta
+from datetime import datetime
 from time import sleep
 
 import fsspec
@@ -91,7 +91,7 @@ def test_basic_remote_spark_ingest(self):
     def test_schedule_on_filtered_by_time(self, partitioned):
         name = f"sched-time-{str(partitioned)}"
 
-        now = datetime.now() + timedelta(minutes=2)
+        now = datetime.now()
 
         path = "v3io:///bigdata/bla.parquet"
         fsys = fsspec.filesystem(v3iofs.fs.V3ioFS.protocol)
@@ -106,7 +106,7 @@ def test_schedule_on_filtered_by_time(self, partitioned):
             }
         ).to_parquet(path=path, filesystem=fsys)
 
-        cron_trigger = "*/1 * * * *"
+        cron_trigger = "*/2 * * * *"
 
         source = ParquetSource(
             "myparquet", path=path, time_field="time", schedule=cron_trigger
@@ -144,8 +144,11 @@ def test_schedule_on_filtered_by_time(self, partitioned):
             targets=targets,
             spark_context=self.spark_service,
         )
-        # ingest starts every round minute.
-        sleep(60 - now.second + 50)
+        # ingest starts every second minute and it takes ~90 seconds to finish.
+        if (now.minute % 2) == 0:
+            sleep(60 - now.second + 60 + 90)
+        else:
+            sleep(60 - now.second + 90)
 
         features = [f"{name}.*"]
         vec = fs.FeatureVector("sched_test-vec", features)
@@ -169,7 +172,7 @@ def test_schedule_on_filtered_by_time(self, partitioned):
             }
         ).to_parquet(path=path)
 
-        sleep(60)
+        sleep(120)
         resp = svc.get(
             [
                 {"first_name": "yosi"},