def ping(test, os_type, match_error, dest, count, session, same_vlan):
    """
    In 'session' ping 'dest'.
    If the two guests are in the same vlan, loss ratio should be 0%.
    Otherwise, loss ratio should be 100%.

    :param test: QEMU test object
    :param dest: dest ip address
    :param count: ping count
    :param session: in which guest to do ping test
    :param same_vlan: whether the two guests are in the same vlan
    """
    if os_type == "linux":
        status, output = utils_test.ping(dest, count,
                                         timeout=int(count) * 1.50,
                                         session=session)
        loss_ratio = utils_test.get_loss_ratio(output)
        ping_result_check(test, loss_ratio, same_vlan)
        logging.debug("%s" % output)
    elif os_type == "windows":  # TODO, not supported by now
        status, output = utils_test.ping(dest, count, timeout=60,
                                         session=session)
        if match_error in str(output):
            ratio = 100
        else:
            loss_ratio = utils_test.get_loss_ratio(output)
        ping_result_check(test, loss_ratio, same_vlan)

    def set_link_test(linkid):
        """
        Issue set_link commands and test its function

        @param linkid: id of netdev or devices to be tested
        """
        ip = vm.get_address(0)
        error.context("Disable guest link by set_link", logging.info)
        vm.set_link(linkid, up=False)
        error.context("Ping guest from host", logging.info)
        s, o = utils_test.ping(ip, count=10, timeout=20)
        if utils_test.get_loss_ratio(o) != 100:
            raise error.TestFail("Still can ping the %s after down %s" %
                                 (ip, linkid))

        error.context("Re-enable guest link by set_link", logging.info)
        vm.set_link(linkid, up=True)
        # Waiting for guest network up again.
        session = vm.wait_for_login(timeout=timeout)
        session.close()
        error.context("Ping guest from host", logging.info)
        s, o = utils_test.ping(ip, count=10, timeout=20)
        # we use 100% here as the notification of link status changed may be
        # delayed in guest driver
        if utils_test.get_loss_ratio(o) == 100:
            raise error.TestFail("Packet loss during ping %s after up %s" %
                                 (ip, linkid))

    def ping(session, nic, dst_ip, strick_check, flood_minutes):
        d_packet_size = [1, 4, 48, 512, 1440, 1500, 1505, 4054, 4055, 4096, 4192, 8878, 9000, 32767, 65507]
        packet_size = params.get("packet_size", "").split() or d_packet_size
        for size in packet_size:
            error.context("Ping with packet size %s" % size, logging.info)
            status, output = utils_test.ping(dst_ip, 10, interface=nic, packetsize=size, timeout=30, session=session)
            if strict_check:
                ratio = utils_test.get_loss_ratio(output)
                if ratio != 0:
                    raise error.TestFail("Loss ratio is %s for packet size" " %s" % (ratio, size))
            else:
                if status != 0:
                    raise error.TestFail("Ping returns non-zero value %s" % output)

        error.context("Flood ping test", logging.info)
        utils_test.ping(
            dst_ip, None, interface=nic, flood=True, output_func=None, timeout=flood_minutes * 60, session=session
        )
        error.context("Final ping test", logging.info)
        counts = params.get("ping_counts", 100)
        status, output = utils_test.ping(dst_ip, counts, interface=nic, timeout=float(counts) * 1.5, session=session)
        if strick_check == "yes":
            ratio = utils_test.get_loss_ratio(output)
            if ratio != 0:
                raise error.TestFail("Packet loss ratio is %s after flood" % ratio)
        else:
            if status != 0:
                raise error.TestFail("Ping returns non-zero value %s" % output)

    def guest_netwok_connecting_check(guest_ip, link_up, change_queues=False):
        """
        Check whether guest network is connective by ping
        """
        if link_up:
            vm.wait_for_login()
            guest_ip = vm.get_address()
        if change_queues:
            env["run_change_queues"] = False
            bg_thread = utils.InterruptedThread(change_queues_number_repeatly,
                                                (guest_ifname,))
            bg_thread.start()

            utils_misc.wait_for(lambda: env["run_change_queues"], 30, 0, 2,
                                "wait queues change start")

        _, output = utils_test.ping(guest_ip, count=10, timeout=20)
        if not link_up and utils_test.get_loss_ratio(output) != 100:
            err_msg = "guest network still connecting after down the link"
            raise error.TestFail(err_msg)
        elif link_up and utils_test.get_loss_ratio(output) == 100:
            err_msg = "All packets lost during ping guest ip after link up"
            raise error.TestFail(err_msg)
        else:
            logging.info("Guest network connecting is exactly as expected")

        if change_queues:
            env["run_change_queues"] = False
            bg_thread.join()

    def guest_netwok_connecting_check(guest_ip, link_up, change_queues=False):
        """
        Check whether guest network is connective by ping
        """
        if change_queues:
            env["run_change_queues"] = False
            bg_thread = utils_misc.InterruptedThread(
                change_queues_number_repeatly, (guest_ifname,))
            bg_thread.start()

            utils_misc.wait_for(lambda: env["run_change_queues"], 30, 0, 2,
                                "wait queues change start")
        time.sleep(0.5)
        output = utils_test.ping(guest_ip, 10, interface=host_interface,
                                 timeout=20, session=None)[1]
        if not link_up and utils_test.get_loss_ratio(output) < 80:
            err_msg = "guest network still connecting after down the link"
            test.fail(err_msg)
        elif link_up and utils_test.get_loss_ratio(output) > 20:
            err_msg = "All packets lost during ping guest ip after link up"
            test.fail(err_msg)

        if change_queues:
            env["run_change_queues"] = False
            bg_thread.join()

def run(test, params, env):
    """
    Boot guest with iommu_platform, then do ping test

    1) Boot a VM with iommu_platform=on
    2) add intel_iommu=on in guest kernel line
    3) reboot guest
    4) do ping test

    :param test: QEMU test object.
    :param params: Dictionary with the test parameters.
    :param env: Dictionary with test environment.
    """

    if utils_misc.get_cpu_vendor(verbose=False) != 'GenuineIntel':
        test.cancel("This case only support Intel platform")

    login_timeout = int(params.get("login_timeout", 360))
    vm = env.get_vm(params["main_vm"])
    vm.verify_alive()
    session = vm.wait_for_login(timeout=login_timeout)
    ping_count = int(params.get("ping_count", 10))
    guest_ip = vm.get_address()

    try:
        status, output = utils_test.ping(guest_ip, ping_count,
                                         timeout=float(ping_count) * 1.5)
        if status != 0:
            test.fail("Ping returns non-zero value %s" % output)
        package_lost = utils_test.get_loss_ratio(output)
        if package_lost != 0:
            test.fail("%s packeage lost when ping guest ip %s " %
                      (package_lost, guest_ip))
    finally:
        session.close()

 def guest_netwok_connecting_check(guest_ip, link_up):
     """
     Check whether guest network is connective by ping
     """
     if link_up:
         vm.wait_for_login()
         guest_ip = vm.get_address()
     s, o = utils_test.ping(guest_ip, count=10, timeout=20)
     if not link_up and utils_test.get_loss_ratio(o) != 100:
         err_msg = "guest network still connecting after down the link"
         raise error.TestFail(err_msg)
     elif link_up and utils_test.get_loss_ratio(o) == 100:
         err_msg = "All packets lost during ping guest ip after link up"
         raise error.TestFail(err_msg)
     else:
         logging.info("Guest network connecting is exactly as expected")

 def large_frame_ping(count=100):
     logging.info("Large frame ping")
     _, output = utils_test.ping(guest_ip, count,
                                 packetsize=max_icmp_pkt_size,
                                 timeout=float(count) * 2)
     ratio = utils_test.get_loss_ratio(output)
     if ratio != 0:
         test.fail("Loss ratio of large frame ping is %s" % ratio)

 def ping_test(dest_ip, ping_time, lost_raito, session=None):
     status, output = utils_test.ping(dest=dest_ip, timeout=ping_time,
                                      session=session)
     packets_lost = utils_test.get_loss_ratio(output)
     if packets_lost > lost_raito:
         err = " %s%% packages lost during ping. " % packets_lost
         err += "Ping command log:\n %s" % "\n".join(output.splitlines()[-3:])
         raise error.TestFail(err)

 def verify_mtu():
     logging.info("Verify the path MTU")
     s, o = utils_test.ping(ip, 10, interface=ifname, packetsize=max_icmp_pkt_size, hint="do", timeout=15)
     if s != 0:
         logging.error(o)
         raise error.TestFail("Path MTU is not as expected")
     if utils_test.get_loss_ratio(o) != 0:
         logging.error(o)
         raise error.TestFail("Packet loss ratio during MTU " "verification is not zero")

