本文整理汇总了Python中virttest.data_dir.get_deps_dir函数的典型用法代码示例。如果您正苦于以下问题:Python get_deps_dir函数的具体用法?Python get_deps_dir怎么用?Python get_deps_dir使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_deps_dir函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: compile_nc_vsock
def compile_nc_vsock(test, vm, session):
"""
Copy and compile nc-vsock on both host and guest
:param test: QEMU test object
:param vm: Object qemu_vm.VM
:param session: vm session
:return: Path to binary nc-vsock or None if compile failed
"""
nc_vsock_dir = '/home/'
nc_vsock_bin = 'nc-vsock'
nc_vsock_c = 'nc-vsock.c'
src_file = os.path.join(data_dir.get_deps_dir("nc_vsock"), nc_vsock_c)
bin_path = os.path.join(nc_vsock_dir, nc_vsock_bin)
rm_cmd = 'rm -rf %s*' % bin_path
session.cmd(rm_cmd)
process.system(rm_cmd, shell=True, ignore_status=True)
cmd_cp = "cp %s %s" % (src_file, nc_vsock_dir)
process.system(cmd_cp)
vm.copy_files_to(src_file, nc_vsock_dir)
compile_cmd = "cd %s && gcc -o %s %s" % (
nc_vsock_dir, nc_vsock_bin, nc_vsock_c)
host_status = process.system(compile_cmd, shell=True)
guest_status = session.cmd_status(compile_cmd)
if (host_status or guest_status) != 0:
process.system(rm_cmd, shell=True, ignore_status=True)
session.cmd_output_safe(rm_cmd)
session.close()
test.error("Compile nc-vsock failed")
return bin_path
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:30,代码来源:vsock_test.py
示例2: get_guest_cpuid
def get_guest_cpuid(self, cpu_model, feature=None, extra_params=None):
test_kernel_dir = os.path.join(data_dir.get_deps_dir(), "cpuid", "src")
os.chdir(test_kernel_dir)
utils.make("cpuid_dump_kernel.bin")
vm_name = params['main_vm']
params_b = params.copy()
params_b["kernel"] = os.path.join(
test_kernel_dir, "cpuid_dump_kernel.bin")
params_b["cpu_model"] = cpu_model
params_b["cpu_model_flags"] = feature
del params_b["images"]
del params_b["nics"]
if extra_params:
params_b.update(extra_params)
env_process.preprocess_vm(self, params_b, env, vm_name)
vm = env.get_vm(vm_name)
dbg('is dead: %r', vm.is_dead())
vm.create()
self.vm = vm
vm.resume()
timeout = float(params.get("login_timeout", 240))
f = lambda: re.search("==END TEST==", vm.serial_console.get_output())
if not utils_misc.wait_for(f, timeout, 1):
raise error.TestFail("Could not get test complete message.")
test_output = parse_cpuid_dump(vm.serial_console.get_output())
if test_output is None:
raise error.TestFail("Test output signature not found in "
"output:\n %s", vm.serial_console.get_output())
vm.destroy(gracefully=False)
return test_output
开发者ID:QiuMike,项目名称:tp-qemu,代码行数:33,代码来源:cpuid.py
示例3: run
def run(test, params, env):
"""
Verify if guests using kvm-clock as the time source have a sane clock
resolution.
:param test: kvm test object.
:param params: Dictionary with test parameters.
:param env: Dictionary with the test environment.
"""
source_name = "clock_getres/clock_getres.c"
source_name = os.path.join(data_dir.get_deps_dir(), source_name)
dest_name = "/tmp/clock_getres.c"
bin_name = "/tmp/clock_getres"
if not os.path.isfile(source_name):
raise error.TestError("Could not find %s" % source_name)
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
timeout = int(params.get("login_timeout", 360))
session = vm.wait_for_login(timeout=timeout)
vm.copy_files_to(source_name, dest_name)
session.cmd("gcc -lrt -o %s %s" % (bin_name, dest_name))
session.cmd(bin_name)
logging.info("PASS: Guest reported appropriate clock resolution")
sub_test = params.get("sub_test")
if sub_test:
error.context("Run sub test '%s' after checking"
" clock resolution" % sub_test, logging.info)
utils_test.run_virt_sub_test(test, params, env, sub_test)
开发者ID:QiuMike,项目名称:tp-qemu,代码行数:31,代码来源:clock_getres.py
示例4: watchdog_test_suit
def watchdog_test_suit():
"""
Run watchdog-test-framework to verify the function of emulated watchdog
devices.
Test steps of the framework are as follows:
1) Set up the watchdog with a 30 second timeout.
2) Ping the watchdog for 60 seconds. During this time the guest should
run normally.
3) Stop pinging the watchdog and just count up. If the virtual watchdog
device is set correctly, then the watchdog action (eg. pause) should
happen around the 30 second mark.
"""
_watchdog_device_check(test, session, watchdog_device_type)
watchdog_test_lib = params["watchdog_test_lib"]
src_path = os.path.join(data_dir.get_deps_dir(), watchdog_test_lib)
test_dir = os.path.basename(watchdog_test_lib)
session.cmd_output("rm -rf /home/%s" % test_dir)
vm.copy_files_to(src_path, "/home")
session.cmd_output("cd /home/%s && make" % test_dir)
try:
session.cmd_output("./watchdog-test --yes &", timeout=130)
except ShellTimeoutError:
# To judge if watchdog action happens after 30s
o = session.get_output().splitlines()[-1]
if 27 <= int(o.rstrip("...")) <= 32:
_action_check(test, session, watchdog_action)
else:
test.fail("Watchdog action doesn't happen after 30s.")
else:
test.fail("Watchdog test suit doesn't run successfully.")
finally:
vm.resume()
session.cmd_output("pkill watchdog-test")
session.cmd_output("rm -rf /home/%s" % test_dir)
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:35,代码来源:watchdog.py
示例5: setup_service
def setup_service(setup_target):
setup_timeout = int(params.get("setup_timeout", 360))
if setup_target == "localhost":
setup_func = _system_output
os_type = "linux"
else:
setup_vm = env.get_vm(setup_target)
setup_session = setup_vm.wait_for_login(timeout=timeout)
setup_func = setup_session.cmd
os_type = params["os_type"]
setup_params = params.object_params(os_type)
setup_cmd = setup_params.get("setup_cmd", "service SERVICE restart")
prepare_cmd = setup_params.get("prepare_cmd")
setup_cmd = re.sub("SERVICE", setup_params.get("service", ""),
setup_cmd)
error_context.context("Set up %s service in %s"
% (setup_params.get("service"), setup_target),
logging.info)
setup_func(setup_cmd, timeout=setup_timeout)
if params.get("copy_ftp_site") and setup_target != "localhost":
ftp_site = os.path.join(data_dir.get_deps_dir(), params.get("copy_ftp_site"))
ftp_dir = params.get("ftp_dir")
setup_vm.copy_files_to(ftp_site, ftp_dir)
if prepare_cmd:
setup_func(prepare_cmd, timeout=setup_timeout)
if setup_target != "localhost":
setup_session.close()
开发者ID:ngu-niny,项目名称:tp-qemu,代码行数:30,代码来源:openflow_acl_test.py
示例6: get_guest_cpuid
def get_guest_cpuid(self, cpu_model, feature=None, extra_params=None, qom_mode=False):
if not qom_mode:
test_kernel_dir = os.path.join(data_dir.get_deps_dir(), "cpuid", "src")
os.chdir(test_kernel_dir)
utils.make("cpuid_dump_kernel.bin")
vm_name = params['main_vm']
params_b = params.copy()
if not qom_mode:
params_b["kernel"] = os.path.join(
test_kernel_dir, "cpuid_dump_kernel.bin")
params_b["cpu_model"] = cpu_model
params_b["cpu_model_flags"] = feature
del params_b["images"]
del params_b["nics"]
if extra_params:
params_b.update(extra_params)
env_process.preprocess_vm(self, params_b, env, vm_name)
vm = env.get_vm(vm_name)
dbg('is dead: %r', vm.is_dead())
vm.create()
self.vm = vm
if qom_mode:
return get_qom_cpuid(self, vm)
else:
return get_test_kernel_cpuid(self, vm)
开发者ID:Zhengtong,项目名称:tp-qemu,代码行数:26,代码来源:cpuid.py
示例7: test
def test(self, test, params, env):
iperf_src_path = os.path.join(data_dir.get_deps_dir(), "iperf")
self.iperf_b_path = os.path.join("iperf-2.0.4", "src", "iperf")
error.context("Install iperf to vms machine.")
utils_misc.ForAllP(
self.machines).compile_autotools_app_tar(iperf_src_path,
"iperf-2.0.4.tar.gz")
allow_iperf_firewall(self.host)
utils_misc.ForAllP(self.mvms).cmd("iptables -F")
self.start_servers()
# Test TCP bandwidth
error.context("Test iperf bandwidth tcp.")
speeds = self.test_bandwidth()
logging.info("TCP Bandwidth from vm->host: %s", speeds[0])
logging.info("TCP Bandwidth from host->vm: %s", speeds[1])
logging.info("TCP Bandwidth from vm->vm: %s", speeds[2])
# test udp bandwidth limited to 1Gb
error.context("Test iperf bandwidth udp.")
speeds = self.test_bandwidth("-u -b 1G")
logging.info("UDP Bandwidth from vm->host: %s", speeds[0])
logging.info("UDP Bandwidth from host->vm: %s", speeds[1])
logging.info("UDP Bandwidth from vm->vm: %s", speeds[2])
开发者ID:CongLi,项目名称:tp-qemu,代码行数:27,代码来源:ovs_basic.py
示例8: run
def run(test, params, env):
"""
Run Iometer for Windows on a Windows guest:
1) Boot guest with additional disk
2) Format the additional disk
3) Install and register Iometer
4) Perpare icf to Iometer.exe
5) Run Iometer.exe with icf
6) Copy result to host
:param test: kvm test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment.
"""
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
timeout = int(params.get("login_timeout", 360))
session = vm.wait_for_login(timeout=timeout)
# format the target disk
utils_test.run_virt_sub_test(test, params, env, "format_disk")
error_context.context("Install Iometer", logging.info)
cmd_timeout = int(params.get("cmd_timeout", 360))
ins_cmd = params["install_cmd"]
vol_utils = utils_misc.get_winutils_vol(session)
if not vol_utils:
raise exceptions.TestError("WIN_UTILS CDROM not found")
ins_cmd = re.sub("WIN_UTILS", vol_utils, ins_cmd)
session.cmd(cmd=ins_cmd, timeout=cmd_timeout)
time.sleep(0.5)
error_context.context("Register Iometer", logging.info)
reg_cmd = params["register_cmd"]
reg_cmd = re.sub("WIN_UTILS", vol_utils, reg_cmd)
session.cmd_output(cmd=reg_cmd, timeout=cmd_timeout)
error_context.context("Prepare icf for Iometer", logging.info)
icf_name = params["icf_name"]
ins_path = params["install_path"]
res_file = params["result_file"]
icf_file = os.path.join(data_dir.get_deps_dir(), "iometer", icf_name)
vm.copy_files_to(icf_file, "%s\\%s" % (ins_path, icf_name))
# Run Iometer
error_context.context("Start Iometer", logging.info)
session.cmd("cd %s" % ins_path)
logging.info("Change dir to: %s" % ins_path)
run_cmd = params["run_cmd"]
run_timeout = int(params.get("run_timeout", 1000))
logging.info("Set Timeout: %ss" % run_timeout)
run_cmd = run_cmd % (icf_name, res_file)
logging.info("Execute Command: %s" % run_cmd)
s, o = session.cmd_status_output(cmd=run_cmd, timeout=run_timeout)
error_context.context("Copy result '%s' to host" % res_file, logging.info)
vm.copy_files_from(res_file, test.resultsdir)
if s != 0:
raise exceptions.TestFail("iometer test failed. {}".format(o))
开发者ID:EIChaoYang,项目名称:tp-qemu,代码行数:60,代码来源:iometer_windows.py
示例9: run
def run(test, params, env):
"""
try to exploit the guest to test whether nx(cpu) bit takes effect.
1) boot the guest
2) cp the exploit prog into the guest
3) run the exploit
:param test: QEMU test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment.
"""
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
session = vm.wait_for_login(timeout=int(params.get("login_timeout", 360)))
exploit_file = os.path.join(data_dir.get_deps_dir(), "nx", "x64_sc_rdo.c")
dst_dir = "/tmp"
error.context("Copy the Exploit file to guest.", logging.info)
vm.copy_files_to(exploit_file, dst_dir)
error.context("Build exploit program in guest.", logging.info)
build_exploit = "gcc -o /tmp/nx_exploit /tmp/x64_sc_rdo.c"
if session.cmd_status(build_exploit):
raise error.TestError("Failed to build the exploit program")
error.context("Run exploit program in guest.", logging.info)
exec_exploit = "/tmp/nx_exploit"
# if nx is enabled (by default), the program failed.
# segmentation error. return value of shell is not zero.
exec_res = session.cmd_status(exec_exploit)
nx_on = params.get("nx_on", "yes")
if nx_on == "yes":
if exec_res:
logging.info("NX works good.")
error.context("Using execstack to remove the protection.", logging.info)
enable_exec = "execstack -s /tmp/nx_exploit"
if session.cmd_status(enable_exec):
if session.cmd_status("execstack --help"):
msg = "Please make sure guest have execstack command."
raise error.TestError(msg)
raise error.TestError("Failed to enable the execstack")
if session.cmd_status(exec_exploit):
raise error.TestFail("NX is still protecting. Error.")
else:
logging.info("NX is disabled as desired. good")
else:
raise error.TestFail("Fatal Error: NX does not protect anything!")
else:
if exec_res:
msg = "qemu fail to disable 'nx' flag or the exploit is corrupted."
raise error.TestError(msg)
else:
logging.info("NX is disabled, and this Test Case passed.")
if session:
session.close()
开发者ID:humanux,项目名称:tp-qemu,代码行数:59,代码来源:nx.py
示例10: receive_data
def receive_data(session, serial_receive_cmd, data_file):
output = session.cmd_output(serial_receive_cmd, timeout=30)
d_file = os.path.join(data_dir.get_deps_dir("win_serial"), data_file)
ori_data = file(data_file, "r").read()
if ori_data.strip() != output.strip():
err = "Data lost during transfer. Origin data is:\n%s" % ori_data
err += "Guest receive data:\n%s" % output
raise error.TestFail(err)
开发者ID:Chenditang,项目名称:tp-qemu,代码行数:8,代码来源:win_virtio_serial_data_transfer_reboot.py
示例11: netperf_setup
def netperf_setup(test, params, env):
"""
Setup netperf in guest.
Copy netperf package into guest. Install netperf in guest (linux only).
"""
params["start_vm"] = "yes"
params["image_snapshot"] = "no"
vm_name = params.get("main_vm")
env_process.preprocess_vm(test, params, env, vm_name)
vm = env.get_vm(vm_name)
vm.verify_alive()
session = vm.wait_for_login(timeout=int(params.get("login_timeout", 360)))
try:
if params.get("os_type") == "linux":
netperf_link = params["netperf_link"]
netperf_path = params["netperf_path"]
src_link = os.path.join(data_dir.get_deps_dir("netperf"),
netperf_link)
vm.copy_files_to(src_link, netperf_path, timeout=60)
setup_cmd = params.get("setup_cmd")
(status, output) = session.cmd_status_output(setup_cmd %
netperf_path,
timeout=600)
if status != 0:
err = "Fail to setup netperf on guest os."
err += " Command output:\n%s" % output
test.error(err)
elif params.get("os_type") == "windows":
# TODO, not suppoted by now
s_link = params.get("netperf_server_link_win",
"netserver-2.6.0.exe")
src_link = os.path.join(data_dir.get_deps_dir("netperf"),
s_link)
netperf_path = params["netperf_path"]
vm.copy_files_to(src_link, netperf_path, timeout=60)
s_link = params.get("netperf_client_link_win", "netperf.exe")
src_link = os.path.join(data_dir.get_deps_dir("netperf"),
s_link)
vm.copy_files_to(src_link, netperf_path, timeout=60)
finally:
if session:
session.close()
vm.destroy()
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:44,代码来源:ovs_host_vlan.py
示例12: dlink_preprcess
def dlink_preprcess(download_link):
"""
Preprocess the download link
"""
if not download_link:
raise error.TestNAError("Can not get the netperf download_link")
if not utils.is_url(download_link):
download_link = utils_misc.get_path(data_dir.get_deps_dir(),
download_link)
return download_link
开发者ID:QiuMike,项目名称:tp-qemu,代码行数:10,代码来源:netperf_udp.py
示例13: run
def run(test, params, env):
"""
Build and install packages from git on the client or guest VM
Supported configurations:
build_install_pkg: name of the package to get from git, build and install
:param test: QEMU test object.
:param params: Dictionary with the test parameters.
:param env: Dictionary with test environment.
"""
# Collect test parameters
pkgName = params.get("build_install_pkg")
script = params.get("script")
vm_name = params.get("vm_name")
dst_dir = params.get("dst_dir")
# Path of the script on the VM
vm_script_path = os.path.join(dst_dir, script)
# Get root session for the VM
(vm, vm_root_session) = connect_to_vm(vm_name, env, params)
# location of the script on the host
host_script_path = os.path.join(data_dir.get_deps_dir(), "spice", script)
logging.info(
"Transferring the script to %s," "destination directory: %s, source script location: %s",
vm_name,
vm_script_path,
host_script_path,
)
vm.copy_files_to(host_script_path, vm_script_path, timeout=60)
time.sleep(5)
# All packages require spice-protocol
build_install_spiceprotocol(vm_root_session, vm_script_path, params)
# Run build_install.py script
if pkgName == "xf86-video-qxl":
build_install_qxl(vm_root_session, vm_script_path, params)
elif pkgName == "spice-vd-agent":
build_install_vdagent(vm_root_session, vm_script_path, params)
elif pkgName == "spice-gtk":
build_install_spicegtk(vm_root_session, vm_script_path, params)
elif pkgName == "virt-viewer":
build_install_virtviewer(vm_root_session, vm_script_path, params)
else:
logging.info("Not supported right now")
raise error.TestFail("Incorrect Test_Setup")
utils_spice.clear_interface(vm)
开发者ID:Andrei-Stepanov,项目名称:tp-qemu,代码行数:54,代码来源:rv_build_install.py
示例14: check_cpuid_dump
def check_cpuid_dump(self):
"""
Compare full CPUID dump data
"""
machine_type = params.get("machine_type_to_check", "")
kvm_enabled = params.get("enable_kvm", "yes") == "yes"
ignore_cpuid_leaves = params.get("ignore_cpuid_leaves", "")
ignore_cpuid_leaves = ignore_cpuid_leaves.split()
whitelist = []
for l in ignore_cpuid_leaves:
l = l.split(',')
# syntax of ignore_cpuid_leaves:
# <in_eax>[,<in_ecx>[,<register>[ ,<bit>]]] ...
for i in 0, 1, 3: # integer fields:
if len(l) > i:
l[i] = int(l[i], 0)
whitelist.append(tuple(l))
if not machine_type:
raise error.TestNAError("No machine_type_to_check defined")
cpu_model_flags = params.get('cpu_model_flags', '')
full_cpu_model_name = cpu_model
if cpu_model_flags:
full_cpu_model_name += ','
full_cpu_model_name += cpu_model_flags.lstrip(',')
ref_file = os.path.join(data_dir.get_deps_dir(), 'cpuid',
"cpuid_dumps",
kvm_enabled and "kvm" or "nokvm",
machine_type, '%s-dump.txt' % (full_cpu_model_name))
if not os.path.exists(ref_file):
raise error.TestNAError("no cpuid dump file: %s" % (ref_file))
reference = open(ref_file, 'r').read()
if not reference:
raise error.TestNAError(
"no cpuid dump data on file: %s" % (ref_file))
reference = parse_cpuid_dump(reference)
if reference is None:
raise error.TestNAError(
"couldn't parse reference cpuid dump from file; %s" % (ref_file))
try:
out = get_guest_cpuid(
self, cpu_model, cpu_model_flags + ',enforce',
extra_params=dict(machine_type=machine_type, smp=1))
except virt_vm.VMStartError, e:
if "host doesn't support requested feature:" in e.reason \
or ("host cpuid" in e.reason and
("lacks requested flag" in e.reason or
"flag restricted to guest" in e.reason)) \
or ("Unable to find CPU definition:" in e.reason):
raise error.TestNAError(
"Can't run CPU model %s on this host" % (full_cpu_model_name))
else:
raise
开发者ID:QiuMike,项目名称:tp-qemu,代码行数:54,代码来源:cpuid.py
示例15: env_setup
def env_setup(session, ip, user, port, password):
error_context.context("Setup env for %s" % ip)
if params.get("env_setup_cmd"):
ssh_cmd(session, params.get("env_setup_cmd"), ignore_status=True)
pkg = params["netperf_pkg"]
pkg = os.path.join(data_dir.get_deps_dir(), pkg)
remote.scp_to_remote(ip, shell_port, username, password, pkg, "/tmp")
ssh_cmd(session, params.get("setup_cmd"))
agent_path = os.path.join(test.virtdir, "scripts/netperf_agent.py")
remote.scp_to_remote(ip, shell_port, username, password,
agent_path, "/tmp")
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:13,代码来源:netperf.py
示例16: deploy_test_form
def deploy_test_form(test, guest_vm, params):
"""
Copy wxPython Test form to guest VM.
Test form is copied to /tmp directory.
:param test
:param guest_vm - vm object
:param params
"""
script = params.get("guest_script")
script_path = os.path.join(data_dir.get_deps_dir(), "spice", script)
guest_vm.copy_files_to(script_path, "/tmp/%s" % params.get("guest_script"),
timeout=60)
开发者ID:CongLi,项目名称:tp-qemu,代码行数:14,代码来源:rv_input.py
示例17: fio_install
def fio_install(tarball):
"""
check whether fio is installed in guest, if no, install it, if yes, do nothing.
:param tarball: fio tar package
"""
if session.cmd_status(check_install_fio):
tarball = os.path.join(data_dir.get_deps_dir(), tarball)
if os_type == "linux":
vm.copy_files_to(tarball, "/tmp")
session.cmd("cd /tmp/ && tar -zxvf /tmp/%s" % os.path.basename(tarball), cmd_timeout)
session.cmd("cd %s && %s" % (fio_path, compile_cmd), cmd_timeout)
elif os_type == "windows":
session.cmd("md %s" % fio_path)
vm.copy_files_to(tarball, fio_path)
开发者ID:EIChaoYang,项目名称:tp-qemu,代码行数:15,代码来源:fio_perf.py
示例18: check_tray_status_test
def check_tray_status_test(vm, qemu_cdrom_device, guest_cdrom_device,
max_times, iso_image_new):
"""
Test cdrom tray status reporting function.
"""
error.context("Change cdrom media via monitor", logging.info)
iso_image_orig = get_cdrom_file(vm, qemu_cdrom_device)
if not iso_image_orig:
raise error.TestError("no media in cdrom")
vm.change_media(qemu_cdrom_device, iso_image_new)
is_opened = is_tray_opened(vm, qemu_cdrom_device)
if is_opened:
raise error.TestFail("cdrom tray not opened after change media")
try:
error.context("Copy test script to guest")
tray_check_src = params.get("tray_check_src")
if tray_check_src:
tray_check_src = os.path.join(data_dir.get_deps_dir(), "cdrom",
tray_check_src)
vm.copy_files_to(tray_check_src, params["tmp_dir"])
if is_tray_opened(vm, qemu_cdrom_device) is None:
logging.warn("Tray status reporting is not supported by qemu!")
logging.warn("cdrom_test_tray_status test is skipped...")
return
error.context("Eject the cdrom in guest %s times" % max_times,
logging.info)
session = vm.wait_for_login(timeout=login_timeout)
for i in range(1, max_times):
session.cmd(params["eject_cdrom_cmd"] % guest_cdrom_device)
if not is_tray_opened(vm, qemu_cdrom_device):
raise error.TestFail("Monitor reports tray closed"
" when ejecting (round %s)" % i)
if params["os_type"] != "windows":
cmd = "dd if=%s of=/dev/null count=1" % guest_cdrom_device
else:
# windows guest does not support auto close door when reading
# cdrom, so close it by eject command;
cmd = params["close_cdrom_cmd"] % guest_cdrom_device
session.cmd(cmd)
if is_tray_opened(vm, qemu_cdrom_device):
raise error.TestFail("Monitor reports tray opened when close"
" cdrom in guest (round %s)" % i)
time.sleep(workaround_eject_time)
finally:
vm.change_media(qemu_cdrom_device, iso_image_orig)
开发者ID:MiriamDeng,项目名称:tp-qemu,代码行数:47,代码来源:cdrom.py
示例19: env_setup
def env_setup(session, ip, user, port, password):
error_context.context("Setup env for %s" % ip)
ssh_cmd(session, "iptables -F", ignore_status=True)
ssh_cmd(session, "service iptables stop", ignore_status=True)
ssh_cmd(session, "systemctl stop firewalld.service",
ignore_status=True)
ssh_cmd(session, "echo 2 > /proc/sys/net/ipv4/conf/all/arp_ignore")
ssh_cmd(session, "echo 0 > /sys/kernel/mm/ksm/run", ignore_status=True)
pkg = params["netperf_pkg"]
pkg = os.path.join(data_dir.get_deps_dir(), pkg)
remote.scp_to_remote(ip, shell_port, username, password, pkg, "/tmp")
ssh_cmd(session, params.get("setup_cmd"))
agent_path = os.path.join(test.virtdir, "scripts/netperf_agent.py")
remote.scp_to_remote(ip, shell_port, username, password,
agent_path, "/tmp")
开发者ID:bssrikanth,项目名称:tp-qemu,代码行数:17,代码来源:netperf.py
示例20: run
def run(test, params, env):
"""
Test tap device deleted after vm quit with error
1) Boot a with invaild params.
1) Check qemu-kvm quit with error.
2) Check vm tap device delete from ovs bridge.
:param test: Kvm test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment.
"""
def get_ovs_ports(ovs):
cmd = "ovs-vsctl list-ports %s" % ovs
return set(utils.system_output(cmd).splitlines())
os_dep.command("ovs-vsctl")
netdst = params.get("netdst")
if netdst not in utils.system_output("ovs-vsctl list-br"):
raise error.TestError("%s isn't is openvswith bridge" % netdst)
deps_dir = data_dir.get_deps_dir("ovs")
nic_script = utils_misc.get_path(deps_dir, params["nic_script"])
nic_downscript = utils_misc.get_path(deps_dir, params["nic_downscript"])
params["nic_script"] = nic_script
params["nic_downscript"] = nic_downscript
params["qemu_command_prefix"] = "export SHELL=/usr/bin/bash;"
params["start_vm"] = "yes"
params["nettype"] = "bridge"
params["nic_model"] = "virtio-net-pci"
try:
ports = get_ovs_ports(netdst)
env_process.preprocess_vm(test, params, env, params["main_vm"])
env.get_vm(params["main_vm"])
except virt_vm.VMStartError:
ports = get_ovs_ports(netdst) - ports
if ports:
for p in ports:
utils.system("ovs-vsctl del-if %s %s" % (netdst, p))
raise error.TestFail("%s not delete after qemu quit." % ports)
else:
raise error.TestFail("Qemu should quit with error")
开发者ID:CongLi,项目名称:tp-qemu,代码行数:45,代码来源:ovs_quit.py
注:本文中的virttest.data_dir.get_deps_dir函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论