add waiters (#3)

Author: skshetry

pytest_test_utils/__init__.py

@@ -1,9 +1,10 @@
-from . import matchers
+from . import matchers, waiters
 from .tmp_dir import TmpDir
 from .tmp_dir_factory import TempDirFactory
 
 __all__ = [
     "matchers",
+    "waiters",
     "TmpDir",
     "TempDirFactory",
 ]

pytest_test_utils/matchers.py

@@ -10,7 +10,6 @@
     Optional,
     Pattern,
     Tuple,
-    Type,
     Union,
 )
 
@@ -124,7 +123,7 @@ def __eq__(self, other: object) -> bool:
 class instance_of:
     def __init__(
         self,
-        expected_type: Union[Type[object], Tuple[Type[object], ...]],
+        expected_type: Union[Any, Tuple[Any, ...]],
     ) -> None:
         self.expected_type = expected_type
 
@@ -198,9 +197,7 @@ def any_of(*items: Any) -> any_of:
         return any_of(*items)
 
     @staticmethod
-    def instance_of(
-        expected_type: Union[Type[object], Tuple[Type[object], ...]]
-    ) -> instance_of:
+    def instance_of(expected_type: Union[Any, Tuple[Any, ...]]) -> instance_of:
         return instance_of(expected_type)
 
     @staticmethod

pytest_test_utils/waiters.py

@@ -0,0 +1,18 @@
+from time import perf_counter, sleep
+from typing import Callable, TypeVar
+
+_T = TypeVar("_T")
+
+
+class TimedOutError(Exception):
+    pass
+
+
+def wait_until(pred: Callable[[], _T], timeout: float, pause: float = 1) -> _T:
+    start = perf_counter()
+    while (perf_counter() - start) < timeout:
+        value = pred()
+        if value:
+            return value
+        sleep(pause)
+    raise TimedOutError("Timeout reached while waiting")

tests.py

@@ -1,14 +1,17 @@
 import os
 from datetime import datetime, timedelta
 from pathlib import Path
+from time import perf_counter
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Type
+from unittest.mock import MagicMock
 
 import pytest
 
 from pytest_test_utils import TmpDir, matchers
 from pytest_test_utils.matchers import Matcher
 from pytest_test_utils.tmp_dir_factory import TempDirFactory
+from pytest_test_utils.waiters import TimedOutError, wait_until
 
 if TYPE_CHECKING:
     from pytest import TempPathFactory
@@ -266,3 +269,24 @@ def test_matcher_alias(  # pylint: disable=invalid-name
     M: Type[Matcher], matcher: Type[Matcher]
 ) -> None:
     assert matcher is M is Matcher
+
+
+def test_wait_until() -> None:
+    pred = MagicMock(side_effect=[False, False, True])
+
+    start = perf_counter()
+    assert wait_until(pred, 0.01, pause=0.0001)
+    assert perf_counter() == pytest.approx(start + 0.0002, rel=1e-3)
+    assert len(pred.call_args_list) == 3
+
+
+def test_wait_until_raises_timedouterror() -> None:
+    pred = MagicMock(return_value=False)
+
+    start = perf_counter()
+
+    with pytest.raises(TimedOutError):
+        wait_until(pred, 0.01, pause=0.0001)
+
+    assert perf_counter() == pytest.approx(start + 0.01, rel=1e-3)
+    assert len(pred.call_args_list) > 1