def launch_client(sessions, servers, server_ctl, clients,
                  l, nf_args, port, params):
    """
    Launch netperf clients
    """
    # Start netserver
    error.context("Start Netserver on guest", logging.info)
    remote_dir = params.get("remote_dir", "/var/tmp")
    client_path = os.path.join(remote_dir, "netperf-2.6.0/src/netperf")
    server_path = os.path.join(remote_dir, "netperf-2.6.0/src/netserver")

    if params.get("os_type") == "windows":
        winutils_vol = utils_misc.get_winutils_vol(server_ctl)
        client_path = "%s:\\netperf" % winutils_vol
        netserv_start_cmd = params.get("netserv_start_cmd") % winutils_vol

        logging.info("Netserver start cmd is '%s'" % netserv_start_cmd)
        if "NETSERVER.EXE" not in server_ctl.cmd_output("tasklist"):
            server_ctl.cmd_output(netserv_start_cmd)
            o_tasklist = server_ctl.cmd_output("tasklist")
            if "NETSERVER.EXE" not in o_tasklist.upper():
                msg = "Can not start netserver in Windows guest"
                raise error.TestError(msg)

    else:
        logging.info("Netserver start cmd is '%s'" % server_path)
        ssh_cmd(server_ctl, "pidof netserver || %s" % server_path)
    logging.info("Netserver start successfully")

    # start netperf
    error.context("Start netperf client threads", logging.info)
    client_threads = []

    for client in clients:
        test_timeout = len(clients) * l
        server = servers[clients.index(client) % len(servers)]
        netperf_cmd = "%s -H %s -l %s %s" % (client_path, server,
                                             int(l), nf_args)
        client_threads.append([ssh_cmd, (client, netperf_cmd, test_timeout)])

    result_info = utils_misc.parallel(client_threads)

    counts = 5
    for server in servers:
        if not re.findall("TEST.*to %s" % server, str(result_info)):
            raise error.TestError("Nerperf stress on nic %s failed" % server)
        logging.info("Network stress on %s successfully" % server)

        status, output = utils_test.ping(server, counts,
                                         timeout=float(counts) * 1.5)
        if status != 0:
            raise error.TestFail("Ping returns non-zero value %s" % output)

        package_lost = utils_test.get_loss_ratio(output)
        if package_lost != 0:
            raise error.TestFail("%s packeage lost when ping server ip %s " %
                                 (package_lost, server))

 def large_frame_ping(count=100):
     logging.info("Large frame ping")
     _, o = utils_test.ping(ip, count, interface=ifname,
                                packetsize=max_icmp_pkt_size,
                                timeout=float(count) * 2)
     ratio = utils_test.get_loss_ratio(o)
     if ratio != 0:
         raise error.TestFail("Loss ratio of large frame ping is %s" %
                              ratio)

    def set_link_test(linkid):
        """
        Issue set_link commands and test its function

        @param linkid: id of netdev or devices to be tested
        """
        ip = vm.get_address(0)

        vm.set_link(linkid, up=False)
        _, o = utils_test.ping(ip, count=10, timeout=20)
        if utils_test.get_loss_ratio(o) != 100:
            raise error.TestFail("Still can ping the %s after down %s" %
                                 (ip, linkid))

        vm.set_link(linkid, up=True)
        _, o = utils_test.ping(ip, count=10, timeout=20)
        # we use 100% here as the notification of link status changed may be
        # delayed in guest driver
        if utils_test.get_loss_ratio(o) == 100:
            raise error.TestFail("Packet loss during ping %s after up %s" %
                                 (ip, linkid))

        def size_increase_ping(step=random.randrange(90, 110)):
            logging.info("Size increase ping")
            for size in range(0, max_icmp_pkt_size + 1, step):
                logging.info("Ping %s with size %s", ip, size)
                s, o = utils_test.ping(ip, 1, interface=ifname, packetsize=size, hint="do", timeout=1)
                if s != 0:
                    s, o = utils_test.ping(
                        ip, 10, interface=ifname, packetsize=size, adaptive=True, hint="do", timeout=20
                    )

                    if utils_test.get_loss_ratio(o) > int(params.get("fail_ratio", 50)):
                        raise error.TestFail("Ping loss ratio is greater " "than 50% for size %s" % size)

 def verify_mtu():
     logging.info("Verify the path MTU")
     status, output = utils_test.ping(guest_ip, 10,
                                      packetsize=max_icmp_pkt_size,
                                      hint="do", timeout=15)
     if status != 0:
         logging.error(output)
         test.fail("Path MTU is not as expected")
     if utils_test.get_loss_ratio(output) != 0:
         logging.error(output)
         test.fail("Packet loss ratio during MTU "
                   "verification is not zero")

        def size_increase_ping(step=random.randrange(90, 110)):
            logging.info("Size increase ping")
            for size in range(0, max_icmp_pkt_size + 1, step):
                logging.info("Ping %s with size %s", guest_ip, size)
                status, output = utils_test.ping(guest_ip, 1, packetsize=size, hint="do", timeout=1)
                if status != 0:
                    status, output = utils_test.ping(
                        guest_ip, 10, packetsize=size, adaptive=True, hint="do", timeout=20
                    )

                    fail_ratio = int(params.get("fail_ratio", 50))
                    if utils_test.get_loss_ratio(output) > fail_ratio:
                        raise error.TestFail("Ping loss ratio is greater " "than 50% for size %s" % size)

 def check_ping(status, output):
     """ ratio <5% is acceptance."""
     if status != 0:
         test.fail("Ping failed, staus:%s, output:%s" % (status, output))
     # if status != 0 the ping process seams hit issue.
     ratio = utils_test.get_loss_ratio(output)
     if ratio == -1:
         test.fail("The ratio is %s, and status is %s, "
                   "output is %s" % (ratio, status, output))
     elif ratio > int(params["failed_ratio"]):
         test.fail("The loss raito is %s, test failed" % ratio)
     logging.info("ping pass with loss raito:%s, that less than %s" %
                  (ratio, params["failed_ratio"]))

