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

Code before:
    def enable_entropy_device(self):
        """Enable entropy device for microVM"""
        self.api.entropy.put()

    def restore_from_path(self, snap_dir: Path, **kwargs):
        """Restore snapshot from a path"""
        return self.restore_from_snapshot(Snapshot.load_from(snap_dir), **kwargs)

    @lru_cache
    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):

Review comment (bug):
How about `raise_if_down(self, timeout=1)` for name?

The timeout is now infinity, that's why some tests time out after 300s (our test limit). By having it centralized here we could lower it so test could fail faster.


Code after:
    def enable_entropy_device(self):
        """Enable entropy device for microVM"""
        self.api.entropy.put()

    def restore_from_path(self, snap_dir: Path, **kwargs):
        """Restore snapshot from a path"""
        return self.restore_from_snapshot(Snapshot.load_from(snap_dir), **kwargs)

    @lru_cache
    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