Function create_host_interface

test/vpp_qemu_utils.py:304–412

Create a host interface of type veth. arguments: host_namespace -- host namespace into which the host_interface needs to be set host_ip_prefixes -- a sequence of ip/prefix-lengths to be set on the host_interface vpp_if_name -- name of the veth interface on the

(
    history_file, host_namespace, *host_ip_prefixes, vpp_if_name=None, host_if_name=None
)

def create_host_interface(
    history_file, host_namespace, *host_ip_prefixes, vpp_if_name=None, host_if_name=None
):
    """Create a host interface of type veth.
    arguments:
    host_namespace -- host namespace into which the host_interface needs to be set
    host_ip_prefixes -- a sequence of ip/prefix-lengths to be set
                        on the host_interface
    vpp_if_name -- name of the veth interface on the VPP side
    host_if_name -- name of the veth interface on the host side
    """
    with lock:
        retries = 5

        for attempt in range(retries):
            if_name = (
                host_if_name
                or f"hostif{''.join(random.choices(string.ascii_lowercase + string.digits, k=8))}"
            )
            new_vpp_if_name = (
                vpp_if_name
                or f"vppout{''.join(random.choices(string.ascii_lowercase + string.digits, k=8))}"
            )

            result = subprocess.run(
                [
                    "ip",
                    "link",
                    "add",
                    "name",
                    new_vpp_if_name,
                    "type",
                    "veth",
                    "peer",
                    "name",
                    if_name,
                ],
                capture_output=True,
            )
            if result.returncode == 0:
                host_if_name = if_name
                vpp_if_name = new_vpp_if_name
                with open(history_file, "a") as if_file:
                    if_file.write(f"{host_if_name}\n{vpp_if_name}\n")
                break
            if attempt >= retries - 1:
                raise Exception(
                    f"Failed to create host interface {if_name} and vpp {new_vpp_if_name} after {retries} attempts. Error: {result.stderr.decode()}"
                )

        result = subprocess.run(
            ["ip", "link", "set", host_if_name, "netns", host_namespace],
            capture_output=True,
        )
        if result.returncode != 0:
            raise Exception(
                f"Error setting host interface namespace: {result.stderr.decode()}"
            )