def _ping_with_params(test, params, dest, interface=None,
                      packet_size=None, interval=None,
                      count=0, session=None, flood=False):
    if flood:
        cmd = "ping " + dest + " -f -q"
        if interface:
            cmd += " -S %s" % interface
        flood_minutes = float(params.get("flood_minutes", 10))
        status, output = utils_net.raw_ping(cmd, flood_minutes * 60,
                                            session, logging.debug)
    else:
        timeout = float(count) * 1.5
        status, output = utils_net.ping(dest, count, interval, interface,
                                        packet_size, session=session,
                                        timeout=timeout)
    if status != 0:
        test.fail("Ping failed, status: %s,"
                  " output: %s" % (status, output))
    if params.get("strict_check", "no") == "yes":
        ratio = utils_test.get_loss_ratio(output)
        if ratio != 0:
            test.fail("Loss ratio is %s" % ratio)

    def ping_test(session, dst, drop_flow=False):
        """
        Ping test, check icmp
        """
        ping_status, ping_output = utils_test.ping(dest=dst, count=10, timeout=20, session=session)
        # when drop_flow is true, ping should failed(return not zero)
        # drop_flow is false, ping should success
        packets_lost = 100
        if ping_status and not drop_flow:
            raise error.TestError("Ping should success when not drop_icmp")
        elif not ping_status:
            packets_lost = utils_test.get_loss_ratio(ping_output)
            if drop_flow and packets_lost != 100:
                raise error.TestError("When drop_icmp, ping shouldn't works")
            if not drop_flow and packets_lost == 100:
                raise error.TestError("When not drop_icmp, ping should works")

        info_msg = "Correct, icmp flow %s dropped, ping '%s', "
        info_msg += "packets lost rate is: '%s'"
        logging.info(
            info_msg % ((drop_flow and "was" or "was not"), (ping_status and "failed" or "success"), packets_lost)
        )

