Add tests to capture abnormal consecutive execution of scheduled tasks

Author: unode

tests/unit_tests/scheduler_test.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 from pathlib import Path
 from typing import Dict
+from unittest.mock import Mock
 
 import pytest
 
@@ -31,6 +32,37 @@ def job(path: str):
         assert float(file.readline()) - 2 == pytest.approx(start, abs=0.3)
 
 
+def test_once_single_call():
+    mock = Mock()
+    mock.side_effect = lambda: time.sleep(0.2)
+
+    schedule.once().do(mock)
+
+    for _ in range(10):
+        schedule.run_pending()
+        time.sleep(0.05)
+
+    mock.assert_called_once()
+
+
+def test_recurring_single_call():
+    mock = Mock()
+    mock.side_effect = lambda: time.sleep(0.2)
+
+    schedule.every(2).seconds.do(mock)
+
+    # Wait 2 seconds so we can run the task once
+    time.sleep(2)
+
+    # This loop corresponds to 0.1 seconds of total time and while there will
+    # be 10 calls to run_pending() the mock function should only run once
+    for _ in range(10):
+        schedule.run_pending()
+        time.sleep(0.01)
+
+    mock.assert_called_once()
+
+
 def test_recurring_thread():
     def job(modifiable_arg: Dict):
         # Modify the variable, which should be shared with the main thread.