pyrox.dev
1import time
2from math import isfinite
3from typing import Callable, Optional
4
5from test_driver.logger import AbstractLogger
6
7
8class PollingConditionError(Exception):
9 pass
10
11
12class PollingCondition:
13 condition: Callable[[], bool]
14 seconds_interval: float
15 description: Optional[str]
16 logger: AbstractLogger
17
18 last_called: float
19 entry_count: int
20
21 def __init__(
22 self,
23 condition: Callable[[], Optional[bool]],
24 logger: AbstractLogger,
25 seconds_interval: float = 2.0,
26 description: Optional[str] = None,
27 ):
28 self.condition = condition # type: ignore
29 self.seconds_interval = seconds_interval
30 self.logger = logger
31
32 if description is None:
33 if condition.__doc__:
34 self.description = condition.__doc__
35 else:
36 self.description = condition.__name__
37 else:
38 self.description = str(description)
39
40 self.last_called = float("-inf")
41 self.entry_count = 0
42
43 def check(self, force: bool = False) -> bool:
44 if (self.entered or not self.overdue) and not force:
45 return True
46
47 with self, self.logger.nested(self.nested_message):
48 time_since_last = time.monotonic() - self.last_called
49 last_message = (
50 f"Time since last: {time_since_last:.2f}s"
51 if isfinite(time_since_last)
52 else "(not called yet)"
53 )
54
55 self.logger.info(last_message)
56 try:
57 res = self.condition() # type: ignore
58 except Exception:
59 res = False
60 res = res is None or res
61 self.logger.info(self.status_message(res))
62 return res
63
64 def maybe_raise(self) -> None:
65 if not self.check():
66 raise PollingConditionError(self.status_message(False))
67
68 def status_message(self, status: bool) -> str:
69 return f"Polling condition {'succeeded' if status else 'failed'}: {self.description}"
70
71 @property
72 def nested_message(self) -> str:
73 nested_message = ["Checking polling condition"]
74 if self.description is not None:
75 nested_message.append(repr(self.description))
76
77 return " ".join(nested_message)
78
79 @property
80 def overdue(self) -> bool:
81 return self.last_called + self.seconds_interval < time.monotonic()
82
83 @property
84 def entered(self) -> bool:
85 # entry_count should never dip *below* zero
86 assert self.entry_count >= 0
87 return self.entry_count > 0
88
89 def __enter__(self) -> None:
90 self.entry_count += 1
91
92 def __exit__(self, exc_type, exc_value, traceback) -> None: # type: ignore
93 assert self.entered
94 self.entry_count -= 1
95 self.last_called = time.monotonic()