Skip to content

T
tp-qemu

openeuler / tp-qemu

通知 1
Star 0
Fork 0
T
tp-qemu

体验新版 GitCode，发现更多精彩内容 >>

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

[qemu] Replace autotest modules - kl 

Signed-off-by: NXu Han <xuhan@redhat.com>
上级 c1b0252f
隐藏空白更改
内联 并排
Showing with 149 addition and 152 deletion
qemu/tests/kdump_with_stress.py
浏览文件 @ 36b98db0
import logging
import os
from autotest.client.shared import error
from autotest.client import utils
from avocado.utils import download
from virttest import error_context
from virttest import utils_test
from virttest import utils_misc
from generic.tests import kdump
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
KVM kdump test with stress:
......@@ -37,13 +38,15 @@ def run(test, params, env):
install_cmd = params.get("install_cmd")
logging.info("Fetch package: '%s'" % link)
pkg = utils.unmap_url_cache(test.tmpdir, link, md5sum)
vm.copy_files_to(pkg, tmp_dir)
pkg_name = os.path.basename(link)
pkg_path = os.path.join(test.tmpdir, pkg_name)
download.get_file(link, pkg_path, hash_expected=md5sum)
vm.copy_files_to(pkg_path, tmp_dir)
logging.info("Install app: '%s' in guest." % install_cmd)
s, o = session.cmd_status_output(install_cmd, timeout=300)
if s != 0:
raise error.TestError("Fail to install stress app(%s)" % o)
test.error("Fail to install stress app(%s)" % o)
logging.info("Install app successed")
......@@ -51,7 +54,7 @@ def run(test, params, env):
"""
Load stress in guest.
"""
error.context("Load stress in guest", logging.info)
error_context.context("Load stress in guest", logging.info)
stress_type = params.get("stress_type", "none")
if stress_type == "none":
......@@ -61,9 +64,9 @@ def run(test, params, env):
bg = ""
bg_stress_test = params.get("run_bgstress")
bg = utils.InterruptedThread(utils_test.run_virt_sub_test,
(test, params, env),
{"sub_type": bg_stress_test})
bg = utils_misc.InterruptedThread(utils_test.run_virt_sub_test,
(test, params, env),
{"sub_type": bg_stress_test})
bg.start()
if stress_type == "io":
......@@ -76,7 +79,7 @@ def run(test, params, env):
running = utils_misc.wait_for(lambda: stress_running(session),
timeout=150, step=5)
if not running:
raise error.TestError("Stress isn't running")
test.error("Stress isn't running")
logging.info("Stress running now")
......@@ -109,8 +112,8 @@ def run(test, params, env):
try:
start_stress(session)
error.context("Kdump Testing, force the Linux kernel to crash",
logging.info)
error_context.context("Kdump Testing, force the Linux kernel to crash",
logging.info)
crash_cmd = params.get("crash_cmd", "echo c > /proc/sysrq-trigger")
if crash_cmd == "nmi":
kdump.crash_test(vm, None, crash_cmd, timeout)
......
qemu/tests/kernbench.py
浏览文件 @ 36b98db0
......@@ -3,8 +3,8 @@ import commands
import os
import re
from autotest.client import utils
from autotest.client.shared import error
from avocado.utils import cpu
from avocado.utils import process
from virttest import env_process
......@@ -24,7 +24,7 @@ def run(test, params, env):
def download_if_not_exists():
if not os.path.exists(file_name):
cmd = "wget -t 10 -c -P %s %s" % (tmp_dir, file_link)
utils.system(cmd)
process.system(cmd)
def cmd_status_output(cmd, timeout=360):
s = 0
......@@ -36,18 +36,17 @@ def run(test, params, env):
return (s, o)
def check_ept():
output = utils.system_output("grep 'flags' /proc/cpuinfo")
output = process.system_output("grep 'flags' /proc/cpuinfo")
flags = output.splitlines()[0].split(':')[1].split()
need_ept = params.get("need_ept", "no")
if 'ept' not in flags and "yes" in need_ept:
raise error.TestNAError(
"This test requires a host that supports EPT")
test.cancel("This test requires a host that supports EPT")
elif 'ept' in flags and "no" in need_ept:
cmd = "modprobe -r kvm_intel && modprobe kvm_intel ept=0"
utils.system(cmd, timeout=100)
process.system(cmd, timeout=100)
elif 'ept' in flags and "yes" in need_ept:
cmd = "modprobe -r kvm_intel && modprobe kvm_intel ept=1"
utils.system(cmd, timeout=100)
process.system(cmd, timeout=100)
def install_gcc():
logging.info("Update gcc to request version....")
......@@ -105,7 +104,7 @@ def run(test, params, env):
# Create tmp folder and download files if need.
if not os.path.exists(tmp_dir):
utils.system("mkdir %s" % tmp_dir)
process.system("mkdir %s" % tmp_dir)
files = params.get("files_need").split()
for file in files:
file_link = params.get("%s_link" % file)
......@@ -123,18 +122,17 @@ def run(test, params, env):
pre_cmd = params.get("pre_cmd")
(s, o) = cmd_status_output(pre_cmd, timeout=cmd_timeout)
if s:
raise error.TestError("Fail command:%s\nOutput: %s" % (pre_cmd, o))
test.error("Fail command:%s\nOutput: %s" % (pre_cmd, o))
if "guest" in test_type:
cpu_num = params.get("smp")
else:
cpu_num = utils.count_cpus()
cpu_num = cpu.online_cpus_count()
test_cmd = params.get("test_cmd") % (int(cpu_num) * cpu_multiplier)
logging.info("Start making the kernel ....")
(s, o) = cmd_status_output(test_cmd, timeout=cmd_timeout)
if s:
raise error.TestError(
"Fail command:%s\n Output:%s" % (test_cmd, o))
test.error("Fail command:%s\n Output:%s" % (test_cmd, o))
else:
logging.info("Output for command %s is:\n %s" % (test_cmd, o))
record_result(o)
......
qemu/tests/kernel_install.py
浏览文件 @ 36b98db0
import logging
import os
from autotest.client.shared import error
from autotest.client import utils
from avocado.utils import aurl
from avocado.utils import download
from virttest import error_context
from virttest import utils_test
from virttest import data_dir
CLIENT_TEST = "kernelinstall"
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
KVM kernel install test:
......@@ -32,11 +33,11 @@ def run(test, params, env):
_tmp_params_dict = {}
def _copy_file_to_test_dir(file_path):
if utils.is_url(file_path):
if aurl.is_url(file_path):
return file_path
file_abs_path = os.path.join(test.bindir, file_path)
dest = os.path.join(sub_test_path, os.path.basename(file_abs_path))
return os.path.basename(utils.get_file(file_path, dest))
return os.path.basename(download.get_file(file_path, dest))
def _save_bootloader_config(session):
"""
......@@ -51,7 +52,7 @@ def run(test, params, env):
return default_kernel
def _restore_bootloader_config(session, default_kernel):
error.context("Restore the grub to old version")
error_context.context("Restore the grub to old version")
if not default_kernel:
logging.warn("Could not get previous grub config, do noting.")
......@@ -61,7 +62,7 @@ def run(test, params, env):
try:
session.cmd(cmd)
except Exception, e:
raise error.TestWarn("Restore grub failed: '%s'" % e)
test.error("Restore grub failed: '%s'" % e)
def _clean_up_tmp_files(file_list):
for f in file_list:
......@@ -80,7 +81,7 @@ def run(test, params, env):
return {param_str: param}
return {param_str: default_value}
error.context("Log into a guest")
error_context.context("Log into a guest")
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
timeout = float(params.get("login_timeout", 240))
......@@ -89,7 +90,7 @@ def run(test, params, env):
logging.info("Guest kernel before install: %s",
session.cmd('uname -a').strip())
error.context("Save current default kernel information")
error_context.context("Save current default kernel information")
default_kernel = _save_bootloader_config(session)
# Check if there is local file in params, move local file to
......@@ -128,7 +129,7 @@ def run(test, params, env):
tag = params.get('kernel_tag')
error.context("Generate control file for kernel install test")
error_context.context("Generate control file for kernel install test")
# Generate control file from parameters
control_base = "params = %s\n"
control_base += "job.run_test('kernelinstall'"
......@@ -149,27 +150,26 @@ def run(test, params, env):
_tmp_file_list.append(os.path.abspath(test_control_path))
except IOError, e:
_clean_up_tmp_files(_tmp_file_list)
raise error.TestError("Fail to Generate control file,"
" error message:\n '%s'" % e)
test.error("Fail to Generate control file, error message:\n '%s'" % e)
params["test_control_file_install"] = test_control_file
error.context("Launch kernel installation test in guest")
error_context.context("Launch kernel installation test in guest")
utils_test.run_virt_sub_test(test, params, env,
sub_type="autotest_control", tag="install")
if params.get("need_reboot", "yes") == "yes":
error.context("Reboot guest after kernel is installed")
error_context.context("Reboot guest after kernel is installed")
session.close()
try:
vm.reboot()
except Exception:
_clean_up_tmp_files(_tmp_file_list)
raise error.TestFail("Could not login guest after install kernel")
test.fail("Could not login guest after install kernel")
# Run Subtest in guest with new kernel
if "sub_test" in params:
error.context("Run sub test in guest with new kernel")
error_context.context("Run sub test in guest with new kernel")
sub_test = params.get("sub_test")
tag = params.get("sub_test_tag", "run")
try:
......@@ -181,15 +181,15 @@ def run(test, params, env):
if params.get("restore_defaut_kernel", "no") == "yes":
# Restore grub
error.context("Restore grub and reboot guest")
error_context.context("Restore grub and reboot guest")
try:
session = vm.wait_for_login(timeout=timeout)
_restore_bootloader_config(session, default_kernel)
except Exception, e:
_clean_up_tmp_files(_tmp_file_list)
session.close()
raise error.TestFail("Fail to restore to default kernel,"
" error message:\n '%s'" % e)
test.fail("Fail to restore to default kernel,"
" error message:\n '%s'" % e)
vm.reboot()
session = vm.wait_for_login(timeout=timeout)
......
qemu/tests/kexec.py
浏览文件 @ 36b98db0
import logging
from autotest.client.shared import error
from virttest import error_context
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
Reboot to new kernel through kexec command:
......@@ -20,24 +20,23 @@ def run(test, params, env):
def check_x2apic_flag():
x2apic_enabled = False
error.context("Check x2apic enabled in guest", logging.info)
error_context.context("Check x2apic enabled in guest", logging.info)
x2apic_output = session.cmd_output(check_x2apic_cmd).strip()
x2apic_check_string = params.get("x2apic_check_string").split(",")
for check_string in x2apic_check_string:
if check_string.strip() in x2apic_output:
x2apic_enabled = True
if not x2apic_enabled:
raise error.TestFail("x2apic is not enabled in guest.")
test.fail("x2apic is not enabled in guest.")
def install_new_kernel():
error.context("Install a new kernel in guest", logging.info)
error_context.context("Install a new kernel in guest", logging.info)
try:
# pylint: disable=E0611
from qemu.tests import rh_kernel_update
rh_kernel_update.run_rh_kernel_update(test, params, env)
except Exception, detail:
raise error.TestError("Failed to install a new kernel in "
"guest: %s" % detail)
test.error("Failed to install a new kernel in guest: %s" % detail)
vm = env.get_vm(params["main_vm"])
vm.verify_alive()
......@@ -57,8 +56,7 @@ def run(test, params, env):
session = vm.wait_for_login(timeout=login_timeout)
count = session.cmd_output(cmd, timeout=cmd_timeout)
if int(count) <= 1:
raise error.TestError("Could not find a new kernel "
"after rh_kernel_update.")
test.error("Could not find a new kernel after rh_kernel_update.")
check_cur_kernel_cmd = params.get("check_cur_kernel_cmd")
cur_kernel_version = session.cmd_output(check_cur_kernel_cmd).strip()
......@@ -72,10 +70,10 @@ def run(test, params, env):
if cur_kernel_version not in kernel:
new_kernel = kernel[7:]
if not new_kernel:
raise error.TestError("Could not find new kernel, "
"command line output: %s" % output)
test.error("Could not find new kernel, "
"command line output: %s" % output)
msg = "Reboot to kernel %s through kexec" % new_kernel
error.context(msg, logging.info)
error_context.context(msg, logging.info)
cmd = params.get("get_kernel_image") % new_kernel
kernel_file = session.cmd_output(cmd).strip().splitlines()[0]
cmd = params.get("get_kernel_ramdisk") % new_kernel
......@@ -88,8 +86,8 @@ def run(test, params, env):
kernel = session.cmd_output(check_cur_kernel_cmd).strip()
logging.info("Current kernel is: %s" % kernel)
if kernel.strip() != new_kernel.strip():
raise error.TestFail("Fail to boot to kernel %s, current kernel is %s"
% (new_kernel, kernel))
test.fail("Fail to boot to kernel %s, current kernel is %s"
% (new_kernel, kernel))
if "yes" in check_x2apic:
check_x2apic_flag()
session.close()
qemu/tests/kill_app.py
浏览文件 @ 36b98db0
......@@ -10,7 +10,6 @@ if application is running when it should .
"""
import logging
import os
from autotest.client.shared import error
def run(test, params, env):
......@@ -29,7 +28,7 @@ def run(test, params, env):
app_name = params.get("kill_app_name", None)
logging.debug("vms %s", vms)
if not vms:
raise error.TestFail("Kill app test launched without any VM parameter")
test.fail("Kill app test launched without any VM parameter")
else:
for vm in vms:
logging.debug("vm %s", vm)
......
qemu/tests/ksm_base.py
浏览文件 @ 36b98db0
......@@ -7,14 +7,14 @@ import re
import aexpect
from autotest.client.shared import error
from virttest import utils_misc, data_dir
from virttest import data_dir
from virttest import error_context
from virttest import utils_misc
TMPFS_OVERHEAD = 0.0022
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
Test how KSM (Kernel Shared Memory) act when more than physical memory is
......@@ -41,8 +41,8 @@ def run(test, params, env):
_ = session.read_until_last_line_matches(["PASS:", "FAIL:"],
timeout)
except aexpect.ExpectProcessTerminatedError, exc:
raise error.TestFail("Command guest script on vm '%s' failed: %s" %
(vm.name, str(exc)))
test.fail("Command guest script on vm '%s' failed: %s" %
(vm.name, str(exc)))
def _execute_allocator(command, vm, session, timeout):
"""
......@@ -66,7 +66,7 @@ def run(test, params, env):
except aexpect.ExpectProcessTerminatedError, exc:
e_str = ("Failed to execute command '%s' on guest script, "
"vm '%s': %s" % (command, vm.name, str(exc)))
raise error.TestFail(e_str)
test.fail(e_str)
return (match, data)
timeout = float(params.get("login_timeout", 240))
......@@ -76,7 +76,7 @@ def run(test, params, env):
session = vm.wait_for_login(timeout=timeout)
# Prepare work in guest
error.context("Turn off swap in guest", logging.info)
error_context.context("Turn off swap in guest", logging.info)
session.cmd_status_output("swapoff -a")
script_file_path = os.path.join(data_dir.get_root_dir(),
"shared/scripts/ksm_overcommit_guest.py")
......@@ -103,9 +103,9 @@ def run(test, params, env):
if query_regex:
sharing_page_0 = re.findall(query_regex, sharing_page_0)[0]
error.context("Start to allocate pages inside guest", logging.info)
error_context.context("Start to allocate pages inside guest", logging.info)
_start_allocator(vm, session, 60)
error.context("Start to fill memory in guest", logging.info)
error_context.context("Start to fill memory in guest", logging.info)
mem_fill = "mem = MemFill(%s, 0, %s)" % (shared_mem, seed)
_execute_allocator(mem_fill, vm, session, fill_timeout)
cmd = "mem.value_fill()"
......@@ -116,8 +116,8 @@ def run(test, params, env):
if query_regex:
sharing_page_1 = re.findall(query_regex, sharing_page_1)[0]
error.context("Start to fill memory with random value in guest",
logging.info)
error_context.context("Start to fill memory with random value in guest",
logging.info)
split = params.get("split")
if split == "yes":
if test_type == "negative":
......@@ -132,7 +132,7 @@ def run(test, params, env):
sharing_page_2 = re.findall(query_regex, sharing_page_2)[0]
# clean up work in guest
error.context("Clean up env in guest", logging.info)
error_context.context("Clean up env in guest", logging.info)
session.cmd_output("die()", 20)
session.cmd_status_output("swapon -a")
session.cmd_output("echo 3 > /proc/sys/vm/drop_caches")
......@@ -168,7 +168,7 @@ def run(test, params, env):
logging.error(fail[turns])
fail_type = fail_type / 2
turns += 1
raise error.TestFail("KSM test failed: %s %s %s" %
(sharing_page_0, sharing_page_1,
sharing_page_2))
test.fail("KSM test failed: %s %s %s" %
(sharing_page_0, sharing_page_1,
sharing_page_2))
session.close()
qemu/tests/ksm_overcommit.py
浏览文件 @ 36b98db0
......@@ -6,9 +6,6 @@ import os
import aexpect
from autotest.client.shared import error
from autotest.client.shared import utils
from avocado.utils import process
from virttest import utils_misc, utils_test, env_process, data_dir
......@@ -81,7 +78,7 @@ def run(test, params, env):
except aexpect.ExpectProcessTerminatedError, details:
e_msg = ("Command ksm_overcommit_guest.py on vm '%s' failed: %s" %
(vm.name, str(details)))
raise error.TestFail(e_msg)
test.fail(e_msg)
def _execute_allocator(command, vm, session, timeout):
"""
......@@ -106,7 +103,7 @@ def run(test, params, env):
e_msg = ("Failed to execute command '%s' on "
"ksm_overcommit_guest.py, vm '%s': %s" %
(command, vm.name, str(details)))
raise error.TestFail(e_msg)
test.fail(e_msg)
return (match, data)
def get_ksmstat():
......@@ -155,8 +152,8 @@ def run(test, params, env):
(not new_ksm and (shm < (ksm_size)))):
if j > 64:
logging.debug(utils_test.get_memory_info(lvms))
raise error.TestError("SHM didn't merge the memory until "
"the DL on guest: %s" % vm.name)
test.error("SHM didn't merge the memory until "
"the DL on guest: %s" % vm.name)
pause = ksm_size / 200 * perf_ratio
logging.debug("Waiting %ds before proceeding...", pause)
time.sleep(pause)
......@@ -209,7 +206,7 @@ def run(test, params, env):
if not lvms[j].is_alive:
e_msg = ("VM %d died while executing static_random_fill on"
" VM %d in allocator loop" % (j, i))
raise error.TestFail(e_msg)
test.fail(e_msg)
vm = lvms[i]
session = lsessions[i]
cmd = "mem.static_random_fill()"
......@@ -226,7 +223,7 @@ def run(test, params, env):
if not vm.is_alive():
e_msg = ("VM %d died while executing "
"static_random_fill on allocator loop" % i)
raise error.TestFail(e_msg)
test.fail(e_msg)
free_mem = int(utils_memory.read_from_meminfo("MemFree"))
if (ksm_swap):
free_mem = (free_mem +
......@@ -323,7 +320,7 @@ def run(test, params, env):
while (shm < ksm_size):
if i > 64:
logging.debug(utils_test.get_memory_info(lvms))
raise error.TestError("SHM didn't merge the memory until DL")
test.error("SHM didn't merge the memory until DL")
pause = ksm_size / 200 * perf_ratio
logging.debug("Waiting %ds before proceed...", pause)
time.sleep(pause)
......@@ -407,28 +404,28 @@ def run(test, params, env):
# Main test code
logging.info("Starting phase 0: Initialization")
if utils.run("ps -C ksmtuned", ignore_status=True).exit_status == 0:
if process.run("ps -C ksmtuned", ignore_status=True).exit_status == 0:
logging.info("Killing ksmtuned...")
utils.run("killall ksmtuned")
process.run("killall ksmtuned")
new_ksm = False
if (os.path.exists("/sys/kernel/mm/ksm/run")):
utils.run("echo 50 > /sys/kernel/mm/ksm/sleep_millisecs")
utils.run("echo 5000 > /sys/kernel/mm/ksm/pages_to_scan")
utils.run("echo 1 > /sys/kernel/mm/ksm/run")
process.run("echo 50 > /sys/kernel/mm/ksm/sleep_millisecs")
process.run("echo 5000 > /sys/kernel/mm/ksm/pages_to_scan")
process.run("echo 1 > /sys/kernel/mm/ksm/run")
e_up = "/sys/kernel/mm/transparent_hugepage/enabled"
e_rh = "/sys/kernel/mm/redhat_transparent_hugepage/enabled"
if os.path.exists(e_up):
utils.run("echo 'never' > %s" % e_up)
process.run("echo 'never' > %s" % e_up)
if os.path.exists(e_rh):
utils.run("echo 'never' > %s" % e_rh)
process.run("echo 'never' > %s" % e_rh)
new_ksm = True
else:
try:
utils.run("modprobe ksm")
utils.run("ksmctl start 5000 100")
except error.CmdError, details:
raise error.TestFail("Failed to load KSM: %s" % details)
process.run("modprobe ksm")
process.run("ksmctl start 5000 100")
except process.CmdError, details:
test.fail("Failed to load KSM: %s" % details)
# host_reserve: mem reserve kept for the host system to run
host_reserve = int(params.get("ksm_host_reserve", -1))
......@@ -597,10 +594,9 @@ def run(test, params, env):
env_process.preprocess_vm(test, params, env, vm_name)
lvms.append(env.get_vm(vm_name))
if not lvms[0]:
raise error.TestError("VM object not found in environment")
test.error("VM object not found in environment")
if not lvms[0].is_alive():
raise error.TestError("VM seems to be dead; Test requires a living "
"VM")
test.error("VM seems to be dead; Test requires a living VM")
logging.debug("Booting first guest %s", lvms[0].name)
......@@ -610,7 +606,7 @@ def run(test, params, env):
tmp = open(params.get('pid_' + vm_name), 'r')
params['pid_' + vm_name] = int(tmp.readline())
except Exception:
raise error.TestFail("Could not get PID of %s" % (vm_name))
test.fail("Could not get PID of %s" % (vm_name))
# Creating other guest systems
for i in range(1, vmsc):
......@@ -630,15 +626,15 @@ def run(test, params, env):
logging.debug("Booting guest %s", lvms[i].name)
lvms[i].create()
if not lvms[i].is_alive():
raise error.TestError("VM %s seems to be dead; Test requires a"
"living VM" % lvms[i].name)
test.error("VM %s seems to be dead; Test requires a"
"living VM" % lvms[i].name)
lsessions.append(lvms[i].wait_for_login(timeout=360))
try:
tmp = open(params.get('pid_' + vm_name), 'r')
params['pid_' + vm_name] = int(tmp.readline())
except Exception:
raise error.TestFail("Could not get PID of %s" % (vm_name))
test.fail("Could not get PID of %s" % (vm_name))
# Let guests rest a little bit :-)
pause = vmsc * 2 * perf_ratio
......
qemu/tests/live_snapshot.py
浏览文件 @ 36b98db0
import time
import logging
from autotest.client.shared import error
from autotest.client.shared import utils
from virttest import error_context
from virttest import utils_misc
from virttest import utils_test
......@@ -20,7 +19,7 @@ def run(test, params, env):
:param env: Dictionary with test environment.
"""
@error.context_aware
@error_context.context_aware
def create_snapshot(vm):
"""
Create live snapshot:
......@@ -28,7 +27,7 @@ def run(test, params, env):
2). Get device info
3). Create snapshot
"""
error.context("Creating live snapshot ...", logging.info)
error_context.context("Creating live snapshot ...", logging.info)
block_info = vm.monitor.info("block")
if vm.monitor.protocol == 'qmp':
device = block_info[0]["device"]
......@@ -42,7 +41,7 @@ def run(test, params, env):
snapshot_info = str(vm.monitor.info("block"))
if snapshot_name not in snapshot_info:
logging.error(snapshot_info)
raise error.TestFail("Snapshot doesn't exist")
test.fail("Snapshot doesn't exist")
timeout = int(params.get("login_timeout", 360))
dd_timeout = int(params.get("dd_timeout", 900))
......@@ -94,7 +93,7 @@ def run(test, params, env):
def installation_test():
args = (test, params, env)
bg = utils.InterruptedThread(
bg = utils_misc.InterruptedThread(
utils_test.run_virt_sub_test, args,
{"sub_type": "unattended_install"})
bg.start()
......
qemu/tests/live_snapshot_base.py
浏览文件 @ 36b98db0
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 virttest import storage
from virttest import data_dir
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
live_snapshot_base test:
......@@ -40,28 +41,29 @@ def run(test, params, env):
dst = "c:\\users\\public\\%s" % tmp_name
try:
error.context("create file on host, copy it to guest", logging.info)
error_context.context("create file on host, copy it to guest",
logging.info)
cmd = params.get("dd_cmd") % src
utils.system(cmd, timeout=dd_timeout)
md5 = utils.hash_file(src, method="md5")
process.system(cmd, timeout=dd_timeout)
md5 = crypto.hash_file(src, method="md5")
vm.copy_files_to(src, dst, timeout=copy_timeout)
utils.system("rm -f %s" % src)
error.context("create live snapshot", logging.info)
process.system("rm -f %s" % src)
error_context.context("create live snapshot", logging.info)
if vm.live_snapshot(base_file, snapshot_file,
snapshot_format) != device:
raise error.TestFail("Fail to create snapshot")
test.fail("Fail to create snapshot")
backing_file = vm.monitor.get_backingfile(device)
if backing_file != base_file:
logging.error(
"backing file: %s, base file: %s", backing_file, base_file)
raise error.TestFail("Got incorrect backing file")
error.context("copy file to host, check content not changed",
logging.info)
test.fail("Got incorrect backing file")
error_context.context("copy file to host, check content not changed",
logging.info)
vm.copy_files_from(dst, src, timeout=copy_timeout)
if md5 and (md5 != utils.hash_file(src, method="md5")):
raise error.TestFail("diff md5 before/after create snapshot")
if md5 and (md5 != crypto.hash_file(src, method="md5")):
test.fail("diff md5 before/after create snapshot")
session.cmd(params.get("alive_check_cmd", "dir"))
finally:
if session:
session.close()
utils.system("rm -f %s %s" % (snapshot_file, src))
process.system("rm -f %s %s" % (snapshot_file, src))
qemu/tests/live_snapshot_chain.py
浏览文件 @ 36b98db0
......@@ -3,14 +3,13 @@ import re
import logging
import time
from autotest.client.shared import error
from virttest import error_context
from virttest import storage
from virttest import qemu_storage
from virttest import data_dir
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
live_snapshot chain test:
......@@ -112,14 +111,15 @@ def run(test, params, env):
data_dir.get_data_dir())
snapshot_format = image_params.get("image_format")
error.context("Do pre snapshot operates", logging.info)
error_context.context("Do pre snapshot operates", logging.info)
if image_params.get("pre_snapshot_cmd"):
do_operate(image_params, "pre_snapshot_cmd")
error.context("Do live snapshot ", logging.info)
error_context.context("Do live snapshot ", logging.info)
vm.live_snapshot(base_file, snapshot_file, snapshot_format)
error.context("Do post snapshot operates", logging.info)
error_context.context("Do post snapshot operates",
logging.info)
if image_params.get("post_snapshot_cmd"):
do_operate(image_params, "post_snapshot_cmd")
md5 = ""
......@@ -141,12 +141,12 @@ def run(test, params, env):
files_in_guest[image] = files_check
session.close()
error.context("Reboot guest", logging.info)
error_context.context("Reboot guest", logging.info)
if image_params.get("need_reboot", "no") == "yes":
vm.monitor.cmd("system_reset")
vm.verify_alive()
error.context("Do base files check", logging.info)
error_context.context("Do base files check", logging.info)
snapshot_chain_backward = snapshot_chain[:]
snapshot_chain_backward.reverse()
......@@ -166,7 +166,7 @@ def run(test, params, env):
(file, image)
error_message += "from '%s' to '%s'(md5)" %\
(md5_value[image][file], md5)
raise error.TestFail(error_message)
test.fail(error_message)
files_check = session.cmd(file_check_cmd % file_dir)
if files_check != files_in_guest[image]:
error_message = "Files in image %s is not as expect:" %\
......@@ -174,22 +174,22 @@ def run(test, params, env):
error_message += "Before shut down: %s" %\
files_in_guest[image]
error_message += "Now: %s" % files_check
raise error.TestFail(error_message)
test.fail(error_message)
if image_params.get("image_check"):
image = qemu_storage.QemuImg(
image_params, data_dir.get_data_dir(), image)
image.check_image(image_params, data_dir.get_data_dir())
session.close()
error.context("Remove snapshot images", logging.info)
error_context.context("Remove snapshot images", logging.info)
if vm.is_alive():
vm.destroy()
errs = cleanup_images(snapshot_chain, params)
test.assertFalse(errs, "Errors occurred while removing images:\n%s"
% "\n".join(errs))
except Exception as details:
error.context("Force-cleaning after exception: %s" % details,
logging.error)
error_context.context("Force-cleaning after exception: %s" % details,
logging.error)
if vm.is_alive():
vm.destroy()
cleanup_images(snapshot_chain, params)
......
qemu/tests/live_snapshot_runtime.py
浏览文件 @ 36b98db0
from autotest.client.shared import utils
from autotest.client.shared import error
from virttest import error_context
from virttest import utils_misc
from qemu.tests import live_snapshot_basic
......@@ -8,7 +9,7 @@ class LiveSnapshotRuntime(live_snapshot_basic.LiveSnapshot):
def __init__(self, test, params, env, tag):
super(LiveSnapshotRuntime, self).__init__(test, params, env, tag)
@error.context_aware
@error_context.context_aware
def reboot(self):
"""
Reset guest with system_reset;
......@@ -16,7 +17,7 @@ class LiveSnapshotRuntime(live_snapshot_basic.LiveSnapshot):
method = self.params.get("reboot_method", "system_reset")
return super(LiveSnapshotRuntime, self).reboot(method=method, boot_check=False)
@error.context_aware
@error_context.context_aware
def action_when_start(self):
"""
start pre-action in new threads;
......@@ -26,7 +27,7 @@ class LiveSnapshotRuntime(live_snapshot_basic.LiveSnapshot):
for test in self.params.get("when_start").split():
if hasattr(self, test):
fun = getattr(self, test)
bg = utils.InterruptedThread(fun)
bg = utils_misc.InterruptedThread(fun)
bg.start()
if bg.isAlive():
self.create_snapshot()
......
qemu/tests/live_snapshot_transaction.py
浏览文件 @ 36b98db0
import logging
from autotest.client.shared import error
from virttest import error_context
from qemu.tests import live_snapshot_basic
@error.context_aware
@error_context.context_aware
def run(test, params, env):
"""
live_snapshot_transaction test:
......@@ -32,14 +33,14 @@ def run(test, params, env):
"data": transaction_test.snapshot_args}
arg_list.append(args)
error.context("Create multiple live snapshots simultaneously"
" with transaction", logging.info)
error_context.context("Create multiple live snapshots simultaneously"
" with transaction", logging.info)
output = transaction_test.vm.monitor.transaction(arg_list)
# return nothing on successful transaction
if bool(output):
raise error.TestFail("Live snapshot transatcion failed,"
" there should be nothing on success.\n"
"More details: %s" % output)
test.fail("Live snapshot transatcion failed,"
" there should be nothing on success.\n"
"More details: %s" % output)
transaction_test.action_after_finished()
finally:
try:
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或