本文整理汇总了Python中virttest.utils_misc.get_qemu_binary函数的典型用法代码示例。如果您正苦于以下问题:Python get_qemu_binary函数的具体用法?Python get_qemu_binary怎么用?Python get_qemu_binary使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_qemu_binary函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: check_watchdog_support
def check_watchdog_support():
"""
check the host qemu-kvm support watchdog device
Test Step:
1. Send qemu command 'qemu -watchdog ?'
2. Check the watchdog type that the host support.
"""
qemu_binary = utils_misc.get_qemu_binary(params)
watchdog_type_check = params.get(
"watchdog_type_check", " -watchdog '?'")
qemu_cmd = qemu_binary + watchdog_type_check
# check the host support watchdog types.
error.context("Checking whether or not the host support WDT '%s'"
% watchdog_device_type, logging.info)
watchdog_device = utils.system_output("%s 2>&1" % qemu_cmd,
retain_output=True)
if watchdog_device:
if re.findall(watchdog_device_type, watchdog_device, re.I):
logging.info("The host support '%s' type watchdog device" %
watchdog_device_type)
else:
raise error.TestFail("Host not support watchdog device type %s "
% watchdog_device_type)
logging.info("The host support watchdog device type is: '%s'"
% watchdog_device)
else:
raise error.TestFail("No watchdog device support in the host!")
开发者ID:hshl1214,项目名称:tp-qemu,代码行数:29,代码来源:watchdog.py
示例2: run_test
def run_test(qemu_src_dir):
"""
run QEMU I/O test suite
:qemu_src_dir: path of qemu source code
"""
iotests_root = params.get("iotests_root", "tests/qemu-iotests")
extra_options = params.get("qemu_io_extra_options", "")
image_format = params.get("qemu_io_image_format")
result_pattern = params.get("iotests_result_pattern")
error_context.context("running qemu-iotests for image format %s"
% image_format, logging.info)
os.environ["QEMU_PROG"] = utils_misc.get_qemu_binary(params)
os.environ["QEMU_IMG_PROG"] = utils_misc.get_qemu_img_binary(params)
os.environ["QEMU_IO_PROG"] = utils_misc.get_qemu_io_binary(params)
os.environ["QEMU_NBD_PROG"] = utils_misc.get_binary('qemu-nbd', params)
os.chdir(os.path.join(qemu_src_dir, iotests_root))
cmd = './check'
if extra_options:
cmd += " %s" % extra_options
cmd += " -%s" % image_format
output = process.system_output(cmd, ignore_status=True, shell=True)
match = re.search(result_pattern, output, re.I | re.M)
if match:
iotests_log_file = "qemu_iotests_%s.log" % image_format
iotests_log_file = utils_misc.get_path(test.debugdir, iotests_log_file)
with open(iotests_log_file, 'w+') as log:
log.write(output)
log.flush()
msg = "Total test %s cases, %s failed"
raise exceptions.TestFail(msg % (match.group(2), match.group(1)))
开发者ID:bssrikanth,项目名称:tp-qemu,代码行数:31,代码来源:rh_qemu_iotests.py
示例3: check_watchdog_support
def check_watchdog_support():
"""
check the host qemu-kvm support watchdog device
Test Step:
1. Send qemu command 'qemu -watchdog ?'
2. Check the watchdog type that the host support.
"""
qemu_binary = utils_misc.get_qemu_binary(params)
watchdog_type_check = params.get(
"watchdog_type_check", " -watchdog '?'")
qemu_cmd = qemu_binary + watchdog_type_check
# check the host support watchdog types.
error_context.context("Checking whether or not the host support"
" WDT '%s'" % watchdog_device_type, logging.info)
watchdog_device = process.system_output("%s 2>&1" % qemu_cmd,
shell=True).decode()
if watchdog_device:
if re.findall(watchdog_device_type, watchdog_device, re.I):
logging.info("The host support '%s' type watchdog device" %
watchdog_device_type)
else:
logging.info("The host support watchdog device type is: '%s'"
% watchdog_device)
test.cancel("watdog %s isn't supported" % watchdog_device_type)
else:
test.cancel("No watchdog device supported by the host!")
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:28,代码来源:watchdog.py
示例4: run
def run(test, params, env):
"""
cpuinfo query test:
1). run query cmd. e.g -cpu ?cpuid
2). check the expected info is inclued in the cmd output.
3). raise error if defined info is missing.
"""
qemu_binary = utils_misc.get_qemu_binary(params)
error.context("run query cmd")
qcmd = params.get("query_cmd")
if qcmd is None:
raise error.TestError("query cmd is missing,"
"pls check query_cmd in config file")
cmd = qemu_binary + qcmd
output = utils.system_output(cmd)
error.context("check if expected info is included in output of %s " % cmd)
cpuinfos = params.get("cpu_info", "Conroe").split(",")
missing = []
for cpuinfo in cpuinfos:
if cpuinfo not in output:
missing.append(cpuinfo)
if missing:
raise error.TestFail("%s is missing in the output\n %s" %
(", ".join(missing), output))
开发者ID:CongLi,项目名称:tp-qemu,代码行数:26,代码来源:cpuinfo_query.py
示例5: run_qemu_option_check
def run_qemu_option_check(test, params, env):
"""
QEMU support options check test
Test Step:
1) Get qemu support device options
2) Check whether qemu have output
:param test: QEMU test object.
:param params: Dictionary with the test parameters.
:param env: Dictionary with test environment.
"""
def get_qemu_support_device(qemu_binary):
"""
Get qemu support device list
"""
support_device = utils.system_output("%s -device ? 2>&1"
% qemu_binary, timeout=10,
ignore_status=True)
if not support_device:
raise error.TestNAError("Can not get qemu support device list")
device_list = re.findall(r'name\s+"(.*)",', support_device)
device_list_alias = re.findall(r'alias\s+"(.*?)"', support_device)
device_list.extend(device_list_alias)
return device_list
def get_device_option(qemu_binary, device_name):
"""
Get qemu device 'device_name' support options
"""
if device_name not in get_qemu_support_device(qemu_binary):
err_msg = "Oops, Your qemu version doesn't support devic '%s', "
err_msg += "make sure you have inputted a correct device name"
raise error.TestNAError(err_msg % device_name)
device_support_option = utils.run("%s -device %s,? 2>&1" %
(qemu_binary, device_name),
timeout=10,
ignore_status=True)
if device_support_option.exit_status:
raise error.TestError("Oops, output status is wrong")
if not re.findall(r"%s\.(.*)=(.*)" % device_name,
device_support_option.stdout):
raise error.TestFail("Qemu option check Failed")
logging.info("Qemu options check successfull. output is:\n%s" %
device_support_option.stdout)
device_name = params.get("device_name")
qemu_binary = utils_misc.get_qemu_binary(params)
error.context("Get qemu support %s device options" % device_name,
logging.info)
get_device_option(qemu_binary, device_name)
开发者ID:xigao,项目名称:virt-test,代码行数:54,代码来源:qemu_option_check.py
示例6: run
def run(test, params, env):
"""
KVM Seabios bin file test:
1) Get available bin files
2) Get supported machine types
3) Check bin file in use
:param test: QEMU test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment.
"""
bin_dict = {'rhel6': 'bios.bin', 'rhel7': 'bios-256k.bin'}
error_context.context("Get available bin files", logging.info)
output = process.system_output('ls /usr/share/seabios', shell=True).decode()
for value in bin_dict.values():
if value not in output:
test.fail("%s is not available" % value)
error_context.context("Get supported machine types", logging.info)
qemu_binary = utils_misc.get_qemu_binary(params)
machine_type_cmd = qemu_binary + " -machine help | awk '{ print $1 }'"
output = process.system_output(machine_type_cmd, shell=True).decode()
machine_types = output.splitlines()
machine_type_remove = params['machine_type_remove'].split()
for i in machine_type_remove:
machine_types.remove(i)
logging.info(machine_types)
for machine in machine_types:
error_context.context("Check bin file with machine type: %s" % machine,
logging.info)
for key in bin_dict:
if key in machine:
bin_file = bin_dict[key]
break
else:
test.error("Uncertain which bin file in use for machine type: %s"
% machine)
params['machine_type'] = machine
params['start_vm'] = 'yes'
env_process.preprocess_vm(test, params, env, params.get("main_vm"))
vm = env.get_vm(params["main_vm"])
info_roms = vm.monitor.info("roms")
if bin_file not in info_roms:
test.fail("Checking bin file fails with %s, info roms: %s"
% (machine, info_roms))
vm.destroy()
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:50,代码来源:seabios_bin.py
示例7: get_all_support_flags
def get_all_support_flags():
"""
Get all supported flags with qemu query cmd.
"""
qemu_binary = utils_misc.get_qemu_binary(params)
cmd = qemu_binary + params.get("query_cmd", " -cpu ?")
output = utils.system_output(cmd)
flags_re = re.compile(params.get("pattern", "flags:(.*)"))
flag_list = flags_re.search(output)
flags = []
if flag_list:
for flag in flag_list.groups():
flags += flag
return set(map(utils_misc.Flag, flags))
开发者ID:arges,项目名称:tp-qemu,代码行数:14,代码来源:flag_check.py
示例8: run
def run(test, params, env):
"""
boot with different machine type:
1) get supported machine type
2) boot guest with different machine types
:param test: kvm test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment.
"""
error_context.context("Get supported machine type", logging.info)
qemu_binary = utils_misc.get_qemu_binary(params)
machine_types = []
machine_type_mapping = {"pc": ["i440FX", "RHEL 6"], "q35": ["Q35"], "pseries": ["pSeries"],
"arm64-pci:virt": ["ARM"], "arm64-mmio:virt": ["ARM"], "s390-ccw-virtio": ["S390"]}
for m_type, s_name in zip(*utils_misc.get_support_machine_type(qemu_binary)):
for item in machine_type_mapping[params["machine_type"]]:
if item in s_name:
if "arm64" in params["machine_type"]:
m_type = re.sub(r'(?<=:)\w+', m_type, params["machine_type"])
machine_types.append(m_type)
if not machine_types:
test.fail("Failed to get machine types")
else:
logging.info("Actual supported machine types are: " + ', '.join(map(str, machine_types)))
for m_type in machine_types:
params["machine_type"] = m_type
params["start_vm"] = "yes"
vm_name = params['main_vm']
error_context.context("Start vm with machine type '%s'"
% m_type, logging.info)
env_process.preprocess(test, params, env)
vm = env.get_vm(vm_name)
vm.verify_alive()
timeout = int(params.get("login_timeout", 360))
session = vm.wait_for_login(timeout=timeout)
if not session.is_responsive():
session.close()
test.fail("Start vm with machine type:%s fail" % m_type)
session.close()
error_context.context("Quit guest and check the process quit normally",
logging.info)
vm.destroy(gracefully=False)
开发者ID:Zhengtong,项目名称:tp-qemu,代码行数:45,代码来源:boot_with_machine_types.py
示例9: verify_machine_type
def verify_machine_type():
f_fail = []
cmd = params.get("check_machine_type_cmd")
fail_log = ""
if cmd is None:
return f_fail
status, actual_mtype = session.cmd_status_output(cmd)
if status != 0:
raise error.TestError("Failed to get machine type from vm")
machine_type_cmd = "%s -M ?" % utils_misc.get_qemu_binary(params)
machine_types = utils.system_output(machine_type_cmd,
ignore_status=True)
machine_types = machine_types.split(':')[-1]
machine_type_map = {}
for machine_type in machine_types.splitlines():
if not machine_type:
continue
type_pair = re.findall("([\w\.-]+)\s+([^(]+).*", machine_type)
if len(type_pair) == 1 and len(type_pair[0]) == 2:
machine_type_map[type_pair[0][0]] = type_pair[0][1]
else:
logging.warn("Unexpect output from qemu-kvm -M "
"?: '%s'" % machine_type)
try:
expect_mtype = machine_type_map[params['machine_type']].strip()
except KeyError:
logging.warn("Can not find machine type '%s' from qemu-kvm -M ?"
" output. Skip this test." % params['machine_type'])
return f_fail
if expect_mtype not in actual_mtype:
fail_log += " Assigned to VM: '%s' \n" % expect_mtype
fail_log += " Reported by OS: '%s'" % actual_mtype
f_fail.append(fail_log)
logging.error(fail_log)
else:
logging.info("MachineType check pass. Expected: %s, Actual: %s" %
(expect_mtype, actual_mtype))
return f_fail
开发者ID:WenliQuan,项目名称:virt-test,代码行数:42,代码来源:physical_resources_check.py
示例10: run_qemu_iotests
def run_qemu_iotests(test, params, env):
"""
Fetch from git and run qemu-iotests using the qemu binaries under test.
1) Fetch qemu-io from git
3) Run test for the file format detected
4) Report any errors found to autotest
:param test: QEMU test object.
:param params: Dictionary with the test parameters.
:param env: Dictionary with test environment.
"""
# First, let's get qemu-io
std = "http://git.kernel.org/pub/scm/virt/kvm/qemu-kvm.git"
uri = params.get("qemu_io_uri", std)
branch = params.get("qemu_io_branch", 'master')
lbranch = params.get("qemu_io_lbranch", 'master')
commit = params.get("qemu_io_commit", None)
base_uri = params.get("qemu_io_base_uri", None)
iotests_dir = params.get("qemu_iotests_dir", "tests/qemu-iotests")
destination_dir = os.path.join(test.srcdir, "qemu_io_tests")
git.get_repo(uri=uri, branch=branch, lbranch=lbranch, commit=commit,
destination_dir=destination_dir, base_uri=base_uri)
# Then, set the qemu paths for the use of the testsuite
os.environ["QEMU_PROG"] = utils_misc.get_qemu_binary(params)
os.environ["QEMU_IMG_PROG"] = utils_misc.get_qemu_img_binary(params)
os.environ["QEMU_IO_PROG"] = utils_misc.get_qemu_io_binary(params)
# qemu-iotests has merged into tests/qemu_iotests folder
os.chdir(os.path.join(destination_dir, iotests_dir))
image_format = params["qemu_io_image_format"]
extra_options = params.get("qemu_io_extra_options", "")
cmd = './check'
if extra_options:
cmd += extra_options
error.context("running qemu-iotests for image format %s" % image_format)
utils.system("%s -%s" % (cmd, image_format))
开发者ID:Antique,项目名称:virt-test,代码行数:40,代码来源:qemu_iotests.py
示例11: install_test
def install_test(build_root):
"""
Download src rpm, unpack src code and applying patches
:param build_root: build path of rpmbuild
:return: A tuple containing directory of qemu source code
and qemu-kvm spec
"""
error_context.context("Get qemu source code", logging.info)
os.chdir(test.tmpdir)
query_format = params["query_format"]
download_rpm_cmd = params["download_rpm_cmd"]
get_src_cmd = params["get_src_cmd"]
qemu_spec = params.get("qemu_spec", 'SPECS/qemu-kvm.spec')
get_rpm_name_cmd = ("rpm -qf %s --queryformat=%s" %
(utils_misc.get_qemu_binary(params), query_format))
src_rpm_name = process.system_output(get_rpm_name_cmd, shell=True)
retry_command(download_rpm_cmd % src_rpm_name)
spec = os.path.join(build_root, qemu_spec)
build_dir = os.path.join(build_root, 'BUILD')
cmd = get_src_cmd % (src_rpm_name, spec)
process.system(cmd, shell=True)
src_dir = os.listdir(build_dir)[0]
return (os.path.join(build_dir, src_dir), spec)
开发者ID:bssrikanth,项目名称:tp-qemu,代码行数:24,代码来源:rh_qemu_iotests.py
示例12: run
def run(test, params, env):
"""
Get qemu src code from src rpm and run qemu-iotests using the
qemu binaries.
1) Download src rpm from brew
2) Unpack src code and apply patches
3) Run test for the file format detected
4) Check result
:param test: QEMU test object.
:param params: Dictionary with the test parameters.
:param env: Dictionary with test environment.
"""
def retry_command(cmd):
"""
Retry the command when it fails, and raise the error once
retry has exceeded the maximum number
:param cmd: command expect to be executed
:return: output of the command
"""
max_retry = int(params.get("max_retry", 3))
retry = max_retry
while retry:
retry -= 1
try:
return process.system(cmd, shell=True)
except process.CmdError as detail:
msg = "Fail to execute command"
logging.error("%s: %s." % (msg, detail))
raise exceptions.TestError("%s after %s times retry: %s" %
(msg, max_retry, detail))
def install_test(build_root):
"""
Download src rpm, unpack src code and applying patches
:param build_root: build path of rpmbuild
:return: A tuple containing directory of qemu source code
and qemu-kvm spec
"""
error_context.context("Get qemu source code", logging.info)
os.chdir(test.tmpdir)
query_format = params["query_format"]
download_rpm_cmd = params["download_rpm_cmd"]
get_src_cmd = params["get_src_cmd"]
qemu_spec = params.get("qemu_spec", 'SPECS/qemu-kvm.spec')
get_rpm_name_cmd = ("rpm -qf %s --queryformat=%s" %
(utils_misc.get_qemu_binary(params), query_format))
src_rpm_name = process.system_output(get_rpm_name_cmd, shell=True)
retry_command(download_rpm_cmd % src_rpm_name)
spec = os.path.join(build_root, qemu_spec)
build_dir = os.path.join(build_root, 'BUILD')
cmd = get_src_cmd % (src_rpm_name, spec)
process.system(cmd, shell=True)
src_dir = os.listdir(build_dir)[0]
return (os.path.join(build_dir, src_dir), spec)
def config_test(qemu_src_dir):
"""
Generate common.env for test
:qemu_src_dir: path of qemu source code
"""
need_run_configure = params.get("need_run_configure", "no")
if need_run_configure == "yes":
make_socket_scm_helper = params.get("make_socket_scm_helper", "")
logging.info("Generate common.env")
os.chdir(qemu_src_dir)
cmd = "./configure"
if make_socket_scm_helper:
cmd += " %s" % make_socket_scm_helper
process.system(cmd, shell=True)
def run_test(qemu_src_dir):
"""
run QEMU I/O test suite
:qemu_src_dir: path of qemu source code
"""
iotests_root = params.get("iotests_root", "tests/qemu-iotests")
extra_options = params.get("qemu_io_extra_options", "")
image_format = params.get("qemu_io_image_format")
result_pattern = params.get("iotests_result_pattern")
error_context.context("running qemu-iotests for image format %s"
% image_format, logging.info)
os.environ["QEMU_PROG"] = utils_misc.get_qemu_binary(params)
os.environ["QEMU_IMG_PROG"] = utils_misc.get_qemu_img_binary(params)
os.environ["QEMU_IO_PROG"] = utils_misc.get_qemu_io_binary(params)
os.environ["QEMU_NBD_PROG"] = utils_misc.get_binary('qemu-nbd', params)
os.chdir(os.path.join(qemu_src_dir, iotests_root))
cmd = './check'
if extra_options:
cmd += " %s" % extra_options
cmd += " -%s" % image_format
output = process.system_output(cmd, ignore_status=True, shell=True)
match = re.search(result_pattern, output, re.I | re.M)
if match:
iotests_log_file = "qemu_iotests_%s.log" % image_format
#.........这里部分代码省略.........
开发者ID:bssrikanth,项目名称:tp-qemu,代码行数:101,代码来源:rh_qemu_iotests.py
示例13: run
def run(test, params, env):
"""
Boot guest with different cpu flags and check if guest works correctly.
:param test: kvm test object.
:param params: Dictionary with the test parameters.
:param env: Dictionary with test environment.
"""
utils_misc.Flag.aliases = utils_misc.kvm_map_flags_aliases
qemu_binary = utils_misc.get_qemu_binary(params)
cpuflags_src = os.path.join(data_dir.get_deps_dir("cpu_flags"), "src")
cpuflags_def = os.path.join(data_dir.get_deps_dir("cpu_flags"),
"cpu_map.xml")
smp = int(params.get("smp", 1))
all_host_supported_flags = params.get("all_host_supported_flags", "no")
mig_timeout = float(params.get("mig_timeout", "3600"))
mig_protocol = params.get("migration_protocol", "tcp")
mig_speed = params.get("mig_speed", "1G")
cpu_model_black_list = params.get("cpu_model_blacklist", "").split(" ")
multi_host_migration = params.get("multi_host_migration", "no")
class HgFlags(object):
def __init__(self, cpu_model, extra_flags=set([])):
virtual_flags = set(map(utils_misc.Flag,
params.get("guest_spec_flags", "").split()))
self.hw_flags = set(map(utils_misc.Flag,
params.get("host_spec_flags", "").split()))
self.qemu_support_flags = get_all_qemu_flags()
self.host_support_flags = set(map(utils_misc.Flag,
utils_misc.get_cpu_flags()))
self.quest_cpu_model_flags = (get_guest_host_cpuflags(cpu_model) -
virtual_flags)
self.supported_flags = (self.qemu_support_flags &
self.host_support_flags)
self.cpumodel_unsupport_flags = (self.supported_flags -
self.quest_cpu_model_flags)
self.host_unsupported_flags = (self.quest_cpu_model_flags -
self.host_support_flags)
self.all_possible_guest_flags = (self.quest_cpu_model_flags -
self.host_unsupported_flags)
self.all_possible_guest_flags |= self.cpumodel_unsupport_flags
self.guest_flags = (self.quest_cpu_model_flags -
self.host_unsupported_flags)
self.guest_flags |= extra_flags
self.host_all_unsupported_flags = set([])
self.host_all_unsupported_flags |= self.qemu_support_flags
self.host_all_unsupported_flags -= (self.host_support_flags |
virtual_flags)
def start_guest_with_cpuflags(cpuflags, smp=None, migration=False,
wait=True):
"""
Try to boot guest with special cpu flags and try login in to them.
"""
params_b = params.copy()
params_b["cpu_model"] = cpuflags
if smp is not None:
params_b["smp"] = smp
vm_name = "vm1-cpuflags"
vm = qemu_vm.VM(vm_name, params_b, test.bindir, env['address_cache'])
env.register_vm(vm_name, vm)
if (migration is True):
vm.create(migration_mode=mig_protocol)
else:
vm.create()
session = None
try:
vm.verify_alive()
if wait:
session = vm.wait_for_login()
except qemu_vm.ImageUnbootableError:
vm.destroy(gracefully=False)
raise
return (vm, session)
def get_guest_system_cpuflags(vm_session):
"""
Get guest system cpuflags.
:param vm_session: session to checked vm.
:return: [corespond flags]
"""
flags_re = re.compile(r'^flags\s*:(.*)$', re.MULTILINE)
out = vm_session.cmd_output("cat /proc/cpuinfo")
#.........这里部分代码省略.........
开发者ID:bssrikanth,项目名称:tp-qemu,代码行数:101,代码来源:cpuflags.py
示例14: run
def run(test, params, env):
"""
change a removable media:
1) Boot VM with QMP/human monitor enabled.
2) Connect to QMP/human monitor server.
3) Eject original cdrom.
4) Eject original cdrom for second time.
5) Insert new image to cdrom.
6) Eject device after add new image by change command.
7) Insert original cdrom to cdrom.
8) Try to eject non-removable device w/o force option.
:param test: QEMU test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment
"""
qemu_binary = utils_misc.get_qemu_binary(params)
if not utils_misc.qemu_has_option("qmp", qemu_binary):
logging.warn("qemu does not support qmp. Human monitor will be used.")
qmp_used = False
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
session = vm.wait_for_login(timeout=int(params.get("login_timeout", 360)))
logging.info("Wait until device is ready")
time.sleep(10)
if vm.monitor.protocol == "qmp":
qmp_used = True
orig_img_name = params.get("cdrom_cd1")
p_dict = {"file": orig_img_name}
device_name = vm.get_block(p_dict)
if device_name is None:
msg = "Fail to get device using image %s" % orig_img_name
raise error.TestFail(msg)
error.context("Eject original device.")
eject_cmd = "eject device=%s" % device_name
vm.monitor.send_args_cmd(eject_cmd)
logging.info("Wait until device is ejected")
time.sleep(10)
blocks_info = vm.monitor.info("block")
if orig_img_name in str(blocks_info):
raise error.TestFail("Fail to eject cdrom %s. " % orig_img_name)
error.context("Eject original device for second time")
vm.monitor.send_args_cmd(eject_cmd)
new_img_name = params.get("new_img_name")
error.context("Insert new image to device.")
change_cmd = "change device=%s,target=%s" % (device_name, new_img_name)
vm.monitor.send_args_cmd(change_cmd)
logging.info("Wait until device changed")
time.sleep(10)
blocks_info = vm.monitor.info("block")
if new_img_name not in str(blocks_info):
raise error.TestFail("Fail to chang cdrom to %s." % new_img_name)
if qmp_used:
eject_cmd = "eject device=%s, force=True" % device_name
else:
eject_cmd = "eject device=%s" % device_name
error.context("Eject device after add new image by change command")
vm.monitor.send_args_cmd(eject_cmd)
logging.info("Wait until new image is ejected")
time.sleep(10)
blocks_info = vm.monitor.info("block")
if new_img_name in str(blocks_info):
raise error.TestFail("Fail to eject cdrom %s." % orig_img_name)
error.context("Insert %s to device %s" % (orig_img_name, device_name))
change_cmd = "change device=%s,target=%s" % (device_name, orig_img_name)
vm.monitor.send_args_cmd(change_cmd)
logging.info("Wait until device changed")
time.sleep(10)
blocks_info = vm.monitor.info("block")
if orig_img_name not in str(blocks_info):
raise error.TestFail("Fail to change cdrom to %s." % orig_img_name)
error.context("Try to eject non-removable device")
p_dict = {"removable": False}
device_name = vm.get_block(p_dict)
if device_name is None:
raise error.TestFail("Could not find non-removable device")
if params.get("force_eject", "no") == "yes":
if not qmp_used:
eject_cmd = "eject -f %s " % device_name
else:
eject_cmd = "eject device=%s, force=True" % device_name
else:
eject_cmd = "eject device=%s," % device_name
try:
vm.monitor.send_args_cmd(eject_cmd)
except Exception, e:
if "is not removable" not in str(e):
raise error.TestFail(e)
logging.debug("Catch exception message: %s" % e)
开发者ID:CongLi,项目名称:tp-qemu,代码行数:97,代码来源:eject_media.py
示例15: int
vm.verify_alive()
timeout = int(params.get("login_timeout", 360))
test_timeout = int(params["test_timeout"])
session = vm.wait_for_login(timeout=timeout)
guest_ver = session.cmd_output("uname -r").strip()
host_ver = os.uname()[2]
kvm_userspace_ver_cmd = params.get("kvm_userspace_ver_cmd", "")
if kvm_userspace_ver_cmd:
try:
cmd_result = utils.run(kvm_userspace_ver_cmd)
qemu_version = cmd_result.stdout.strip()
except error.CmdError:
qemu_version = "Unknown"
else:
qemu_path = utils_misc.get_qemu_binary(params)
version_line = utils.system_output("%s -help | head -n 1" % qemu_path)
matches = re.findall("version .*?,", version_line, re.I)
if matches:
qemu_version = " ".join(matches[0].split()[1:]).strip(",")
else:
qemu_version = "Unknown"
# After STEP 1
try:
result_name = params.get("result_name", "nfs-perf.RHS")
result_file_path = utils_misc.get_path(test.resultsdir, result_name)
result_file = open(result_file_path, 'w')
except Exception:
_clean_up(STEP_1)
raise
开发者ID:CongLi,项目名称:tp-qemu,代码行数:31,代码来源:nfs_perf.py
示例16: run
def run(test, params, env):
"""
Check smbios table :
1) Get the smbios table from config file,if there is no config option in
config file, the script will generate the config parameters automately.
2) Boot a guest with smbios options and/or -M option
3) Verify if bios options have been emulated correctly.
:param test: QEMU test object.
:param params: Dictionary with the test parameters.
:param env: Dictionary with test environment.
"""
smbios_type = params.get("smbios_type")
notset_output = params.get("notset_output")
dmidecode_exp = params.get("dmidecode_exp")
login_timeout = float(params.get("login_timeout", 360))
smbios = ""
if params.get("smbios_type_disable", "no") == "no":
# Set the smbios parameter, if you have set it in the config file,
# it'll be honored, else, one will be generated.
for sm_type in smbios_type.split():
if sm_type == "Bios":
smbios_type_number = 0
elif sm_type == "System":
smbios_type_number = 1
smbios += " -smbios type=%s" % smbios_type_number
dmidecode_key = params.object_params(sm_type).get("dmikeyword")
dmidecode_key = dmidecode_key.split()
for key in dmidecode_key:
cmd = (dmidecode_exp % (smbios_type_number, key))
default_key_para = decode_to_text(process.system_output(
cmd, shell=True).strip())
smbios_key_para_set = params.object_params(sm_type).get(key,
default_key_para)
smbios += ",%s='%s'" % (key.lower(), smbios_key_para_set)
if params.get("extra_params"):
params["extra_params"] += smbios
else:
params["extra_params"] = smbios
support_machine_types = []
if params.get("traversal_machine_emulated", "no") == "no":
support_machine_types.append(params.get("machine_type"))
else:
qemu_binary = utils_misc.get_qemu_binary(params)
tmp = utils_misc.get_support_machine_type(qemu_binary, remove_alias=True)[:2]
(support_machine_types, expect_system_versions) = tmp
machine_type = params.get("machine_type", "")
if ':' in machine_type:
prefix = machine_type.split(':', 1)[0]
support_machine_types = ["%s:%s" % (prefix, m_type)
for m_type in support_machine_types]
failures = []
rhel_system_version = params.get('smbios_system_version') == 'rhel'
if not rhel_system_version:
re_pc_lt_2 = re.compile(r'^pc-(i440fx-)?[01].\d+$')
host_dmidecode_system_version = decode_to_text(
process.system_output("dmidecode -s system-version"))
for m_type in support_machine_types:
if m_type in ("isapc", "xenfv", "xenpv"):
continue
params["machine_type"] = m_type
params["start_vm"] = "yes"
error_context.context("Boot the vm using -M option:'-M %s', smbios "
"para: '%s'" % (m_type, smbios), logging.info)
env_process.preprocess_vm(test, params, env, params.get("main_vm"))
vm1 = env.get_vm(params["main_vm"])
session = vm1.wait_for_login(timeout=login_timeout)
error_context.context("Check smbios info on guest "
"is setted as expected")
for sm_type in smbios_type.split():
if sm_type == "Bios":
smbios_type_number = 0
elif sm_type == "System":
smbios_type_number = 1
dmidecode_key = params.object_params(sm_type).get("dmikeyword")
dmidecode_key = dmidecode_key.split()
for key in dmidecode_key:
cmd = (dmidecode_exp % (smbios_type_number, key))
smbios_get_para = session.cmd(cmd).strip()
default_key_para = decode_to_text(process.system_output(
cmd, shell=True).strip())
if params.get("smbios_type_disable", "no") == "no":
smbios_set_para = params.object_params(sm_type).get(key,
default_key_para)
else:
# The System.Version is different on RHEL and upstream
if (rhel_system_version or sm_type != 'System' or
key != 'Version'):
key_index = support_machine_types.index(m_type)
smbios_set_para = expect_system_versions[key_index]
elif re_pc_lt_2.match(m_type):
# pc<2.0 inherits host system-version
smbios_set_para = host_dmidecode_system_version
#.........这里部分代码省略.........
开发者ID:ldoktor,项目名称:tp-qemu,代码行数:101,代码来源:smbios_table.py
示例17: run
def run(test, params, env):
"""
slof_device_tree test:
steps:
1. Boot up a guest.
2. Check the guest device tree information
3. Compare the value with expected result.
:param test: Kvm test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment.
"""
def get_info(vm_session, guest_info):
"""
Check the corresponding information from vm.
:param vm_session: session to checked vm.
:return: corresponding prompt
"""
(status, output) = vm_session.cmd_status_output("echo `cat /proc/device-tree/%s`" % guest_info)
if status != 0:
raise exceptions.TestFail("Failed to get %s" % guest_info)
return output.strip()
def check_nonexist_aliases(vm_session, devices):
"""
Check a nonexist device aliases.
:param vm_session: session to checked vm.
:return: corresponding prompt
"""
status = vm_session.cmd_status("test -f /proc/device-tree/aliases/cdrom")
error_context.context("Checking whether aliases file is indeed nonexisting", logging.info)
if status == 0:
raise exceptions.TestFail("Nonexist cdrom aliases check failed.")
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
timeout = int(params.get("login_timeout", 600))
session = vm.wait_for_login(timeout=timeout)
try:
guest_system_id = get_info(session, "system-id")
if guest_system_id != vm.get_uuid():
raise exceptions.TestFail("Guest system id does not match to uuid")
guest_uuid = get_info(session, "vm,uuid")
if guest_uuid != vm.get_uuid():
raise exceptions.TestFail("Guest uuid does not match to expected id.")
host_system_id = process.system_output("echo `cat /proc/device-tree/system-id`", shell=True).strip()
host_system_id_in_guest = get_info(session, "host-serial")
if host_system_id != host_system_id_in_guest:
raise exceptions.TestFail("Host system id does not match to value in guest.")
guest_partition_name = get_info(session, "ibm,partition-name")
if guest_partition_name != params.get("main_vm&qu
|
请发表评论