nixos/lib/test-driver/test_driver/polling_condition.py

   1 import time
   2 from math import isfinite
   3 from typing import Callable, Optional
   4
   5 from test_driver.logger import AbstractLogger
   6
   7
   8 class PollingConditionError(Exception):
   9     pass
  10
  11
  12 class 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()