gador/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nixos/test-driver: add optional address arg to wait_for_{open,closed}_port

Browse Source 
This is useful for testing servers configured to listen on specific
addresses.
This commit is contained in:
Bjørn Forsman
2022-12-29 13:33:25 +01:00
parent 904f07f595
commit dba49a43a0
3 changed files with 16 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
nixos/lib/test-driver/test_driver/machine.py
View File

@@ -699,20 +699,22 @@ class Machine:
with self.nested("waiting for file ‘{}‘".format(filename)):
retry(check_file)
def wait_for_open_port(self, port: int) -> None:
def wait_for_open_port(self, port: int, addr: str = "localhost") -> None:
def port_is_open(_: Any) -> bool:
status, _ = self.execute("nc -z localhost {}".format(port))
status, _ = self.execute("nc -z {} {}".format(addr, port))
return status == 0
with self.nested("waiting for TCP port {}".format(port)):
with self.nested("waiting for TCP port {} on {}".format(port, addr)):
retry(port_is_open)
def wait_for_closed_port(self, port: int) -> None:
def wait_for_closed_port(self, port: int, addr: str = "localhost") -> None:
def port_is_closed(_: Any) -> bool:
status, _ = self.execute("nc -z localhost {}".format(port))
status, _ = self.execute("nc -z {} {}".format(addr, port))
return status != 0
with self.nested("waiting for TCP port {} to be closed".format(port)):
with self.nested(
"waiting for TCP port {} on {} to be closed".format(port, addr)
):
retry(port_is_closed)
def start_job(self, jobname: str, user: Optional[str] = None) -> Tuple[int, str]: