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

Code before:
    def scp_get(self, remote_path, local_path, recursive=False):
        """Copy files from the VM using scp."""
        opts = self.options.copy()
        if recursive:
            opts.append("-r")
        self._scp(self.remote_path(remote_path), local_path, opts)

    @retry(
        retry=retry_if_exception_type(ConnectionError),
        wait=wait_fixed(0.15),
        stop=stop_after_attempt(20),
        reraise=True,
    )
    def _init_connection(self):
        """Create an initial SSH client connection (retry until it works).

        Since we're connecting to a microVM we just started, we'll probably
        have to wait for it to boot up and start the SSH server.
        We'll keep trying to execute a remote command that can't fail
        (`/bin/true`), until we get a successful (0) exit code.
        """
        ecode, _, _ = self.run("true")
        if ecode != 0:
            raise ConnectionError

    def run(self, cmd_string, timeout=None, *, check_output=False):
        """Execute the command passed as a string in the ssh context."""
        return self._exec(
            [
                "ssh",
                *self.options,
                f"{self.user}@{self.host}",
                cmd_string,
            ],
            timeout,
            check_output=check_output,
        )

    def check_output(self, cmd_string, timeout=None):
        """Same as `run`, but raises an exception on non-zero return code of remote command"""
        return self.run(cmd_string, timeout, check_output=True)

    def _exec(self, cmd, timeout=None, check_output=False):
        """Private function that handles the ssh client invocation."""
        if self.netns is not None:
            cmd = ["ip", "netns", "exec", self.netns] + cmd
        return utils.run_cmd(cmd, ignore_return_code=not check_output, timeout=timeout)


def mac_from_ip(ip_address):
    """Create a MAC address based on the provided IP.

Review comment (suggestion):
I'd rather try to mimic the `subprocess` module here. So if we want it to raise we call `check_output`, otherwise we call `run`. 

That would involve converting all current uses of run to check_output, so maybe that ship has sailed.

Code after:
    def scp_get(self, remote_path, local_path, recursive=False):
        """Copy files from the VM using scp."""
        opts = self.options.copy()
        if recursive:
            opts.append("-r")
        self._scp(self.remote_path(remote_path), local_path, opts)

    @retry(
        retry=retry_if_exception_type(ChildProcessError),
        wait=wait_fixed(0.15),
        stop=stop_after_attempt(20),
        reraise=True,
    )
    def _init_connection(self):
        """Create an initial SSH client connection (retry until it works).

        Since we're connecting to a microVM we just started, we'll probably
        have to wait for it to boot up and start the SSH server.
        We'll keep trying to execute a remote command that can't fail
        (`/bin/true`), until we get a successful (0) exit code.
        """
        self.check_output("true")

    def run(self, cmd_string, timeout=None, *, check=False):
        """Execute the command passed as a string in the ssh context."""
        return self._exec(
            [
                "ssh",
                *self.options,
                f"{self.user}@{self.host}",
                cmd_string,
            ],
            timeout,
            check=check,
        )

    def check_output(self, cmd_string, timeout=None):
        """Same as `run`, but raises an exception on non-zero return code of remote command"""
        return self.run(cmd_string, timeout, check=True)

    def _exec(self, cmd, timeout=None, check=False):
        """Private function that handles the ssh client invocation."""
        if self.netns is not None:
            cmd = ["ip", "netns", "exec", self.netns] + cmd

        return utils.run_cmd(cmd, check=check, timeout=timeout)


def mac_from_ip(ip_address):
    """Create a MAC address based on the provided IP.