Repository: firecracker-microvm/firecracker
File: tests/framework/microvm.py (Python)

Code before:
    def ssh_iface(self, iface_idx=0):
        """Return a cached SSH connection on a given interface id."""
        guest_ip = list(self.iface.values())[iface_idx]["iface"].guest_ip
        self.ssh_key = Path(self.ssh_key)
        return net_tools.SSHConnection(
            netns=self.netns.id,
            ssh_key=self.ssh_key,
            user="root",
            host=guest_ip,
        )

    @property
    def ssh(self):
        """Return a cached SSH connection on the 1st interface"""
        return self.ssh_iface(0)

    def verify_if_guest_is_up(self, up=True):
        """Verify if the guest running inside the microVM is up and running.

        It is checked by running a command via SSH connection. This method can also be used to wait
        for guest to boot up.

        :param up: expected to be up (True) or down (False).
        """
        rc, _, stderr = self.ssh.run("true")
        if up:
            assert rc == 0, stderr
        else:
            # SSH connection should timeout and the return code should not be 0.
            assert rc != 0


class MicroVMFactory:
    """MicroVM factory"""

    def __init__(self, fc_binary_path: Path, jailer_binary_path: Path, **kwargs):
        self.vms = []
        self.fc_binary_path = Path(fc_binary_path)
        self.jailer_binary_path = Path(jailer_binary_path)
        self.kwargs = kwargs

    def build(self, kernel=None, rootfs=None, **kwargs):
        """Build a microvm"""
        kwargs = self.kwargs | kwargs
        microvm_id = kwargs.pop("microvm_id", str(uuid.uuid4()))
        vm = Microvm(
            microvm_id=microvm_id,
            fc_binary_path=kwargs.pop("fc_binary_path", self.fc_binary_path),
            jailer_binary_path=kwargs.pop(
                "jailer_binary_path", self.jailer_binary_path
            ),

Review comment (suggestion):
hmm since only one test needs this, we could just have that test do:

```python
with pytest.raises(AssertionError):
    uv.verify_if_guest_is_up()
```

Code after:
    def ssh_iface(self, iface_idx=0):
        """Return a cached SSH connection on a given interface id."""
        guest_ip = list(self.iface.values())[iface_idx]["iface"].guest_ip
        self.ssh_key = Path(self.ssh_key)
        return net_tools.SSHConnection(
            netns=self.netns.id,
            ssh_key=self.ssh_key,
            user="root",
            host=guest_ip,
        )

    @property
    def ssh(self):
        """Return a cached SSH connection on the 1st interface"""
        return self.ssh_iface(0)

    @property
    def thread_backtraces(self):
        """Return backtraces of all threads"""
        backtraces = []
        for thread_name, thread_pids in utils.get_threads(self.firecracker_pid).items():
            for pid in thread_pids:
                backtraces.append(
                    f"{thread_name} ({pid=}):\n"
                    f"{utils.run_cmd(f'cat /proc/{pid}/stack').stdout}"
                )
        return "\n".join(backtraces)

    def wait_for_up(self, timeout=10):
        """Wait for guest running inside the microVM to come up and respond.

        :param timeout: seconds to wait.
        """
        try:
            rc, stdout, stderr = self.ssh.run("true", timeout)
        except subprocess.TimeoutExpired:
            print(
                f"Remote command did not respond within {timeout}s\n\n"
                f"Firecracker logs:\n{self.log_data}\n"
                f"Thread backtraces:\n{self.thread_backtraces}"
            )
            raise
        assert rc == 0, (
            f"Remote command exited with non-0 status code\n\n"
            f"{rc=}\n{stdout=}\n{stderr=}\n\n"
            f"Firecracker logs:\n{self.log_data}\n"
            f"Thread backtraces:\n{self.thread_backtraces}"
        )


class MicroVMFactory: