Skip to content

T
tp-qemu

openeuler / tp-qemu

通知 1
Star 0
Fork 0
T
tp-qemu

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 e41006a0 编写于 作者: X Xu Han
浏览文件
操作

[qemu] Replace autotest modules - q 

Signed-off-by: NXu Han <xuhan@redhat.com>
上级 59f74519
隐藏空白更改
内联 并排
Showing with 361 addition and 375 deletion
qemu/tests/qcow2perf.py
浏览文件 @ e41006a0
import re
import logging
from autotest.client.shared import error
from autotest.client import utils
from avocado.utils import process
from virttest import data_dir
from virttest import error_context
from virttest.qemu_storage import QemuImg
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
Run qcow2 performance tests:
......@@ -41,7 +41,7 @@ def run(test, params, env):
interval_size = int(interval_size)
write_unit = ""
error.context("Init images for testing", logging.info)
error_context.context("Init images for testing", logging.info)
sn_list = []
for img in re.split("\s+", image_chain.strip()):
image_params = params.object_params(img)
......@@ -50,8 +50,8 @@ def run(test, params, env):
sn_list.append((sn_tmp, image_params))
# Write to the test image
error.context("Prepare the image with write a certain size block",
logging.info)
error_context.context("Prepare the image with write a certain size block",
logging.info)
dropcache = 'echo 3 > /proc/sys/vm/drop_caches && sleep 5'
snapshot_file = sn_list[test_image][0].image_filename
......@@ -61,30 +61,30 @@ def run(test, params, env):
write_unit, interval_size, write_unit)
iocmd0 = iocmd % (writecmd0, io_options, snapshot_file)
logging.info("writecmd-offset-0: %s", writecmd0)
utils.run(dropcache)
output = utils.run(iocmd0)
process.run(dropcache)
output = process.run(iocmd0)
else:
offset = 1
writecmd1 = writecmd % (write_round, offset, interval_size,
write_unit, interval_size, write_unit)
iocmd1 = iocmd % (writecmd1, io_options, snapshot_file)
logging.info("writecmd-offset-1: %s", writecmd1)
utils.run(dropcache)
output = utils.run(iocmd1)
process.run(dropcache)
output = process.run(iocmd1)
error.context("Do one operations to the image and measure the time",
logging.info)
error_context.context("Do one operations to the image and "
"measure the time", logging.info)
if op_type == "read":
readcmd = opcmd % (io_options, snapshot_file)
logging.info("read: %s", readcmd)
utils.run(dropcache)
output = utils.run(readcmd)
process.run(dropcache)
output = process.run(readcmd)
elif op_type == "commit":
commitcmd = opcmd % (cache_mode, snapshot_file)
logging.info("commit: %s", commitcmd)
utils.run(dropcache)
output = utils.run(commitcmd)
process.run(dropcache)
output = process.run(commitcmd)
elif op_type == "rebase":
new_base_img = QemuImg(params.object_params(new_base), image_dir,
new_base)
......@@ -92,16 +92,16 @@ def run(test, params, env):
rebasecmd = opcmd % (new_base_img.image_filename,
cache_mode, snapshot_file)
logging.info("rebase: %s", rebasecmd)
utils.run(dropcache)
output = utils.run(rebasecmd)
process.run(dropcache)
output = process.run(rebasecmd)
elif op_type == "convert":
convertname = sn_list[test_image][0].image_filename + "_convert"
convertcmd = opcmd % (snapshot_file, cache_mode, convertname)
logging.info("convert: %s", convertcmd)
utils.run(dropcache)
output = utils.run(convertcmd)
process.run(dropcache)
output = process.run(convertcmd)
error.context("Result recording", logging.info)
error_context.context("Result recording", logging.info)
result_file = open("%s/%s_%s_results" %
(test.resultsdir, "qcow2perf", op_type), 'w')
result_file.write("%s:%s\n" % (op_type, output))
......
qemu/tests/qemu_disk_img.py
浏览文件 @ e41006a0
......@@ -2,11 +2,11 @@ import os
import re
import logging
from autotest.client.shared import error
from autotest.client import utils
from avocado.utils import process
from virttest import data_dir
from virttest import env_process
from virttest import error_context
from virttest import storage
from virttest import qemu_storage
from virttest import utils_test
......@@ -28,12 +28,12 @@ class QemuImgTest(qemu_storage.QemuImg):
t_params = params.object_params(tag)
super(QemuImgTest, self).__init__(t_params, self.data_dir, tag)
@error.context_aware
@error_context.context_aware
def create_snapshot(self, t_params=None):
"""
create snapshot image file
"""
error.context("create snapshot image")
error_context.context("create snapshot image")
params = self.params.object_params(self.tag)
if t_params:
params.update(t_params)
......@@ -41,17 +41,17 @@ class QemuImgTest(qemu_storage.QemuImg):
return {}
snapshot = storage.get_image_filename(params, self.data_dir)
if os.path.exists(snapshot):
utils.run("rm -f %s" % snapshot)
process.run("rm -f %s" % snapshot)
super(QemuImgTest, self).create(params)
self.trash.append(snapshot)
return params
@error.context_aware
@error_context.context_aware
def start_vm(self, t_params=None):
"""
Start a vm and wait for it bootup;
"""
error.context("start vm", logging.info)
error_context.context("start vm", logging.info)
params = self.params.object_params(self.tag)
if t_params:
params.update(t_params)
......@@ -71,14 +71,14 @@ class QemuImgTest(qemu_storage.QemuImg):
self.vm = vm
return vm
@error.context_aware
@error_context.context_aware
def __create_file(self, dst):
logging.info("create tmp file on host")
if not self.vm:
return False
src = self.params["tmp_file_name"]
cmd = self.params["file_create_cmd"] % src
utils.run(cmd)
process.run(cmd)
self.vm.copy_files_to(src, dst)
self.trash.append(src)
return True
......@@ -98,11 +98,12 @@ class QemuImgTest(qemu_storage.QemuImg):
md5 = re.findall("\w{32}", output)[0]
return md5
@error.context_aware
@error_context.context_aware
def save_file(self, dst):
login_timeout = int(self.params.get("login_timeout", 360))
cmd = self.params.get("sync_bin", "sync")
error.context("save file('%s') md5sum in guest" % dst, logging.info)
error_context.context("save file('%s') md5sum in guest" % dst,
logging.info)
self.__create_file(dst)
session = self.vm.wait_for_login(timeout=login_timeout)
logging.info("sync guest data")
......@@ -114,9 +115,10 @@ class QemuImgTest(qemu_storage.QemuImg):
session.close()
return self.__md5sum(dst)
@error.context_aware
@error_context.context_aware
def check_file(self, dst, md5):
error.context("check file('%s') md5sum in guest" % dst, logging.info)
error_context.context("check file('%s') md5sum in guest" % dst,
logging.info)
if md5 != self.__md5sum(dst):
err = ("Md5 value does not match. "
"Expected value: %s Actual value: %s" %
......@@ -125,31 +127,31 @@ class QemuImgTest(qemu_storage.QemuImg):
return False
return True
@error.context_aware
@error_context.context_aware
def destroy_vm(self):
error.context("destroy vm", logging.info)
error_context.context("destroy vm", logging.info)
if self.vm:
self.vm.destroy()
self.vm = None
@error.context_aware
@error_context.context_aware
def check_image(self, t_params=None):
error.context("check image file ('%s')" % self.image_filename,
logging.info)
error_context.context("check image file ('%s')" % self.image_filename,
logging.info)
t_params = t_params or {}
return super(QemuImgTest, self).check_image(t_params, self.data_dir)
@error.context_aware
@error_context.context_aware
def get_info(self):
error.context("get image file ('%s')" % self.image_filename)
error_context.context("get image file ('%s')" % self.image_filename)
return super(QemuImgTest, self).info()
@error.context_aware
@error_context.context_aware
def verify_info(self, params=None):
"""
verify option is applied to image file correctly
"""
error.context("verify option of converted image", logging.info)
error_context.context("verify option of converted image", logging.info)
image_filename = storage.get_image_filename(params, self.data_dir)
info = utils_test.get_image_info(image_filename)
avalue = evalue = ""
......@@ -174,7 +176,7 @@ class QemuImgTest(qemu_storage.QemuImg):
if avalue is not None and avalue != evalue:
msg = "Get wrong %s from image %s!" % (option, image_filename)
msg += "Expect: %s, actual: %s" % (evalue, avalue)
raise error.TestFail(msg)
self.test.fail(msg)
@error_context.context_aware
def check_backingfile(self):
......@@ -197,12 +199,12 @@ class QemuImgTest(qemu_storage.QemuImg):
self.image_filename)
raise exceptions.TestFail(msg)
@error.context_aware
@error_context.context_aware
def clean(self):
error.context("clean up useless images")
error_context.context("clean up useless images")
self.destroy_vm()
for temp in self.trash:
utils.run("rm -f %s" % temp)
process.run("rm -f %s" % temp)
def run(test, params, env):
......
qemu/tests/qemu_disk_img_commit.py
浏览文件 @ e41006a0
import logging
from autotest.client.shared import error
from avocado.utils import process
from virttest import error_context
from qemu.tests import qemu_disk_img
from autotest.client import utils
class CommitTest(qemu_disk_img.QemuImgTest):
......@@ -11,12 +14,12 @@ class CommitTest(qemu_disk_img.QemuImgTest):
t_params = params.object_params(self.tag)
super(CommitTest, self).__init__(test, t_params, env, self.tag)
@error.context_aware
@error_context.context_aware
def commit(self, t_params=None):
"""
commit snapshot to backing file;
"""
error.context("commit snapshot to backingfile", logging.info)
error_context.context("commit snapshot to backingfile", logging.info)
params = self.params.object_params(self.tag)
if t_params:
params.update(t_params)
......@@ -24,7 +27,7 @@ class CommitTest(qemu_disk_img.QemuImgTest):
return super(CommitTest, self).commit(params, cache_mode)
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
'qemu-img' commit functions test:
......@@ -46,15 +49,15 @@ def run(test, params, env):
# save file md5sum before commit
md5 = commit_test.save_file(t_file)
if not md5:
raise error.TestError("Fail to save tmp file")
test.error("Fail to save tmp file")
commit_test.destroy_vm()
commit_test.commit()
error.context("sync host data after commit", logging.info)
utils.system("sync")
error_context.context("sync host data after commit", logging.info)
process.system("sync")
commit_test.start_vm(params)
# check md5sum after commit
ret = commit_test.check_file(t_file, md5)
if not ret:
raise error.TestError("image content changed after commit")
test.error("image content changed after commit")
commit_test.clean()
qemu/tests/qemu_disk_img_convert.py
浏览文件 @ e41006a0
import logging
from autotest.client import utils
from autotest.client.shared import error
from avocado.core import exceptions
from avocado.utils import process
from virttest import storage
from virttest import error_context
from avocado.core import exceptions
from qemu.tests import qemu_disk_img
......@@ -18,12 +16,12 @@ class ConvertTest(qemu_disk_img.QemuImgTest):
t_params = params.object_params(self.tag)
super(ConvertTest, self).__init__(test, t_params, env, self.tag)
@error.context_aware
@error_context.context_aware
def convert(self, t_params=None):
"""
create image file from one format to another format
"""
error.context("convert image file", logging.info)
error_context.context("convert image file", logging.info)
params = self.params.object_params(self.tag)
if t_params:
params.update(t_params)
......@@ -32,7 +30,7 @@ class ConvertTest(qemu_disk_img.QemuImgTest):
params["image_name"] = params["convert_name"]
params["image_format"] = params["convert_format"]
converted = storage.get_image_filename(params, self.data_dir)
utils.run("sync")
process.run("sync")
self.trash.append(converted)
return params
......@@ -56,7 +54,7 @@ class ConvertTest(qemu_disk_img.QemuImgTest):
if not is_strict:
raise
if is_strict and cmd_result:
raise error.TestFail("images are identical in strict mode")
self.test.fail("images are identical in strict mode")
def run(test, params, env):
......@@ -80,7 +78,7 @@ def run(test, params, env):
# save file md5sum before conversion
md5 = convert_test.save_file(t_file)
if not md5:
raise error.TestError("Fail to save tmp file")
test.error("Fail to save tmp file")
convert_test.destroy_vm()
n_params = convert_test.convert()
convert_test.compare_test(n_params)
......@@ -90,5 +88,5 @@ def run(test, params, env):
# check md5sum after conversion
ret = convert_test.check_file(t_file, md5)
if not ret:
raise error.TestError("image content changed after convert")
test.error("image content changed after convert")
convert_test.clean()
qemu/tests/qemu_disk_img_info.py
浏览文件 @ e41006a0
from autotest.client import utils
from autotest.client.shared import error
from avocado.utils import process
from virttest import storage
......@@ -18,7 +17,7 @@ class InfoTest(qemu_disk_img.QemuImgTest):
for sn in params.get("image_chain").split()[1:]:
_params = params.object_params(sn)
_image = storage.get_image_filename(_params, self.data_dir)
utils.run("rm -f %s" % _image)
process.run("rm -f %s" % _image)
def run(test, params, env):
......@@ -45,12 +44,12 @@ def run(test, params, env):
for _file in check_files:
ret = info_test.check_file(_file, md5_dict[_file])
if not ret:
raise error.TestError("Check md5sum fail (file:%s)" % _file)
test.error("Check md5sum fail (file:%s)" % _file)
#save file in guest
t_file = params["guest_file_name_%s" % tag]
md5 = info_test.save_file(t_file)
if not md5:
raise error.TestError("Fail to save tmp file")
test.error("Fail to save tmp file")
check_files.append(t_file)
md5_dict[t_file] = md5
info_test.destroy_vm()
......
qemu/tests/qemu_disk_img_rebase.py
浏览文件 @ e41006a0
import copy
import logging
from autotest.client import utils
from autotest.client.shared import error
from avocado.utils import process
from virttest import error_context
from virttest import storage
from qemu.tests import qemu_disk_img
......@@ -16,12 +16,12 @@ class RebaseTest(qemu_disk_img.QemuImgTest):
t_params = params.object_params(tag)
super(RebaseTest, self).__init__(test, t_params, env, tag)
@error.context_aware
@error_context.context_aware
def rebase(self, t_params=None):
"""
Rebase snapshot, AKA changes backing file to new image;
"""
error.context("rebase snapshot to backingfile", logging.info)
error_context.context("rebase snapshot to backingfile", logging.info)
params = self.params.object_params(self.tag)
if t_params:
params.update(t_params)
......@@ -34,7 +34,7 @@ class RebaseTest(qemu_disk_img.QemuImgTest):
for sn in params.get("image_chain").split()[1:]:
_params = params.object_params(sn)
_image = storage.get_image_filename(_params, self.data_dir)
utils.run("rm -f %s" % _image)
process.run("rm -f %s" % _image)
def run(test, params, env):
......@@ -60,7 +60,7 @@ def run(test, params, env):
t_file = params["guest_file_name_%s" % tag]
md5 = rebase_test.save_file(t_file)
if not md5:
raise error.TestError("Fail to save tmp file")
test.error("Fail to save tmp file")
md5_dict[t_file] = md5
rebase_test.destroy_vm()
......@@ -74,7 +74,7 @@ def run(test, params, env):
except IndexError:
msg = "Invalid format of'rebase_chain' params \n"
msg += "format like: 'image > base;image> base2;image2 > base2'"
raise error.TestError(msg)
test.error(msg)
params["image_chain"] = " ".join([base, image])
params["base_image_filename"] = image
......@@ -86,7 +86,7 @@ def run(test, params, env):
for _file in check_files:
ret = rebase_test.check_file(_file, md5_dict[_file])
if not ret:
raise error.TestError("Check md5sum fail (file:%s)" % _file)
test.error("Check md5sum fail (file:%s)" % _file)
rebase_test.destroy_vm()
rebase_test.check_image()
......
qemu/tests/qemu_guest_agent_snapshot.py
浏览文件 @ e41006a0
import logging
from autotest.client import utils
from autotest.client.shared import error
from avocado.utils import crypto
from avocado.utils import process
from virttest import error_context
from virttest import utils_misc
from qemu.tests.live_snapshot_basic import LiveSnapshot
......@@ -11,7 +12,7 @@ from qemu.tests.qemu_guest_agent import QemuGuestAgentBasicCheck
class QemuGuestAgentSnapshotTest(QemuGuestAgentBasicCheck):
@error.context_aware
@error_context.context_aware
def _action_before_fsfreeze(self, *args):
copy_timeout = int(self.params.get("copy_timeoout", 600))
file_size = int(self.params.get("file_size", "500"))
......@@ -20,49 +21,50 @@ class QemuGuestAgentSnapshotTest(QemuGuestAgentBasicCheck):
if self.params.get("os_type") != "linux":
self.guest_path = r"c:\%s" % tmp_name
error.context("Create a file in host.")
utils.run("dd if=/dev/urandom of=%s bs=1M count=%s" % (self.host_path,
file_size))
self.orig_hash = utils.hash_file(self.host_path)
error.context("Transfer file from %s to %s" % (self.host_path,
self.guest_path), logging.info)
self.bg = utils.InterruptedThread(self.vm.copy_files_to,
(self.host_path, self.guest_path),
dict(verbose=True, timeout=copy_timeout))
error_context.context("Create a file in host.")
process.run("dd if=/dev/urandom of=%s bs=1M count=%s"
% (self.host_path, file_size))
self.orig_hash = crypto.hash_file(self.host_path)
error_context.context("Transfer file from %s to %s" %
(self.host_path, self.guest_path), logging.info)
self.bg = utils_misc.InterruptedThread(
self.vm.copy_files_to,
(self.host_path, self.guest_path),
dict(verbose=True, timeout=copy_timeout))
self.bg.start()
@error.context_aware
@error_context.context_aware
def _action_after_fsfreeze(self, *args):
if self.bg.isAlive():
image_tag = self.params.get("image_name", "image1")
image_params = self.params.object_params(image_tag)
snapshot_test = LiveSnapshot(self.test, self.params,
self.env, image_tag)
error.context("Creating snapshot", logging.info)
error_context.context("Creating snapshot", logging.info)
snapshot_test.create_snapshot()
error.context("Checking snapshot created successfully",
logging.info)
error_context.context("Checking snapshot created successfully",
logging.info)
snapshot_test.check_snapshot()
@error.context_aware
@error_context.context_aware
def _action_before_fsthaw(self, *args):
pass
@error.context_aware
@error_context.context_aware
def _action_after_fsthaw(self, *args):
if self.bg:
self.bg.join()
# Make sure the returned file is identical to the original one
self.host_path_returned = "%s-returned" % self.host_path
self.vm.copy_files_from(self.guest_path, self.host_path_returned)
error.context("comparing hashes", logging.info)
self.curr_hash = utils.hash_file(self.host_path_returned)
error_context.context("comparing hashes", logging.info)
self.curr_hash = crypto.hash_file(self.host_path_returned)
if self.orig_hash != self.curr_hash:
raise error.TestFail("Current file hash (%s) differs from "
"original one (%s)" % (self.curr_hash,
self.orig_hash))
self.test.fail("Current file hash (%s) differs from "
"original one (%s)" % (self.curr_hash,
self.orig_hash))
error.context("Reboot and shutdown guest.")
error_context.context("Reboot and shutdown guest.")
self.vm.reboot()
self.vm.destroy()
......
qemu/tests/qemu_guest_agent_suspend.py
浏览文件 @ e41006a0
import logging
from autotest.client.shared import error
from virttest import error_context
from virttest import guest_agent
from generic.tests.guest_suspend import GuestSuspendBaseTest
......@@ -13,12 +12,12 @@ class SuspendViaGA(GuestSuspendBaseTest):
guest_agent = None
suspend_mode = ""
@error.context_aware
@error_context.context_aware
def start_suspend(self, **args):
"""
Start suspend via qemu guest agent.
"""
error.context("Suspend guest via guest agent", logging.info)
error_context.context("Suspend guest via guest agent", logging.info)
if self.guest_agent:
self.guest_agent.suspend(self.suspend_mode)
......@@ -36,19 +35,19 @@ class QemuGASuspendTest(QemuGuestAgentTest):
def run_once(self, test, params, env):
QemuGuestAgentTest.run_once(self, test, params, env)
error.context("Suspend guest to memory", logging.info)
error_context.context("Suspend guest to memory", logging.info)
gs = SuspendViaGA(params, self.vm)
gs.guest_agent = self.gagent
gs.suspend_mode = guest_agent.QemuAgent.SUSPEND_MODE_RAM
gs.guest_suspend_mem(params)
error.context("Suspend guest to disk", logging.info)
error_context.context("Suspend guest to disk", logging.info)
gs.suspend_mode = guest_agent.QemuAgent.SUSPEND_MODE_DISK
gs.guest_suspend_disk(params)
# Reset guest agent object to None after guest reboot.
self.gagent = None
error.context("Check if guest agent work again.", logging.info)
error_context.context("Check if guest agent work again.", logging.info)
session = self._get_session(params, self.vm)
self.gagent_start(session, self.vm)
session.close()
......@@ -57,7 +56,7 @@ class QemuGASuspendTest(QemuGuestAgentTest):
self.gagent.verify_responsive()
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
Test suspend commands in qemu guest agent.
......
qemu/tests/qemu_img.py
浏览文件 @ e41006a0
......@@ -6,9 +6,9 @@ import commands
import shutil
import tempfile
from autotest.client.shared import error
from autotest.client.shared import utils
from avocado.utils import process
from virttest import error_context
from virttest import utils_misc
from virttest import env_process
from virttest import storage
......@@ -16,7 +16,7 @@ from virttest import data_dir
from virttest import gluster
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
'qemu-img' functions test:
......@@ -30,7 +30,7 @@ def run(test, params, env):
qemu_img_binary = utils_misc.get_qemu_img_binary(params)
cmd = qemu_img_binary
if not os.path.exists(cmd):
raise error.TestError("Binary of 'qemu-img' not found")
test.error("Binary of 'qemu-img' not found")
image_format = params["image_format"]
image_size = params.get("image_size", "10G")
enable_gluster = params.get("enable_gluster", "no") == "yes"
......@@ -70,11 +70,11 @@ def run(test, params, env):
:param img: image to be checked
"""
cmd += " check %s" % img
error.context("Checking image '%s' by command '%s'" % (img, cmd),
logging.info)
error_context.context("Checking image '%s' by command '%s'"
% (img, cmd), logging.info)
try:
output = utils.system_output(cmd, verbose=False)
except error.CmdError, err:
output = process.system_output(cmd, verbose=False)
except process.CmdError, err:
if "does not support checks" in str(err):
return (True, "")
else:
......@@ -95,19 +95,19 @@ def run(test, params, env):
create_image_cmd = params["create_image_cmd"]
create_image_cmd = create_image_cmd % test_image
msg = " Create image %s by command %s" % (test_image, create_image_cmd)
error.context(msg, logging.info)
utils.system(create_image_cmd, verbose=False)
error_context.context(msg, logging.info)
process.system(create_image_cmd, verbose=False)
status, output = _check(cmd, test_image)
if not status:
raise error.TestFail("Check image '%s' failed with error: %s" %
(test_image, output))
test.fail("Check image '%s' failed with error: %s" %
(test_image, output))
for fmt in params["supported_image_formats"].split():
output_image = test_image + ".%s" % fmt
_convert(cmd, fmt, test_image, output_image)
status, output = _check(cmd, output_image)
if not status:
raise error.TestFail("Check image '%s' got error: %s" %
(output_image, output))
test.fail("Check image '%s' got error: %s" %
(output_image, output))
remove(output_image)
remove(test_image)
......@@ -152,12 +152,11 @@ def run(test, params, env):
cmd += " %s" % img_size
msg = "Creating image %s by command %s" % (img_name, cmd)
error.context(msg, logging.info)
utils.system(cmd, verbose=False)
error_context.context(msg, logging.info)
process.system(cmd, verbose=False)
status, out = _check(qemu_img_binary, img_name)
if not status:
raise error.TestFail("Check image '%s' got error: %s" %
(img_name, out))
test.fail("Check image '%s' got error: %s" % (img_name, out))
def create_test(cmd):
"""
......@@ -187,8 +186,8 @@ def run(test, params, env):
end_time = time.time() + timeout
while time.time() < end_time:
time.sleep(1)
status = utils.system("kill -SIGUSR1 `pidof qemu-img`",
ignore_status=True)
status = process.system("kill -SIGUSR1 `pidof qemu-img`",
ignore_status=True)
if status == 0:
return None
logging.info("Fail to get pid of qemu-img")
......@@ -201,18 +200,18 @@ def run(test, params, env):
logging.info("Check result of command")
check_output = params.get("check_output", "exit_status")
if not hasattr(CmdResult, check_output):
raise error.TestError("Unknown check output '%s'" % check_output)
test.error("Unknown check output '%s'" % check_output)
output = getattr(CmdResult, check_output)
if check_output == "exit_status" and output == 0:
return None
if check_output == "exit_status" and output != 0:
err_msg = "Get nonzero exit status(%d) '%s'"
raise error.TestFail(err_msg % (output, CmdResult.command))
test.fail(err_msg % (output, CmdResult.command))
pattern = params.get("command_result_pattern")
if not re.findall(pattern, output):
err_msg = "Fail to get expected result!"
err_msg += "Output: %s, expected pattern: %s" % (output, pattern)
raise error.TestFail(err_msg)
test.fail(err_msg)
def _convert(cmd, output_fmt, img_name, output_filename,
fmt=None, compressed="no", encrypted="no"):
......@@ -249,11 +248,11 @@ def run(test, params, env):
cmd += " %s %s" % (img_name, output_filename)
msg = "Converting '%s' from format '%s'" % (img_name, fmt)
msg += " to '%s'" % output_fmt
error.context(msg, logging.info)
error_context.context(msg, logging.info)
if show_progress == "off":
bg = utils.InterruptedThread(send_signal)
bg = utils_misc.InterruptedThread(send_signal)
bg.start()
check_command_output(utils.run(cmd, ignore_status=True))
check_command_output(process.run(cmd, ignore_status=True))
def convert_test(cmd):
"""
......@@ -277,8 +276,8 @@ def run(test, params, env):
if status:
remove(output_filename)
else:
raise error.TestFail("Check image '%s' failed with error: %s" %
(output_filename, output))
test.fail("Check image '%s' failed with error: %s" %
(output_filename, output))
else:
remove(output_filename)
......@@ -297,8 +296,8 @@ def run(test, params, env):
cmd += " %s" % img
try:
output = utils.system_output(cmd)
except error.CmdError, err:
output = process.system_output(cmd)
except process.CmdError, err:
logging.error("Get info of image '%s' failed: %s", img, str(err))
return None
......@@ -323,11 +322,11 @@ def run(test, params, env):
img_info = _info(cmd, image_name)
logging.info("Info of image '%s':\n%s", image_name, img_info)
if image_format not in img_info:
raise error.TestFail("Got unexpected format of image '%s'"
" in info test" % image_name)
test.fail("Got unexpected format of image '%s'"
" in info test" % image_name)
if image_size not in img_info:
raise error.TestFail("Got unexpected size of image '%s'"
" in info test" % image_name)
test.fail("Got unexpected size of image '%s'"
" in info test" % image_name)
def snapshot_test(cmd):
"""
......@@ -343,27 +342,27 @@ def run(test, params, env):
msg = "Created snapshot '%s' in '%s' by command %s" % (sn_name,
image_name,
crtcmd)
error.context(msg, logging.info)
error_context.context(msg, logging.info)
status, output = commands.getstatusoutput(crtcmd)
if status != 0:
raise error.TestFail("Create snapshot failed via command: %s;"
"Output is: %s" % (crtcmd, output))
test.fail("Create snapshot failed via command: %s;"
"Output is: %s" % (crtcmd, output))
listcmd = cmd
listcmd += " -l %s" % image_name
status, out = commands.getstatusoutput(listcmd)
if not ("snapshot0" in out and "snapshot1" in out and status == 0):
raise error.TestFail("Snapshot created failed or missed;"
"snapshot list is: \n%s" % out)
test.fail("Snapshot created failed or missed;"
"snapshot list is: \n%s" % out)
for i in range(2):
sn_name = "snapshot%d" % i
delcmd = cmd
delcmd += " -d %s %s" % (sn_name, image_name)
msg = "Delete snapshot '%s' by command %s" % (sn_name, delcmd)
error.context(msg, logging.info)
error_context.context(msg, logging.info)
status, output = commands.getstatusoutput(delcmd)
if status != 0:
raise error.TestFail("Delete snapshot '%s' failed: %s" %
(sn_name, output))
test.fail("Delete snapshot '%s' failed: %s" %
(sn_name, output))
def commit_test(cmd):
"""
......@@ -404,11 +403,11 @@ def run(test, params, env):
image_format,
overlay_file_name)
msg = "Create overlay file by command: %s" % create_cmd
error.context(msg, logging.info)
error_context.context(msg, logging.info)
try:
utils.system(create_cmd, verbose=False)
except error.CmdError:
raise error.TestFail("Could not create a overlay file!")
process.system(create_cmd, verbose=False)
except process.CmdError:
test.fail("Could not create a overlay file!")
logging.info("overlay file (%s) created!" % overlay_file_name)
# Set the qemu harddisk to the overlay file
......@@ -419,7 +418,7 @@ def run(test, params, env):
params.get('image_name'))
msg = "Start a new VM, using overlay file as its harddisk"
error.context(msg, logging.info)
error_context.context(msg, logging.info)
vm_name = params['main_vm']
env_process.preprocess_vm(test, params, env, vm_name)
vm = env.get_vm(vm_name)
......@@ -434,23 +433,23 @@ def run(test, params, env):
output = session.cmd(file_info_cmd)
logging.info("Output of %s: %s", file_info_cmd, output)
except Exception, err:
raise error.TestFail("Could not create commit_testfile in the "
"overlay file %s" % err)
test.fail("Could not create commit_testfile in the "
"overlay file %s" % err)
vm.destroy()
# Execute the commit command
cmitcmd = "%s commit -f %s %s" % (cmd, image_format,
overlay_file_name)
error.context("Committing image by command %s" % cmitcmd,
logging.info)
error_context.context("Committing image by command %s" % cmitcmd,
logging.info)
try:
utils.system(cmitcmd, verbose=False)
except error.CmdError:
raise error.TestFail("Could not commit the overlay file")
process.system(cmitcmd, verbose=False)
except process.CmdError:
test.fail("Could not commit the overlay file")
logging.info("overlay file (%s) committed!" % overlay_file_name)
msg = "Start a new VM, using image_name as its harddisk"
error.context(msg, logging.info)
error_context.context(msg, logging.info)
params['image_name'] = pre_name
vm_name = params['main_vm']
env_process.preprocess_vm(test, params, env, vm_name)
......@@ -463,8 +462,7 @@ def run(test, params, env):
logging.info("Output of %s: %s", file_exist_chk_cmd, output)
session.cmd(file_del_cmd)
except Exception:
raise error.TestFail("Could not find commit_testfile after a "
"commit")
test.fail("Could not find commit_testfile after a commit")
vm.destroy()
finally:
......@@ -491,11 +489,11 @@ def run(test, params, env):
cmd += " -b %s -F %s %s" % (base_img, backing_fmt, img_name)
msg = "Trying to rebase '%s' to '%s' by command %s" % (img_name,
base_img, cmd)
error.context(msg, logging.info)
error_context.context(msg, logging.info)
if show_progress == "off":
bg = utils.InterruptedThread(send_signal)
bg = utils_misc.InterruptedThread(send_signal)
bg.start()
check_command_output(utils.run(cmd))
check_command_output(process.run(cmd))
def rebase_test(cmd):
"""
......@@ -508,10 +506,10 @@ def run(test, params, env):
:param cmd: qemu-img base command.
"""
if 'rebase' not in utils.system_output(cmd + ' --help',
ignore_status=True):
raise error.TestNAError("Current kvm user space version does not"
" support 'rebase' subcommand")
if 'rebase' not in process.system_output(cmd + ' --help',
ignore_status=True):
test.cancel("Current kvm user space version does not"
" support 'rebase' subcommand")
sn_fmt = params.get("snapshot_format", "qcow2")
sn1 = params["image_name_snapshot1"]
sn1 = _get_image_filename(sn1, enable_gluster, sn_fmt)
......@@ -537,13 +535,13 @@ def run(test, params, env):
actual_base_img = _info(cmd, sn2, "backing file")
base_img_name = os.path.basename(base_img)
if base_img_name not in actual_base_img:
raise error.TestFail("After rebase the backing_file of 'sn2' is "
"'%s' which is not expected as '%s'"
% (actual_base_img, base_img_name))
test.fail("After rebase the backing_file of 'sn2' is "
"'%s' which is not expected as '%s'"
% (actual_base_img, base_img_name))
status, output = _check(cmd, sn2)
if not status:
raise error.TestFail("Check image '%s' failed after rebase;"
"got error: %s" % (sn2, output))
test.fail("Check image '%s' failed after rebase;"
"got error: %s" % (sn2, output))
remove(sn2)
remove(sn1)
......@@ -570,8 +568,8 @@ def run(test, params, env):
cmd += "%s=%s," % (option, params.get(option))
cmd = cmd.rstrip(',')
cmd += " %s" % img_name
error.context(msg, logging.info)
check_command_output(utils.run(cmd, ignore_status=True))
error_context.context(msg, logging.info)
check_command_output(process.run(cmd, ignore_status=True))
def amend_test(cmd):
"""
......@@ -595,11 +593,11 @@ def run(test, params, env):
if actual is not None and actual != expect:
msg = "Get wrong %s from image %s!" % (option, img_name)
msg += "Expect: %s, actual: %s" % (expect, actual)
raise error.TestFail(msg)
test.fail(msg)
status, output = _check(cmd, img)
if not status:
raise error.TestFail("Check image '%s' failed after rebase;"
"got error: %s" % (img, output))
test.fail("Check image '%s' failed after rebase;"
"got error: %s" % (img, output))
def _boot(img_name, img_fmt):
"""
......@@ -615,7 +613,7 @@ def run(test, params, env):
params['image_format'] = img_fmt
image_name = "%s.%s" % (img_name, img_fmt)
msg = "Try to boot vm with image %s" % image_name
error.context(msg, logging.info)
error_context.context(msg, logging.info)
vm_name = params.get("main_vm")
dd_timeout = int(params.get("dd_timeout", 60))
params['vms'] = vm_name
......@@ -630,9 +628,9 @@ def run(test, params, env):
cmd = "dd if=/dev/zero of=/mnt/test bs=1000 count=1000"
status = session.get_command_status(cmd, timeout=dd_timeout)
if status != 0:
raise error.TestError("dd failed")
test.error("dd failed")
error.context("Shutdown guest", logging.info)
error_context.context("Shutdown guest", logging.info)
try:
vm.graceful_shutdown(timeout=login_timeout)
except Exception:
......@@ -643,7 +641,7 @@ def run(test, params, env):
raise
finally:
vm.destroy(gracefully=True)
utils.system("sync")
process.system("sync")
def backup_img_chain(image_file):
"""
......@@ -670,5 +668,5 @@ def run(test, params, env):
# Here starts test
subcommand = params["subcommand"]
error.context("Running %s_test(cmd)" % subcommand, logging.info)
error_context.context("Running %s_test(cmd)" % subcommand, logging.info)
eval("%s_test(cmd)" % subcommand)
qemu/tests/qemu_io.py
浏览文件 @ e41006a0
......@@ -5,9 +5,9 @@ import time
import aexpect
from autotest.client.shared import error
from autotest.client import utils
from avocado.utils import process
from virttest import error_context
from virttest import utils_misc
from virttest import data_dir
......@@ -39,31 +39,31 @@ class QemuIOConfig(object):
except AttributeError:
self.loopback = []
@error.context_aware
@error_context.context_aware
def setup(self):
error.context("performing setup", logging.debug)
error_context.context("performing setup", logging.debug)
utils_misc.display_attributes(self)
# Double check if there aren't any leftovers
self.cleanup()
try:
for f in self.raw_files:
utils.run("%s create -f raw %s 10G" %
(self.qemu_img_binary, f))
process.run("%s create -f raw %s 10G" %
(self.qemu_img_binary, f))
# Associate a loopback device with the raw file.
# Subject to race conditions, that's why try here to associate
# it with the raw file as quickly as possible
l_result = utils.run("losetup -f")
utils.run("losetup -f %s" % f)
l_result = process.run("losetup -f")
process.run("losetup -f %s" % f)
loopback = l_result.stdout.strip()
self.loopback.append(loopback)
# Add the loopback device configured to the list of pvs
# recognized by LVM
utils.run("pvcreate %s" % loopback)
process.run("pvcreate %s" % loopback)
loopbacks = " ".join(self.loopback)
utils.run("vgcreate %s %s" % (self.vgtest_name, loopbacks))
process.run("vgcreate %s %s" % (self.vgtest_name, loopbacks))
# Create an lv inside the vg with starting size of 200M
utils.run("lvcreate -L 19G -n %s %s" %
(self.lvtest_name, self.vgtest_name))
process.run("lvcreate -L 19G -n %s %s" %
(self.lvtest_name, self.vgtest_name))
except Exception:
try:
self.cleanup()
......@@ -71,30 +71,30 @@ class QemuIOConfig(object):
logging.warn(e)
raise
@error.context_aware
@error_context.context_aware
def cleanup(self):
error.context("performing qemu_io cleanup", logging.debug)
error_context.context("performing qemu_io cleanup", logging.debug)
if os.path.isfile(self.lvtest_device):
utils.run("fuser -k %s" % self.lvtest_device)
process.run("fuser -k %s" % self.lvtest_device)
time.sleep(2)
l_result = utils.run("lvdisplay")
l_result = process.run("lvdisplay")
# Let's remove all volumes inside the volume group created
if self.lvtest_name in l_result.stdout:
utils.run("lvremove -f %s" % self.lvtest_device)
process.run("lvremove -f %s" % self.lvtest_device)
# Now, removing the volume group itself
v_result = utils.run("vgdisplay")
v_result = process.run("vgdisplay")
if self.vgtest_name in v_result.stdout:
utils.run("vgremove -f %s" % self.vgtest_name)
process.run("vgremove -f %s" % self.vgtest_name)
# Now, if we can, let's remove the physical volume from lvm list
p_result = utils.run("pvdisplay")
l_result = utils.run('losetup -a')
p_result = process.run("pvdisplay")
l_result = process.run('losetup -a')
for l in self.loopback:
if l in p_result.stdout:
utils.run("pvremove -f %s" % l)
process.run("pvremove -f %s" % l)
if l in l_result.stdout:
try:
utils.run("losetup -d %s" % l)
except error.CmdError, e:
process.run("losetup -d %s" % l)
except process.CmdError, e:
logging.error("Failed to liberate loopback %s, "
"error msg: '%s'", l, e)
......@@ -145,7 +145,7 @@ def run(test, params, env):
for err_type in err_string.keys():
msg = re.findall(err_string.get(err_type), test_result)
if msg:
raise error.TestFail(msg)
test.fail(msg)
finally:
try:
if qemu_io_config:
......
qemu/tests/qemu_io_blkdebug.py
浏览文件 @ e41006a0
......@@ -3,16 +3,16 @@ import re
import logging
import ConfigParser
from autotest.client.shared import error
from avocado.utils import process
from virttest import error_context
from virttest import qemu_io
from virttest import data_dir
from virttest import utils_misc
from virttest.qemu_storage import QemuImg
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
Run qemu-io blkdebug tests:
......@@ -44,7 +44,7 @@ def run(test, params, env):
pre_snapshot = params.get("pre_snapshot", "no") == "yes"
del_snapshot = params.get("del_snapshot", "no") == "yes"
error.context("Create image", logging.info)
error_context.context("Create image", logging.info)
image_io = QemuImg(
params.object_params(image), data_dir.get_data_dir(), image)
image_name, _ = image_io.create(params.object_params(image))
......@@ -56,11 +56,11 @@ def run(test, params, env):
for errn in errn_list:
log_filename = utils_misc.get_path(test.outputdir,
"qemu-io-log-%s" % errn)
error.context("Write the blkdebug config file", logging.info)
error_context.context("Write the blkdebug config file", logging.info)
template.set("inject-error", "event", '"%s"' % err_event)
template.set("inject-error", "errno", '"%s"' % errn)
error.context("Write blkdebug config file", logging.info)
error_context.context("Write blkdebug config file", logging.info)
blkdebug = None
try:
blkdebug = open(blkdebug_cfg, 'w')
......@@ -69,12 +69,13 @@ def run(test, params, env):
if blkdebug is not None:
blkdebug.close()
error.context("Create image", logging.info)
error_context.context("Create image", logging.info)
image_io = QemuImg(params.object_params(
image), data_dir.get_data_dir(), image)
image_name = image_io.create(params.object_params(image))[0]
error.context("Operate in qemu-io to trigger the error", logging.info)
error_context.context("Operate in qemu-io to trigger the error",
logging.info)
session = qemu_io.QemuIOShellSession(test, params, image_name,
blkdebug_cfg=blkdebug_cfg,
log_filename=log_filename)
......@@ -109,19 +110,19 @@ def run(test, params, env):
params_sn, data_dir.get_data_dir(), image_sn)
image_snapshot.remove()
error.context("Get error message", logging.info)
error_context.context("Get error message", logging.info)
try:
std_msg = os.strerror(int(errn))
except ValueError:
raise error.TestError("Can not find error message:\n"
" error code is %s" % errn)
test.error("Can not find error message:\n"
" error code is %s" % errn)
session.close()
error.context("Compare the error message", logging.info)
error_context.context("Compare the error message", logging.info)
if std_msg in output:
logging.info("Error message is correct in qemu-io")
else:
fail_log = "The error message is mismatch:\n"
fail_log += " qemu-io reports: '%s',\n" % output
fail_log += " os.strerror reports: '%s'" % std_msg
raise error.TestFail(fail_log)
test.fail(fail_log)
qemu/tests/qemu_iotests.py
浏览文件 @ e41006a0
import os
from autotest.client.shared import git
from autotest.client.shared import error
from autotest.client import utils
from avocado.utils import git
from avocado.utils import process
from virttest import error_context
from virttest import utils_misc
@error.context_aware
@error_context.context_aware
def run(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
4) Report any errors found to test
:param test: QEMU test object.
:param params: Dictionary with the test parameters.
......@@ -46,5 +46,6 @@ def run(test, params, env):
if extra_options:
cmd += extra_options
error.context("running qemu-iotests for image format %s" % image_format)
utils.system("%s -%s" % (cmd, image_format))
error_context.context("running qemu-iotests for image format %s"
% image_format)
process.system("%s -%s" % (cmd, image_format))
qemu/tests/qemu_killer_report.py
浏览文件 @ e41006a0
......@@ -2,11 +2,11 @@ import logging
import os
import re
from autotest.client.shared import error
from virttest import error_context
from virttest import utils_misc
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
Test that QEMU report the process ID that sent it kill signals.
......@@ -41,18 +41,19 @@ def run(test, params, env):
re_str = "terminating on signal 15 from pid ([0-9]+)"
re_str = params.get("qemu_error_re", re_str)
error.context("Kill VM by signal 15", logging.info)
error_context.context("Kill VM by signal 15", logging.info)
thread_pid = kill_vm_by_signal_15()
# Wait QEMU print error log.
results = utils_misc.wait_for(lambda: killer_report(re_str),
60, 2, 2)
error.context("Check that QEMU can report who killed it", logging.info)
error_context.context("Check that QEMU can report who killed it",
logging.info)
if not results:
raise error.TestFail("QEMU did not tell us who killed it")
test.fail("QEMU did not tell us who killed it")
elif int(results[-1]) != thread_pid:
msg = "QEMU identified the process that killed it incorrectly. "
msg += "Killer PID: %s, " % thread_pid
msg += "QEMU reported PID: %s" % int(results[-1])
raise error.TestFail(msg)
test.fail(msg)
else:
logging.info("QEMU identified the process that killed it properly")
qemu/tests/qemu_no_shutdown.py
浏览文件 @ e41006a0
import logging
from autotest.client.shared import error
from virttest import error_context
from virttest import utils_misc
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
KVM -no-shutdown flag test:
......@@ -23,7 +23,7 @@ def run(test, params, env):
timeout = int(params.get("login_timeout", 360))
repeat_times = int(params.get("repeat_times", 5))
error.base_context("Qemu -no-shutdown test")
error_context.base_context("Qemu -no-shutdown test")
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
......@@ -32,33 +32,33 @@ def run(test, params, env):
logging.info("The guest bootup successfully.")
for i in xrange(repeat_times):
error.context("Round %s : Send monitor cmd system_powerdown."
% str(i + 1), logging.info)
error_context.context("Round %s : Send monitor cmd system_powerdown."
% str(i + 1), logging.info)
# Send a system_powerdown monitor command
vm.monitor.system_powerdown()
# Wait for the session to become unresponsive and close it
if not utils_misc.wait_for(lambda: not session.is_responsive(),
timeout, 0, 1):
raise error.TestFail("Oops, Guest refuses to go down!")
test.fail("Oops, Guest refuses to go down!")
if session:
session.close()
# Check the qemu id is not change
if not utils_misc.wait_for(lambda: vm.is_alive(), 5, 0, 1):
raise error.TestFail("VM not responsive after system_"
"powerdown with -no-shutdown!")
test.fail("VM not responsive after system_powerdown "
"with -no-shutdown!")
if vm.get_pid() != qemu_process_id:
raise error.TestFail("Qemu pid changed after system_powerdown!")
test.fail("Qemu pid changed after system_powerdown!")
logging.info("Round %s -> System_powerdown successfully.", str(i + 1))
# Send monitor command system_reset and cont
error.context("Round %s : Send monitor command system_reset and cont."
% str(i + 1), logging.info)
error_context.context("Round %s : Send monitor command system_reset "
"and cont." % str(i + 1), logging.info)
vm.monitor.cmd("system_reset")
vm.resume()
session = vm.wait_for_login(timeout=timeout)
logging.info("Round %s -> Guest is up successfully." % str(i + 1))
if vm.get_pid() != qemu_process_id:
raise error.TestFail("Qemu pid changed after system_reset & cont!")
test.fail("Qemu pid changed after system_reset & cont!")
if session:
session.close()
qemu/tests/qemu_nobody.py
浏览文件 @ e41006a0
import logging
import re
from autotest.client.shared import utils
from autotest.client.shared import error
from avocado.utils import process
from virttest import env_process
from virttest import error_context
@error.context_aware
@env_process.context_aware
def run(test, params, env):
"""
Check smbios table :
......@@ -22,8 +22,8 @@ def run(test, params, env):
"""
return user uid and gid as a list
"""
user_uid = utils.system_output("id -u %s" % username).split()
user_gid = utils.system_output("id -g %s" % username).split()
user_uid = process.system_output("id -u %s" % username).split()
user_gid = process.system_output("id -g %s" % username).split()
return(user_uid, user_gid)
def get_ugid_from_processid(pid):
......@@ -31,22 +31,21 @@ def run(test, params, env):
return a list[uid,euid,suid,fsuid,gid,egid,sgid,fsgid] of pid
"""
grep_ugid_cmd = "cat /proc/%s/status | grep -iE '^(U|G)id'"
o = utils.system_output(grep_ugid_cmd % pid.strip())
o = process.system_output(grep_ugid_cmd % pid.strip())
ugid = re.findall("(\d+)", o)
# real UID, effective UID, saved set UID, and file system UID
if ugid:
return ugid
else:
raise error.TestError("Could not find the correct UID for process %s"
% pid)
test.error("Could not find the correct UID for process %s" % pid)
exec_username = params.get("user_runas", "nobody")
error.base_context("Run QEMU %s test:" % exec_username)
error.context("Get the user uid and gid,using 'id -u/g username'")
env_process.base_context("Run QEMU %s test:" % exec_username)
error_context.context("Get the user uid and gid,using 'id -u/g username'")
(exec_uid, exec_gid) = get_user_ugid(exec_username)
error.context("Run the qemu as user '%s'" % exec_username)
error_context.context("Run the qemu as user '%s'" % exec_username)
logging.info("The user %s :uid='%s', gid='%s'" %
(exec_username, exec_uid, exec_gid))
......@@ -56,17 +55,17 @@ def run(test, params, env):
vm = env.get_vm(params["main_vm"])
failures = []
for pid in utils.get_children_pids(vm.get_shell_pid()):
error.context("Get the process '%s' u/gid, using 'cat /proc/%s/status'"
% (pid, pid), logging.info)
for pid in process.get_children_pids(vm.get_shell_pid()):
error_context.context("Get the process '%s' u/gid, using 'cat "
"/proc/%s/status'" % (pid, pid), logging.info)
qemu_ugid = get_ugid_from_processid(pid)
logging.info("Process run as uid=%s,euid=%s,suid=%s,fsuid=%s"
% tuple(qemu_ugid[0:4]))
logging.info("Process run as gid=%s,egid=%s,sgid=%s,fsgid=%s"
% tuple(qemu_ugid[4:]))
error.context("Check if the user %s ugid is equal to the process %s"
% (exec_username, pid))
error_context.context("Check if the user %s ugid is equal to the "
"process %s" % (exec_username, pid))
# generate user uid, euid, suid, fsuid, gid, egid, sgid, fsgid
user_ugid_extend = exec_uid * 4 + exec_gid * 4
if cmp(user_ugid_extend, qemu_ugid) != 0:
......@@ -75,5 +74,5 @@ def run(test, params, env):
failures.append(e_msg)
if failures:
raise error.TestFail("FAIL: Test reported %s failures:\n%s" %
(len(failures), "\n".join(failures)))
test.fail("FAIL: Test reported %s failures:\n%s" %
(len(failures), "\n".join(failures)))
qemu/tests/qemu_option_check.py
浏览文件 @ e41006a0
import re
import logging
from autotest.client import utils
from autotest.client.shared import error
from avocado.utils import process
from virttest import error_context
from virttest import utils_misc
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
QEMU support options check test
......@@ -24,11 +24,11 @@ def run(test, params, env):
"""
Get qemu support device list
"""
support_device = utils.system_output("%s -device ? 2>&1"
% qemu_binary, timeout=10,
ignore_status=True)
support_device = process.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")
test.cancel("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)
......@@ -41,22 +41,22 @@ def run(test, params, env):
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)
test.cancel(err_msg % device_name)
device_support_option = process.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")
test.error("Oops, output status is wrong")
if not re.findall(r"%s\.(.*)=(.*)" % device_name,
device_support_option.stdout):
raise error.TestFail("Qemu option check Failed")
test.fail("Qemu option check Failed")
logging.info("Qemu options check successful. 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)
error_context.context("Get qemu support %s device options" % device_name,
logging.info)
get_device_option(qemu_binary, device_name)
qemu/tests/qmp_basic.py
浏览文件 @ e41006a0
from autotest.client.shared import error
def run(test, params, env):
"""
QMP Specification test-suite: this checks if the *basic* protocol conforms
......@@ -31,11 +28,10 @@ def run(test, params, env):
"""
def fail_no_key(qmp_dict, key):
if not isinstance(qmp_dict, dict):
raise error.TestFail("qmp_dict is not a dict (it's '%s')" %
type(qmp_dict))
test.fail("qmp_dict is not a dict (it's '%s')" % type(qmp_dict))
if key not in qmp_dict:
raise error.TestFail("'%s' key doesn't exist in dict ('%s')" %
(key, str(qmp_dict)))
test.fail("'%s' key doesn't exist in dict ('%s')" %
(key, str(qmp_dict)))
def check_dict_key(qmp_dict, key, keytype):
"""
......@@ -49,8 +45,8 @@ def run(test, params, env):
"""
fail_no_key(qmp_dict, key)
if not isinstance(qmp_dict[key], keytype):
raise error.TestFail("'%s' key is not of type '%s', it's '%s'" %
(key, keytype, type(qmp_dict[key])))
test.fail("'%s' key is not of type '%s', it's '%s'" %
(key, keytype, type(qmp_dict[key])))
def check_key_is_dict(qmp_dict, key):
check_dict_key(qmp_dict, key, dict)
......@@ -64,22 +60,22 @@ def run(test, params, env):
def check_str_key(qmp_dict, keyname, value=None):
check_dict_key(qmp_dict, keyname, unicode)
if value and value != qmp_dict[keyname]:
raise error.TestFail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
test.fail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
def check_key_is_int(qmp_dict, key):
fail_no_key(qmp_dict, key)
try:
int(qmp_dict[key])
except Exception:
raise error.TestFail("'%s' key is not of type int, it's '%s'" %
(key, type(qmp_dict[key])))
test.fail("'%s' key is not of type int, it's '%s'" %
(key, type(qmp_dict[key])))
def check_bool_key(qmp_dict, keyname, value=None):
check_dict_key(qmp_dict, keyname, bool)
if value and value != qmp_dict[keyname]:
raise error.TestFail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
test.fail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
def check_success_resp(resp, empty=False):
"""
......@@ -90,8 +86,7 @@ def run(test, params, env):
"""
check_key_is_dict(resp, "return")
if empty and len(resp["return"]) > 0:
raise error.TestFail("success response is not empty ('%s')" %
str(resp))
test.fail("success response is not empty ('%s')" % str(resp))
def check_error_resp(resp, classname=None, datadict=None):
"""
......@@ -104,8 +99,8 @@ def run(test, params, env):
check_key_is_dict(resp, "error")
check_key_is_str(resp["error"], "class")
if classname and resp["error"]["class"] != classname:
raise error.TestFail("got error class '%s' expected '%s'" %
(resp["error"]["class"], classname))
test.fail("got error class '%s' expected '%s'" %
(resp["error"]["class"], classname))
def test_version(version):
"""
......@@ -195,8 +190,8 @@ def run(test, params, env):
resp = monitor.cmd_qmp("query-status", q_id=id_key)
check_success_resp(resp)
if resp["id"] != id_key:
raise error.TestFail("expected id '%s' but got '%s'" %
(str(id_key), str(resp["id"])))
test.fail("expected id '%s' but got '%s'" %
(str(id_key), str(resp["id"])))
def test_invalid_arg_key(monitor):
"""
......@@ -360,7 +355,7 @@ def run(test, params, env):
if qmp_monitor:
qmp_monitor = qmp_monitor[0]
else:
raise error.TestError('Could not find a QMP monitor, aborting test')
test.error('Could not find a QMP monitor, aborting test')
# Run all suites
greeting_suite(qmp_monitor)
......@@ -371,4 +366,4 @@ def run(test, params, env):
# check if QMP is still alive
if not qmp_monitor.is_responsive():
raise error.TestFail('QMP monitor is not responsive after testing')
test.fail('QMP monitor is not responsive after testing')
qemu/tests/qmp_basic_rhel6.py
浏览文件 @ e41006a0
import logging
from autotest.client.shared import error
def run(test, params, env):
"""
......@@ -33,11 +31,10 @@ def run(test, params, env):
"""
def fail_no_key(qmp_dict, key):
if not isinstance(qmp_dict, dict):
raise error.TestFail("qmp_dict is not a dict (it's '%s')" %
type(qmp_dict))
test.fail("qmp_dict is not a dict (it's '%s')" % type(qmp_dict))
if key not in qmp_dict:
raise error.TestFail("'%s' key doesn't exist in dict ('%s')" %
(key, str(qmp_dict)))
test.fail("'%s' key doesn't exist in dict ('%s')" %
(key, str(qmp_dict)))
def check_dict_key(qmp_dict, key, keytype):
"""
......@@ -51,8 +48,8 @@ def run(test, params, env):
"""
fail_no_key(qmp_dict, key)
if not isinstance(qmp_dict[key], keytype):
raise error.TestFail("'%s' key is not of type '%s', it's '%s'" %
(key, keytype, type(qmp_dict[key])))
test.fail("'%s' key is not of type '%s', it's '%s'" %
(key, keytype, type(qmp_dict[key])))
def check_key_is_dict(qmp_dict, key):
check_dict_key(qmp_dict, key, dict)
......@@ -66,22 +63,22 @@ def run(test, params, env):
def check_str_key(qmp_dict, keyname, value=None):
check_dict_key(qmp_dict, keyname, unicode)
if value and value != qmp_dict[keyname]:
raise error.TestFail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
test.fail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
def check_key_is_int(qmp_dict, key):
fail_no_key(qmp_dict, key)
try:
int(qmp_dict[key])
except Exception:
raise error.TestFail("'%s' key is not of type int, it's '%s'" %
(key, type(qmp_dict[key])))
test.fail("'%s' key is not of type int, it's '%s'" %
(key, type(qmp_dict[key])))
def check_bool_key(qmp_dict, keyname, value=None):
check_dict_key(qmp_dict, keyname, bool)
if value and value != qmp_dict[keyname]:
raise error.TestFail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
test.fail("'%s' key value '%s' should be '%s'" %
(keyname, str(qmp_dict[keyname]), str(value)))
def check_success_resp(resp, empty=False):
"""
......@@ -92,8 +89,7 @@ def run(test, params, env):
"""
check_key_is_dict(resp, "return")
if empty and len(resp["return"]) > 0:
raise error.TestFail("success response is not empty ('%s')" %
str(resp))
test.fail("success response is not empty ('%s')" % str(resp))
def check_error_resp(resp, classname=None, datadict=None):
"""
......@@ -107,12 +103,12 @@ def run(test, params, env):
check_key_is_dict(resp, "error")
check_key_is_str(resp["error"], "class")
if classname and resp["error"]["class"] != classname:
raise error.TestFail("got error class '%s' expected '%s'" %
(resp["error"]["class"], classname))
test.fail("got error class '%s' expected '%s'" %
(resp["error"]["class"], classname))
check_key_is_dict(resp["error"], "data")
if datadict and resp["error"]["data"] != datadict:
raise error.TestFail("got data dict '%s' expected '%s'" %
(resp["error"]["data"], datadict))
test.fail("got data dict '%s' expected '%s'" %
(resp["error"]["data"], datadict))
def test_version(version):
"""
......@@ -200,8 +196,8 @@ def run(test, params, env):
resp = monitor.cmd_qmp("query-status", q_id=id_key)
check_success_resp(resp)
if resp["id"] != id_key:
raise error.TestFail("expected id '%s' but got '%s'" %
(str(id_key), str(resp["id"])))
test.fail("expected id '%s' but got '%s'" %
(str(id_key), str(resp["id"])))
def test_invalid_arg_key(monitor):
"""
......@@ -348,7 +344,7 @@ def run(test, params, env):
if qmp_monitor:
qmp_monitor = qmp_monitor[0]
else:
raise error.TestError('Could not find a QMP monitor, aborting test')
test.error('Could not find a QMP monitor, aborting test')
# Run all suites
greeting_suite(qmp_monitor)
......@@ -359,4 +355,4 @@ def run(test, params, env):
# check if QMP is still alive
if not qmp_monitor.is_responsive():
raise error.TestFail('QMP monitor is not responsive after testing')
test.fail('QMP monitor is not responsive after testing')
qemu/tests/qmp_command.py
浏览文件 @ e41006a0
import logging
import re
from autotest.client.shared import error
from avocado.core import exceptions
from avocado.utils import process
from virttest import utils_misc
......@@ -45,29 +42,27 @@ def run(test, params, env):
if result_check == "equal":
value = output
if value != str(qmp_o):
raise exceptions.TestFail("QMP command return value does not match "
"the expect result. Expect result: '%s'\n"
"Actual result: '%s'" % (value, qmp_o))
test.fail("QMP command return value does not match "
"the expect result. Expect result: '%s'\n"
"Actual result: '%s'" % (value, qmp_o))
elif result_check == "contain":
values = output.split(';')
for value in values:
if value in exception_list:
continue
if value.strip() not in str(qmp_o):
raise exceptions.TestFail("QMP command output does not contain "
"expect result. Expect result: '%s'\n"
"Actual result: '%s'"
% (value, qmp_o))
test.fail("QMP command output does not contain "
"expect result. Expect result: '%s'\n"
"Actual result: '%s'" % (value, qmp_o))
elif result_check == "not_contain":
values = output.split(';')
for value in values:
if value in exception_list:
continue
if value in str(qmp_o):
raise exceptions.TestFail("QMP command output contains unexpect"
" result. Unexpect result: '%s'\n"
"Actual result: '%s'"
% (value, qmp_o))
test.fail("QMP command output contains unexpect"
" result. Unexpect result: '%s'\n"
"Actual result: '%s'" % (value, qmp_o))
elif result_check == "m_equal_q":
msg = "QMP command ouput is not equal to in human monitor command."
msg += "\nQMP command output: '%s'" % qmp_o
......@@ -79,9 +74,9 @@ def run(test, params, env):
len_o = len(qmp_o)
if len(res) != len_o:
if res[0].startswith(' '):
raise exceptions.TestFail("Human command starts with ' ', "
"there is probably some garbage in "
"the output.\n" + msg)
test.fail("Human command starts with ' ', "
"there is probably some garbage in "
"the output.\n" + msg)
res_tmp = []
#(qemu)info block in RHEL7 divided into 3 lines
for line in res:
......@@ -91,7 +86,7 @@ def run(test, params, env):
res_tmp[-1] += line
res = res_tmp
if len(res) != len_o:
raise exceptions.TestFail(msg)
test.fail(msg)
re_str = r'([^ \t\n\r\f\v=]*)=([^ \t\n\r\f\v=]*)'
for i in range(len(res)):
if qmp_cmd == "query-version":
......@@ -107,7 +102,7 @@ def run(test, params, env):
package = package.strip()
hmp_version = hmp_version.strip()
if version != hmp_version or package != hmp_package:
raise exceptions.TestFail(msg)
test.fail(msg)
else:
matches = re.findall(re_str, res[i])
for key, val in matches:
......@@ -129,7 +124,7 @@ def run(test, params, env):
if value != qmp_o[i][key]:
msg += "\nValue in human monitor: '%s'" % value
msg += "\nValue in qmp: '%s'" % qmp_o[i][key]
raise exceptions.TestFail(msg)
test.fail(msg)
elif qmp_cmd == "query-block":
cmp_str = "u'%s': u'%s'" % (key, val)
cmp_s = "u'%s': %s" % (key, val)
......@@ -145,19 +140,19 @@ def run(test, params, env):
msg += ("\nCan not find '%s', '%s' or '%s' in "
" QMP command output."
% (cmp_s, cmp_str_b, cmp_str))
raise exceptions.TestFail(msg)
test.fail(msg)
elif qmp_cmd == "query-balloon":
if (int(val) * 1024 * 1024 != qmp_o[key] and
val not in str(qmp_o[key])):
msg += ("\n'%s' is not in QMP command output"
% val)
raise exceptions.TestFail(msg)
test.fail(msg)
else:
if (val not in str(qmp_o[i][key]) and
str(bool(int(val))) not in str(qmp_o[i][key])):
msg += ("\n'%s' is not in QMP command output"
% val)
raise exceptions.TestFail(msg)
test.fail(msg)
elif result_check == "m_in_q":
res = output.splitlines(True)
msg = "Key value from human monitor command is not in"
......@@ -174,21 +169,21 @@ def run(test, params, env):
str_o = qmp_o
if param.rstrip() not in str(str_o):
msg += "\nKey value is '%s'" % param.rstrip()
raise error.TestFail(msg)
test.fail(msg)
elif result_check == "m_format_q":
match_flag = True
for i in qmp_o:
if output is None:
raise exceptions.TestError("QMP output pattern is missing")
test.error("QMP output pattern is missing")
if re.match(output.strip(), str(i)) is None:
match_flag = False
if not match_flag:
msg = "Output does not match the pattern: '%s'" % output
raise exceptions.TestFail(msg)
test.fail(msg)
qemu_binary = utils_misc.get_qemu_binary(params)
if not utils_misc.qemu_has_option("qmp", qemu_binary):
raise exceptions.TestSkipError("Host qemu does not support qmp.")
test.cancel("Host qemu does not support qmp.")
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
......@@ -204,12 +199,12 @@ def run(test, params, env):
if qmp_ports:
qmp_port = qmp_ports[0]
else:
raise exceptions.TestError("Incorrect configuration, no QMP monitor found.")
test.error("Incorrect configuration, no QMP monitor found.")
hmp_ports = vm.get_monitors_by_type('human')
if hmp_ports:
hmp_port = hmp_ports[0]
else:
raise exceptions.TestError("Incorrect configuration, no QMP monitor found.")
test.error("Incorrect configuration, no QMP monitor found.")
callback = {"host_cmd": lambda cmd: process.system_output(cmd, shell=True),
"guest_cmd": session.get_command_output,
"monitor_cmd": hmp_port.send_args_cmd,
......@@ -220,7 +215,7 @@ def run(test, params, env):
if cmd_type in callback.keys():
return callback[cmd_type](cmd)
else:
raise exceptions.TestError("cmd_type is not supported")
test.error("cmd_type is not supported")
pre_cmd = params.get("pre_cmd")
qmp_cmd = params.get("qmp_cmd")
......@@ -247,14 +242,14 @@ def run(test, params, env):
if params.get("negative_check_pattern"):
check_pattern = params.get("negative_check_pattern")
if check_pattern not in str(err):
raise exceptions.TestFail("'%s' not in exception '%s'"
% (check_pattern, err))
test.fail("'%s' not in exception '%s'"
% (check_pattern, err))
else:
raise exceptions.TestFail(err)
test.fail(err)
except qemu_monitor.MonitorProtocolError, err:
raise exceptions.TestFail(err)
test.fail(err)
except Exception, err:
raise exceptions.TestFail(err)
test.fail(err)
# Post command
if post_cmd is not None:
......
qemu/tests/qmp_event_notification.py
浏览文件 @ e41006a0
......@@ -2,8 +2,6 @@ import logging
import time
import commands
from autotest.client.shared import error
from virttest import utils_misc
......@@ -21,8 +19,8 @@ def run(test, params, env):
qemu_binary = utils_misc.get_qemu_binary(params)
if not utils_misc.qemu_has_option("qmp", qemu_binary):
error.TestNAError("This test case requires a host QEMU with QMP "
"monitor support")
test.cancel("This test case requires a host QEMU with QMP "
"monitor support")
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
......@@ -38,7 +36,7 @@ def run(test, params, env):
if cmd_type in callback.keys():
return callback[cmd_type](cmd, **options)
else:
raise error.TestError("cmd_type is not supported")
test.error("cmd_type is not supported")
cmd_type = params["event_cmd_type"]
pre_event_cmd = params.get("pre_event_cmd", "")
......@@ -86,8 +84,7 @@ def run(test, params, env):
break
if qmp_num > 0:
raise error.TestFail("Did not receive qmp %s event notification"
% event_check)
test.fail("Did not receive qmp %s event notification" % event_check)
if post_event_cmd:
send_cmd(post_event_cmd, post_event_cmd_type,
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或