diff --git a/nixos/lib/test-driver/src/test_driver/machine/__init__.py b/nixos/lib/test-driver/src/test_driver/machine/__init__.py index 926b743b17b2..71e14a87eb36 100644 --- a/nixos/lib/test-driver/src/test_driver/machine/__init__.py +++ b/nixos/lib/test-driver/src/test_driver/machine/__init__.py @@ -292,12 +292,18 @@ class Machine: return self.booted and self.connected def log(self, msg: str) -> None: + """ + Log a message to console. + """ self.logger.log(msg, {"machine": self.name}) def log_serial(self, msg: str) -> None: self.logger.log_serial(msg, self.name) def nested(self, msg: str, attrs: dict[str, str] = {}) -> _GeneratorContextManager: + """ + Get nested logger, optionally with extra attributes. + """ my_attrs = {"machine": self.name} my_attrs.update(attrs) return self.logger.nested(msg, my_attrs) @@ -357,6 +363,9 @@ class Machine: retry(check_active, timeout) def get_unit_info(self, unit: str, user: str | None = None) -> dict[str, str]: + """ + Get a dictionary of systemd unit properties, as obtained via `systemctl show`. + """ status, lines = self.systemctl(f'--no-pager show "{unit}"', user) if status != 0: raise RequestedAssertionFailed( @@ -384,6 +393,9 @@ class Machine: property: str, user: str | None = None, ) -> str: + """ + Get the string value of a single systemd unit property + """ status, lines = self.systemctl( f'--no-pager show "{unit}" --property="{property}"', user, @@ -431,6 +443,9 @@ class Machine: return self.execute(f"systemctl {q}") def require_unit_state(self, unit: str, require_state: str = "active") -> None: + """ + Assert that the current state of a unit has a specific value. The default state is "active". + """ with self.nested( f"checking if unit '{unit}' has reached state '{require_state}'" ): @@ -648,6 +663,10 @@ class Machine: return output def wait_for_shutdown(self) -> None: + """ + Wait for the VM to power off. This does *not* initiate a shutdown; + that's usually done via `shutdown()`. + """ if not self.booted: return @@ -687,6 +706,9 @@ class Machine: raise TimeoutError def get_tty_text(self, tty: str) -> str: + """ + Get the output printed to a given TTY. + """ status, output = self.execute( f"fold -w$(stty -F /dev/tty{tty} size | awk '{{print $2}}') /dev/vcs{tty}" ) @@ -785,12 +807,22 @@ class Machine: retry(port_is_closed, timeout) def start_job(self, jobname: str, user: str | None = None) -> tuple[int, str]: + """ + Start systemd service. + """ return self.systemctl(f"start {jobname}", user) def stop_job(self, jobname: str, user: str | None = None) -> tuple[int, str]: + """ + Stop systemd service. + """ return self.systemctl(f"stop {jobname}", user) def connect(self) -> None: + """ + Wait for a connection to the guest root shell + """ + def shell_ready(timeout_secs: int) -> bool: """We sent some data from the backdoor service running on the guest to indicate that the backdoor shell is ready.