def run(test, params, env):
    """
    Expose host MTU to guest test

    1) Boot up guest with param 'host_mtu=4000' in nic part
    2) Disable NetworkManager in guest
    3) set mtu of guest tap (eg: tap0) and physical nic (eg: eno1) to
       4000 in host
    4) check the mtu in guest
    5) ping from guest to external host with packet size 3972

    :param test: kvm test object
    :param params: Dictionary with the test parameters
    :param env: Dictionary with test environment
    """

    def get_ovs_ports(ovs):
        """
        Get ovs ports

        :param ovs: ovs bridge name
        """

        cmd = "ovs-vsctl list-ports %s" % ovs
        return process.system_output(cmd, shell=True).decode()

    def is_ovs_backend(netdst):
        """
        Check whether the host is OVS backend

        :param netdst: netdst get from command line
        """

        return netdst in process.system_output("ovs-vsctl list-br",
                                               ignore_status=True,
                                               shell=True).decode()

    def cleanup_ovs_ports(netdst, ports):
        """
        Clean up created ovs ports in this case

        :param netdst: netdst get from command line
        :param ports: existing ports need to be remain before this test
        """

        if is_ovs_backend(netdst) is True:
            ports = set(get_ovs_ports(netdst).splitlines()) - \
                    set(ports.splitlines())
            for p in ports:
                process.system("ovs-vsctl del-port %s %s" % (netdst, p))

    netdst = params.get("netdst", "switch")
    if netdst in utils_net.Bridge().list_br():
        host_hw_interface = utils_net.Bridge().list_iface()[0]
    elif is_ovs_backend(netdst) is True:
        host_hw_interface = get_ovs_ports(netdst)
        tmp_ports = re.findall(r"t[0-9]-[a-zA-Z0-9]{6}", host_hw_interface)
        if tmp_ports:
            for p in tmp_ports:
                process.system_output("ovs-vsctl del-port %s %s" %
                                      (netdst, p))
            host_hw_interface = get_ovs_ports(netdst)
    else:
        test.cancel("The host is using Macvtap backend, which is not"
                    " supported by now!")

    params["start_vm"] = "yes"
    env_process.preprocess_vm(test, params, env, params["main_vm"])

    vm = env.get_vm(params["main_vm"])
    vm.verify_alive()

    vm_iface = vm.get_ifname()
    # TODO, will support windows later
    process.system_output(params["set_mtu_cmd"] % host_hw_interface)
    process.system_output(params["set_mtu_cmd"] % vm_iface)

    os_type = params.get("os_type", "linux")
    login_timeout = float(params.get("login_timeout", 360))
    session = vm.wait_for_login(timeout=login_timeout)

    host_ip = utils_net.get_ip_address_by_interface(params["netdst"])
    if os_type == "linux":  # TODO, will support windows later
        session.cmd_output_safe(params["nm_stop_cmd"])
        guest_ifname = utils_net.get_linux_ifname(session,
                                                  vm.get_mac_address())
        output = session.cmd_output_safe(
                params["check_guest_mtu_cmd"] % guest_ifname)
        error_context.context(output, logging.info)
        match_string = "mtu %s" % params["mtu_value"]
        if match_string in output:
            logging.info("Host mtu %s exposed to guest as expected!" %
                         params["mtu_value"])
            logging.info("Ping from guest to host with packet size 3972")
            status, output = utils_test.ping(host_ip, 10, packetsize=3972,
                                             timeout=30, session=session)
            ratio = utils_test.get_loss_ratio(output)
            if ratio != 0:
                test.fail("Loss ratio is %s", ratio)
        else:
#.........这里部分代码省略.........