ilyakooo0/nixpkgs
1
0
Fork 0
mirror of https://github.com/ilyakooo0/nixpkgs.git synced 2024-09-23 21:57:52 +03:00
Code Issues Projects Releases Wiki Activity

Merge pull request #208225 from bjornfor/nixos-test-driver-improvement

Browse Source 
nixos/test-driver: add optional address arg to wait_for_{open,closed}_port
This commit is contained in:
Jacek Galowicz 2022-12-29 18:31:11 +01:00 committed by GitHub
commit d4513957f6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 16 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
nixos/doc/manual/development/writing-nixos-tests.section.md
View File

@ -273,12 +273,13 @@ The following methods are available on machine objects:
`wait_for_open_port`
: Wait until a process is listening on the given TCP port (on
`localhost`, at least).
: Wait until a process is listening on the given TCP port and IP address
(default `localhost`).
`wait_for_closed_port`
: Wait until nobody is listening on the given TCP port.
: Wait until nobody is listening on the given TCP port and IP address
(default `localhost`).
`wait_for_x`

7
nixos/doc/manual/from_md/development/writing-nixos-tests.section.xml
View File

@ -483,8 +483,8 @@ start_all()
</term>
<listitem>
<para>
Wait until a process is listening on the given TCP port (on
<literal>localhost</literal>, at least).
Wait until a process is listening on the given TCP port and
IP address (default <literal>localhost</literal>).
</para>
</listitem>
</varlistentry>
@ -494,7 +494,8 @@ start_all()
</term>
<listitem>
<para>
Wait until nobody is listening on the given TCP port.
Wait until nobody is listening on the given TCP port and IP
address (default <literal>localhost</literal>).
</para>
</listitem>
</varlistentry>

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]: