b12f/nixpkgs
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

nixos/test-driver: add (functional) timeouts to more functions

Browse source 
A retry timeout doesn't really help if the thing it's retrying may block
forever.
This commit is contained in:
Cole Helbling 2021-11-15 12:23:25 -08:00
parent 363d7f3ae8
commit e62b8020f3
1 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
nixos/lib/test-driver/test-driver.py
View file

@ -634,12 +634,12 @@ class Machine:
pass_fds=[self.shell.fileno()],
)
def succeed(self, *commands: str) -> str:
def succeed(self, *commands: str, timeout: Optional[int] = None) -> str:
"""Execute each command and check that it succeeds."""
output = ""
for command in commands:
with self.nested("must succeed: {}".format(command)):
(status, out) = self.execute(command)
(status, out) = self.execute(command, timeout=timeout)
if status != 0:
self.log("output: {}".format(out))
raise Exception(
@ -648,12 +648,12 @@ class Machine:
output += out
return output
def fail(self, *commands: str) -> str:
def fail(self, *commands: str, timeout: Optional[int] = None) -> str:
"""Execute each command and check that it fails."""
output = ""
for command in commands:
with self.nested("must fail: {}".format(command)):
(status, out) = self.execute(command)
(status, out) = self.execute(command, timeout=timeout)
if status == 0:
raise Exception(
"command `{}` unexpectedly succeeded".format(command)
@ -669,14 +669,14 @@ class Machine:
def check_success(_: Any) -> bool:
nonlocal output
status, output = self.execute(command)
status, output = self.execute(command, timeout=timeout)
return status == 0
with self.nested("waiting for success: {}".format(command)):
retry(check_success, timeout)
return output
def wait_until_fails(self, command: str) -> str:
def wait_until_fails(self, command: str, timeout: int = 900) -> str:
"""Wait until a command returns failure.
Throws an exception on timeout.
"""
@ -684,7 +684,7 @@ class Machine:
def check_failure(_: Any) -> bool:
nonlocal output
status, output = self.execute(command)
status, output = self.execute(command, timeout=timeout)
return status != 0
with self.nested("waiting for failure: {}".format(command)):