pyrox.dev
Merge pull request #170636 from Synthetica9/wait_before_entry
nixos/test-driver: add wait_before_entry
authored by Jacek Galowicz and committed by GitHub 9938dec6 58236ec8
+14
nixos/lib/test-driver/test_driver/driver.py
+14
nixos/lib/test-driver/test_driver/driver.py
···
220
220
res = driver.polling_conditions.pop()
221
221
assert res is self.condition
222
222
223
+
def wait(self, timeout: int = 900) -> None:
224
+
def condition(last: bool) -> bool:
225
+
if last:
226
+
rootlog.info(f"Last chance for {self.condition.description}")
227
+
ret = self.condition.check(force=True)
228
+
if not ret and not last:
229
+
rootlog.info(
230
+
f"({self.condition.description} failure not fatal yet)"
231
+
)
232
+
return ret
233
+
234
+
with rootlog.nested(f"waiting for {self.condition.description}"):
235
+
retry(condition, timeout=timeout)
236
+
223
237
if fun_ is None:
224
238
return Poll
225
239
else:
+22
-7
nixos/lib/test-driver/test_driver/polling_condition.py
+22
-7
nixos/lib/test-driver/test_driver/polling_condition.py
···
1
1
from typing import Callable, Optional
2
+
from math import isfinite
2
3
import time
3
4
4
5
from .logger import rootlog
···
14
15
description: Optional[str]
15
16
16
17
last_called: float
17
-
entered: bool
18
+
entry_count: int
18
19
19
20
def __init__(
20
21
self,
···
34
35
self.description = str(description)
35
36
36
37
self.last_called = float("-inf")
37
-
self.entered = False
38
+
self.entry_count = 0
38
39
39
-
def check(self) -> bool:
40
-
if self.entered or not self.overdue:
40
+
def check(self, force: bool = False) -> bool:
41
+
if (self.entered or not self.overdue) and not force:
41
42
return True
42
43
43
44
with self, rootlog.nested(self.nested_message):
44
-
rootlog.info(f"Time since last: {time.monotonic() - self.last_called:.2f}s")
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)
45
53
try:
46
54
res = self.condition() # type: ignore
47
55
except Exception:
···
69
77
def overdue(self) -> bool:
70
78
return self.last_called + self.seconds_interval < time.monotonic()
71
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
+
72
86
def __enter__(self) -> None:
73
-
self.entered = True
87
+
self.entry_count += 1
74
88
75
89
def __exit__(self, exc_type, exc_value, traceback) -> None: # type: ignore
76
-
self.entered = False
90
+
assert self.entered
91
+
self.entry_count -= 1
77
92
self.last_called = time.monotonic()
+1
-1
nixos/tests/vscodium.nix
+1
-1
nixos/tests/vscodium.nix