commit 57beeedbdff89e3d1fbe80d6445e4925c36205e4 · pyrox.dev/nixpkgs

+22 -7

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

···

       1
        
       from typing import Callable, Optional

     

       0
        
       
     

       2
        
       import time

     

       3
        
       

     

       4
        
       from .logger import rootlog

     
···

       14
        
           description: Optional[str]

     

       15
        
       

     

       16
        
           last_called: float

     

       17
       -
           entered: bool

     

       18
        
       

     

       19
        
           def __init__(

     

       20
        
               self,

     
···

       34
        
                   self.description = str(description)

     

       35
        
       

     

       36
        
               self.last_called = float("-inf")

     

       37
       -
               self.entered = False

     

       38
        
       

     

       39
       -
           def check(self) -> bool:

     

       40
       -
               if self.entered or not self.overdue:

     

       41
        
                   return True

     

       42
        
       

     

       43
        
               with self, rootlog.nested(self.nested_message):

     

       44
       -
                   rootlog.info(f"Time since last: {time.monotonic() - self.last_called:.2f}s")

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       45
        
                   try:

     

       46
        
                       res = self.condition()  # type: ignore

     

       47
        
                   except Exception:

     
···

       69
        
           def overdue(self) -> bool:

     

       70
        
               return self.last_called + self.seconds_interval < time.monotonic()

     

       71
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       72
        
           def __enter__(self) -> None:

     

       73
       -
               self.entered = True

     

       74
        
       

     

       75
        
           def __exit__(self, exc_type, exc_value, traceback) -> None:  # type: ignore

     

       76
       -
               self.entered = False

     

       0
        
       
     

       77
        
               self.last_called = time.monotonic()

···

       1
        
       from typing import Callable, Optional

     

       2
       +
       from math import isfinite

     

       3
        
       import time

     

       4
        
       

     

       5
        
       from .logger import rootlog

     
···

       15
        
           description: Optional[str]

     

       16
        
       

     

       17
        
           last_called: float

     

       18
       +
           entry_count: int

     

       19
        
       

     

       20
        
           def __init__(

     

       21
        
               self,

     
···

       35
        
                   self.description = str(description)

     

       36
        
       

     

       37
        
               self.last_called = float("-inf")

     

       38
       +
               self.entry_count = 0

     

       39
        
       

     

       40
       +
           def check(self, force: bool = False) -> bool:

     

       41
       +
               if (self.entered or not self.overdue) and not force:

     

       42
        
                   return True

     

       43
        
       

     

       44
        
               with self, rootlog.nested(self.nested_message):

     

       45
       +
                   time_since_last = time.monotonic() - self.last_called

     

       46
       +
                   last_message = (

     

       47
       +
                       f"Time since last: {time_since_last:.2f}s"

     

       48
       +
                       if isfinite(time_since_last)

     

       49
       +
                       else "(not called yet)"

     

       50
       +
                   )

     

       51
       +
       

     

       52
       +
                   rootlog.info(last_message)

     

       53
        
                   try:

     

       54
        
                       res = self.condition()  # type: ignore

     

       55
        
                   except Exception:

     
···

       77
        
           def overdue(self) -> bool:

     

       78
        
               return self.last_called + self.seconds_interval < time.monotonic()

     

       79
        
       

     

       80
       +
           @property

     

       81
       +
           def entered(self) -> bool:

     

       82
       +
               # entry_count should never dip *below* zero

     

       83
       +
               assert self.entry_count >= 0

     

       84
       +
               return self.entry_count > 0

     

       85
       +
       

     

       86
        
           def __enter__(self) -> None:

     

       87
       +
               self.entry_count += 1

     

       88
        
       

     

       89
        
           def __exit__(self, exc_type, exc_value, traceback) -> None:  # type: ignore

     

       90
       +
               assert self.entered

     

       91
       +
               self.entry_count -= 1

     

       92
        
               self.last_called = time.monotonic()