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 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 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 flood_ping(src, dst):
     # we must use a dedicated session becuase the aexpect
     # does not have the other method to interrupt the process in
     # the guest rather than close the session.
     error.context("Flood ping from %s interface %s to %s" % (vm[src].name, ifname[src], vlan_ip[dst]), logging.info)
     session_flood = vm[src].wait_for_login(timeout=60)
     utils_test.ping(vlan_ip[dst], flood=True, interface=ifname[src], session=session_flood, timeout=10)
     session_flood.close()

        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 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 ping_hotplug_nic(ip, mac, session, is_linux_guest):
     status, output = utils_test.ping(ip, 10, timeout=30)
     if status != 0:
         if not is_linux_guest:
             return status, output
         ifname = utils_net.get_linux_ifname(session, mac)
         add_route_cmd = "route add %s dev %s" % (ip, ifname)
         del_route_cmd = "route del %s dev %s" % (ip, ifname)
         logging.warn("Failed to ping %s from host.")
         logging.info("Add route and try again")
         session.cmd_output_safe(add_route_cmd)
         status, output = utils_test.ping(hotnic_ip, 10, timeout=30)
         logging.info("Del the route.")
         status, output = session.cmd_output_safe(del_route_cmd)
     return status, output

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 run(test, params, env):
    """
    General stress test for linux:
       1). Install stress if need
       2). Start stress process
       3). If no stress_time defined, keep stress until test_timeout;
       otherwise execute below steps after sleeping stress_time long
       4). Stop stress process
       5). Uninstall stress
       6). Verify guest kernel crash

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

    stress_duration = int(params.get("stress_duration", "0"))
    # NOTE: stress_duration = 0 ONLY for some legacy test cases using
    # autotest stress.control as their sub test.
    # Please DO define stress_duration to make sure the clean action
    # being performed, if your case can not be controlled by time,
    # use utils_test.VMStress() directly
    stress_type = params.get("stress_type", "stress")
    vms = env.get_all_vms()
    up_time = {}
    error = False
    stress_server = {}

    for vm in vms:
        try:
            up_time[vm.name] = vm.uptime()
            stress_server[vm.name] = utils_test.VMStress(vm, stress_type, params)
            stress_server[vm.name].load_stress_tool()
        except exceptions.TestError as err_msg:
            error = True
            logging.error(err_msg)

    if stress_duration:
        time.sleep(stress_duration)

    for vm in vms:
        try:
            s_ping, o_ping = utils_test.ping(vm.get_address(), count=5, timeout=20)
            if s_ping != 0:
                error = True
                logging.error("%s seem to have gone out of network", vm.name)
                continue
            uptime = vm.uptime()
            if up_time[vm.name] > uptime:
                error = True
                logging.error("%s seem to have rebooted during the stress run", vm.name)
            stress_server[vm.name].unload_stress()
            stress_server[vm.name].clean()
            vm.verify_dmesg()
        except exceptions.TestError as err_msg:
            error = True
            logging.error(err_msg)

    if error:
        test.fail("Run failed: see error messages above")

    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 private_test(session):
     """
     private mode test.
     Check guest cannot ping other guest, but can pin remote host
     """
     ping_s, _ = ping(remote_ip, count=1, timeout=5, session=session)
     if ping_s:
         raise error.TestFail("%s ping %s failed." % (vm1.name, remote_ip))
     ping_s, _ = ping(vm2_ip, count=1, timeout=5, session=session)
     if not ping_s:
         raise error.TestFail("%s ping %s succeed, but expect failed."
                              % (vm1.name, vm2.name))
     try:
         iface_cls.down()
     except error.CmdError, detail:
         raise error.TestNAError(str(detail))

    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 vepa_test(session):
     """
     vepa mode test.
     Check guest can ping remote host
     """
     ping_s, _ = ping(remote_ip, count=1, timeout=5, session=session)
     if ping_s:
         raise error.TestFail("%s ping %s failed." % (vm1.name, remote_ip))

 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 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 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 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 run(test, params, env):
    """
    Test ntpd service, in default setting the execution
    will take longer than 24 hours.

    1.Configure ntpd service in server
    2.Set the date and configure ntpd service in host
    3.Set the date and configure ntpd service in guest
    4.Check ntpd service valid in guest
    5.After long time, test ntpd service still works on guest.
    """

    ntp_test = NTPTest(test, params, env)
    ping_s, _ = utils_test.ping(ntp_test.server_ip, count=1,
                                timeout=5, session=ntp_test.session)
    if ping_s:
        ntp_test.close_session()
        test.cancel("Please make sure the guest can ping server!")

    # Start test from here
    try:
        # Server configuration
        try:
            ntp_test.server_config()
        except (aexpect.ShellError, remote.LoginTimeoutError) as detail:
            test.fail("server config failed. %s" % detail)
        logging.info("waiting for ntp server : %s s" % ntp_test.ntpdate_sleep)
        # Host and Guest will use server's ntpd service to set time.
        # here wait for some seconds for server ntpd service valid
        time.sleep(ntp_test.ntpdate_sleep)

        # Host configuration
        try:
            ntp_test.host_config()
        except (aexpect.ShellError, remote.LoginTimeoutError) as detail:
            test.fail("host config failed.%s" % detail)

        # Guest configuration
        try:
            ntp_test.guest_config()
        except (aexpect.ShellError, remote.LoginTimeoutError) as detail:
            test.fail("guest config failed.%s" % detail)

        try:
            # Wait 20min for ntpq test
            ntp_test.ntpq_test()
        except (aexpect.ShellError, remote.LoginTimeoutError) as detail:
            test.fail("ntpq test failed.%s" % detail)

        try:
            # Wait 24h for  test
            ntp_test.long_time_test()
        except (aexpect.ShellError, remote.LoginTimeoutError) as detail:
            test.fail("long time test failed.%s" % detail)
    finally:
        ntp_test.close_session()

 def bridge_test(session):
     """
     bridge mode test.
     Check guest can ping remote host
     guest can ping other guest when macvtap nic is up
     guest cannot ping remote host when macvtap nic is up
     """
     ping_s, _ = ping(remote_ip, count=1, timeout=5, session=session)
     if ping_s:
         raise error.TestFail("%s ping %s failed."
                              % (vm1.name, remote_ip))
     ping_s, _ = ping(vm2_ip, count=1, timeout=5, session=session)
     if ping_s:
         raise error.TestFail("%s ping %s failed."
                              % (vm1.name, vm2.name))
     try:
         iface_cls.down()
     except error.CmdError, detail:
         raise error.TestNAError(str(detail))