Signed-off-by: Lucas Meneghel Rodrigues <lmr@xxxxxxxxxx>
---
client/tests/kvm/build.py | 335 --------------------------------
client/tests/kvm/stepmaker.py | 357 -----------------------------------
client/tests/kvm/steps.py | 249 ------------------------
client/tests/kvm/tests/build.py | 335 ++++++++++++++++++++++++++++++++
client/tests/kvm/tests/stepmaker.py | 357 +++++++++++++++++++++++++++++++++++
client/tests/kvm/tests/steps.py | 249 ++++++++++++++++++++++++
6 files changed, 941 insertions(+), 941 deletions(-)
delete mode 100644 client/tests/kvm/build.py
delete mode 100755 client/tests/kvm/stepmaker.py
delete mode 100644 client/tests/kvm/steps.py
create mode 100644 client/tests/kvm/tests/__init__.py
create mode 100644 client/tests/kvm/tests/build.py
create mode 100755 client/tests/kvm/tests/stepmaker.py
create mode 100644 client/tests/kvm/tests/steps.py
diff --git a/client/tests/kvm/build.py b/client/tests/kvm/build.py
deleted file mode 100644
index f6856ad..0000000
--- a/client/tests/kvm/build.py
+++ /dev/null
@@ -1,335 +0,0 @@
-import time, os, sys, urllib, re, signal, logging, datetime
-from autotest_lib.client.bin import utils, test
-from autotest_lib.client.common_lib import error
-import kvm_utils
-
-
-def load_kvm_modules(module_dir):
- """
- Unload previously loaded kvm modules, then load modules present on any
- sub directory of module_dir. Function will walk through module_dir until
- it finds the modules.
-
- @param module_dir: Directory where the KVM modules are located.
- """
- vendor = "intel"
- if os.system("grep vmx /proc/cpuinfo 1>/dev/null") != 0:
- vendor = "amd"
- logging.debug("Detected CPU vendor as '%s'" %(vendor))
-
- logging.debug("Killing any qemu processes that might be left behind")
- utils.system("pkill qemu", ignore_status=True)
-
- logging.info("Unloading previously loaded KVM modules")
- kvm_utils.unload_module("kvm")
- if utils.module_is_loaded("kvm"):
- message = "Failed to remove old KVM modules"
- logging.error(message)
- raise error.TestError(message)
-
- logging.info("Loading new KVM modules...")
- kvm_module_path = None
- kvm_vendor_module_path = None
- # Search for the built KVM modules
- for folder, subdirs, files in os.walk(module_dir):
- if "kvm.ko" in files:
- kvm_module_path = os.path.join(folder, "kvm.ko")
- kvm_vendor_module_path = os.path.join(folder, "kvm-%s.ko" % vendor)
- abort = False
- if not kvm_module_path:
- logging.error("Need a directory containing both kernel module and "
- "userspace sources.")
- logging.error("If you are trying to build only KVM userspace and use "
- "the KVM modules you have already loaded, put "
- "'load_modules': 'no' on the control file 'params' "
- "dictionary.")
- raise error.TestError("Could not find a built kvm.ko module on the "
- "source dir.")
- elif not os.path.isfile(kvm_vendor_module_path):
- logging.error("Could not find KVM (%s) module that was supposed to be"
- " built on the source dir", vendor)
- abort = True
- if abort:
- raise error.TestError("Could not load KVM modules.")
- utils.system("/sbin/insmod %s" % kvm_module_path)
- time.sleep(1)
- utils.system("/sbin/insmod %s" % kvm_vendor_module_path)
-
- if not utils.module_is_loaded("kvm"):
- message = "Failed to load the KVM modules built for the test"
- logging.error(message)
- raise error.TestError(message)
-
-
-def create_symlinks(test_bindir, prefix):
- """
- Create symbolic links for the appropriate qemu and qemu-img commands on
- the kvm test bindir.
-
- @param test_bindir: KVM test bindir
- @param prefix: KVM prefix path
- """
- qemu_path = os.path.join(test_bindir, "qemu")
- qemu_img_path = os.path.join(test_bindir, "qemu-img")
- if os.path.lexists(qemu_path):
- os.unlink(qemu_path)
- if os.path.lexists(qemu_img_path):
- os.unlink(qemu_img_path)
- kvm_qemu = os.path.join(prefix, "bin", "qemu-system-x86_64")
- if not os.path.isfile(kvm_qemu):
- raise error.TestError('Invalid qemu path')
- kvm_qemu_img = os.path.join(prefix, "bin", "qemu-img")
- if not os.path.isfile(kvm_qemu_img):
- raise error.TestError('Invalid qemu-img path')
- os.symlink(kvm_qemu, qemu_path)
- os.symlink(kvm_qemu_img, qemu_img_path)
-
-
-class SourceDirInstaller:
- """
- Class that handles building/installing KVM directly from a tarball or
- a single source code dir.
- """
- def __init__(self, test, params):
- """
- Initializes class attributes, and retrieves KVM code.
-
- @param test: kvm test object
- @param params: Dictionary with test arguments
- """
- install_mode = params["mode"]
- srcdir = params.get("srcdir")
- # KVM build prefix
- self.test_bindir = test.bindir
- prefix = os.path.join(test.bindir, 'build')
- self.prefix = os.path.abspath(prefix)
- # Are we going to load modules?
- load_modules = params.get('load_modules')
- if not load_modules:
- self.load_modules = True
- elif load_modules == 'yes':
- self.load_modules = True
- elif load_modules == 'no':
- self.load_modules = False
-
- if install_mode == 'localsrc':
- if not srcdir:
- raise error.TestError("Install from source directory specified"
- "but no source directory provided on the"
- "control file.")
- else:
- self.srcdir = srcdir
- self.repo_type = kvm_utils.check_kvm_source_dir(self.srcdir)
- return
- else:
- srcdir = test.srcdir
- if not os.path.isdir(srcdir):
- os.makedirs(srcdir)
-
- if install_mode == 'release':
- release_tag = params.get("release_tag")
- release_dir = params.get("release_dir")
- release_listing = params.get("release_listing")
- logging.info("Installing KVM from release tarball")
- if not release_tag:
- release_tag = kvm_utils.get_latest_kvm_release_tag(
- release_listing)
- tarball = os.path.join(release_dir, "kvm-%s.tar.gz" % release_tag)
- logging.info("Retrieving release kvm-%s" % release_tag)
- tarball = utils.unmap_url("/", tarball, "/tmp")
-
- elif install_mode == 'snapshot':
- logging.info("Installing KVM from snapshot")
- snapshot_dir = params.get("snapshot_dir")
- if not snapshot_dir:
- raise error.TestError("Snapshot dir not provided")
- snapshot_date = params.get("snapshot_date")
- if not snapshot_date:
- # Take yesterday's snapshot
- d = (datetime.date.today() -
- datetime.timedelta(1)).strftime("%Y%m%d")
- else:
- d = snapshot_date
- tarball = os.path.join(snapshot_dir, "kvm-snapshot-%s.tar.gz" % d)
- logging.info("Retrieving kvm-snapshot-%s" % d)
- tarball = utils.unmap_url("/", tarball, "/tmp")
-
- elif install_mode == 'localtar':
- tarball = params.get("tarball")
- if not tarball:
- raise error.TestError("KVM Tarball install specified but no"
- " tarball provided on control file.")
- logging.info("Installing KVM from a local tarball")
- logging.info("Using tarball %s")
- tarball = utils.unmap_url("/", params.get("tarball"), "/tmp")
-
- os.chdir(srcdir)
- self.srcdir = os.path.join(srcdir, utils.extract_tarball(tarball))
- self.repo_type = kvm_utils.check_kvm_source_dir(self.srcdir)
-
-
- def __build(self):
- os.chdir(self.srcdir)
- # For testing purposes, it's better to build qemu binaries with
- # debugging symbols, so we can extract more meaningful stack traces.
- cfg = "./configure --disable-strip --prefix=%s" % self.prefix
- steps = [cfg, "make clean", "make -j %s" % (utils.count_cpus() + 1)]
- logging.info("Building KVM")
- for step in steps:
- utils.system(step)
-
-
- def __install(self):
- os.chdir(self.srcdir)
- logging.info("Installing KVM userspace")
- if self.repo_type == 1:
- utils.system("make -C qemu install")
- elif self.repo_type == 2:
- utils.system("make install")
- create_symlinks(self.test_bindir, self.prefix)
-
-
- def __load_modules(self):
- load_kvm_modules(self.srcdir)
-
-
- def install(self):
- self.__build()
- self.__install()
- if self.load_modules:
- self.__load_modules()
-
-
-class GitInstaller:
- def __init__(self, test, params):
- """
- Initialize class parameters and retrieves code from git repositories.
-
- @param test: kvm test object.
- @param params: Dictionary with test parameters.
- """
- install_mode = params["mode"]
- srcdir = params.get("srcdir", test.bindir)
- if not srcdir:
- os.makedirs(srcdir)
- self.srcdir = srcdir
- # KVM build prefix
- self.test_bindir = test.bindir
- prefix = os.path.join(test.bindir, 'build')
- self.prefix = os.path.abspath(prefix)
- # Are we going to load modules?
- load_modules = params.get('load_modules')
- if not load_modules:
- self.load_modules = True
- elif load_modules == 'yes':
- self.load_modules = True
- elif load_modules == 'no':
- self.load_modules = False
-
- kernel_repo = params.get("git_repo")
- user_repo = params.get("user_git_repo")
- kmod_repo = params.get("kmod_repo")
-
- branch = params.get("git_branch", "master")
- lbranch = params.get("lbranch")
-
- tag = params.get("git_tag", "HEAD")
- user_tag = params.get("user_git_tag", "HEAD")
- kmod_tag = params.get("kmod_git_tag", "HEAD")
-
- if not kernel_repo:
- message = "KVM git repository path not specified"
- logging.error(message)
- raise error.TestError(message)
- if not user_repo:
- message = "KVM user git repository path not specified"
- logging.error(message)
- raise error.TestError(message)
-
- kernel_srcdir = os.path.join(srcdir, "kvm")
- kvm_utils.get_git_branch(kernel_repo, branch, kernel_srcdir, tag,
- lbranch)
- self.kernel_srcdir = kernel_srcdir
-
- userspace_srcdir = os.path.join(srcdir, "kvm_userspace")
- kvm_utils.get_git_branch(user_repo, branch, userspace_srcdir, user_tag,
- lbranch)
- self.userspace_srcdir = userspace_srcdir
-
- if kmod_repo:
- kmod_srcdir = os.path.join (srcdir, "kvm_kmod")
- kvm_utils.get_git_branch(kmod_repo, branch, kmod_srcdir, user_tag,
- lbranch)
- self.kmod_srcdir = kmod_srcdir
-
-
- def __build(self):
- if self.kmod_srcdir:
- logging.info('Building KVM modules')
- os.chdir(self.kmod_srcdir)
- utils.system('./configure')
- utils.system('make clean')
- utils.system('make sync LINUX=%s' % self.kernel_srcdir)
- utils.system('make -j %s' % utils.count_cpus())
- logging.info('Building KVM userspace code')
- os.chdir(self.userspace_srcdir)
- utils.system('./configure --disable-strip --prefix=%s' %
- self.prefix)
- utils.system('make clean')
- utils.system('make -j %s' % utils.count_cpus())
- else:
- os.chdir(self.userspace_srcdir)
- utils.system('./configure --disable-strip --prefix=%s' %
- self.prefix)
- logging.info('Building KVM modules')
- utils.system('make clean')
- utils.system('make -C kernel LINUX=%s sync' % self.kernel_srcdir)
- logging.info('Building KVM userspace code')
- utils.system('make -j %s' % utils.count_cpus())
-
-
- def __install(self):
- os.chdir(self.userspace_srcdir)
- utils.system('make install')
- create_symlinks(self.test_bindir, self.prefix)
-
-
- def __load_modules(self):
- if self.kmod_srcdir:
- load_kvm_modules(self.kmod_srcdir)
- else:
- load_kvm_modules(self.userspace_srcdir)
-
-
- def install(self):
- self.__build()
- self.__install()
- if self.load_modules:
- self.__load_modules()
-
-
-def run_build(test, params, env):
- """
- Installs KVM using the selected install mode. Most install methods will
- take kvm source code, build it and install it to a given location.
-
- @param test: kvm test object.
- @param params: Dictionary with test parameters.
- @param env: Test environment.
- """
- install_mode = params.get("mode")
- srcdir = params.get("srcdir", test.srcdir)
- params["srcdir"] = srcdir
-
- if install_mode == 'noinstall':
- logging.info("Skipping installation")
- return
- elif install_mode in ['localsrc', 'localtar', 'release', 'snapshot']:
- installer = SourceDirInstaller(test, params)
- elif install_mode == 'git':
- installer = GitInstaller(test, params)
- else:
- raise error.TestError('Invalid or unsupported'
- ' install mode: %s' % install_mode)
-
- installer.install()
diff --git a/client/tests/kvm/stepmaker.py b/client/tests/kvm/stepmaker.py
deleted file mode 100755
index 24060db..0000000
--- a/client/tests/kvm/stepmaker.py
+++ /dev/null
@@ -1,357 +0,0 @@
-#!/usr/bin/python
-"""
-Step file creator/editor.
-
-@copyright: Red Hat Inc 2009
-@author: mgoldish@xxxxxxxxxx (Michael Goldish)
-@version: "20090401"
-"""
-
-import pygtk, gtk, gobject, time, os, commands
-import common
-from autotest_lib.client.common_lib import error
-import kvm_utils, logging, ppm_utils, stepeditor
-pygtk.require('2.0')
-
-
-class StepMaker(stepeditor.StepMakerWindow):
- """
- Application used to create a step file. It will grab your input to the
- virtual machine and record it on a 'step file', that can be played
- making it possible to do unattended installs.
- """
- # Constructor
- def __init__(self, vm, steps_filename, tempdir, params):
- stepeditor.StepMakerWindow.__init__(self)
-
- self.vm = vm
- self.steps_filename = steps_filename
- self.steps_data_dir = ppm_utils.get_data_dir(steps_filename)
- self.tempdir = tempdir
- self.screendump_filename = os.path.join(tempdir, "scrdump.ppm")
- self.params = params
-
- if not os.path.exists(self.steps_data_dir):
- os.makedirs(self.steps_data_dir)
-
- self.steps_file = open(self.steps_filename, "w")
- self.vars_file = open(os.path.join(self.steps_data_dir, "vars"), "w")
-
- self.step_num = 1
- self.run_time = 0
- self.update_delay = 1000
- self.prev_x = 0
- self.prev_y = 0
- self.vars = {}
- self.timer_id = None
-
- self.time_when_done_clicked = time.time()
- self.time_when_actions_completed = time.time()
-
- self.steps_file.write("# Generated by Step Maker\n")
- self.steps_file.write("# Generated on %s\n" % time.asctime())
- self.steps_file.write("# uname -a: %s\n" %
- commands.getoutput("uname -a"))
- self.steps_file.flush()
-
- self.vars_file.write("# This file lists the vars used during recording"
- " with Step Maker\n")
- self.vars_file.flush()
-
- # Done/Break HBox
- hbox = gtk.HBox(spacing=10)
- self.user_vbox.pack_start(hbox)
- hbox.show()
-
- self.button_break = gtk.Button("Break")
- self.button_break.connect("clicked", self.event_break_clicked)
- hbox.pack_start(self.button_break)
- self.button_break.show()
-
- self.button_done = gtk.Button("Done")
- self.button_done.connect("clicked", self.event_done_clicked)
- hbox.pack_start(self.button_done)
- self.button_done.show()
-
- # Set window title
- self.window.set_title("Step Maker")
-
- # Connect "capture" button
- self.button_capture.connect("clicked", self.event_capture_clicked)
-
- # Switch to run mode
- self.switch_to_run_mode()
-
-
- def destroy(self, widget):
- self.vm.send_monitor_cmd("cont")
- self.steps_file.close()
- self.vars_file.close()
- stepeditor.StepMakerWindow.destroy(self, widget)
-
-
- # Utilities
- def redirect_timer(self, delay=0, func=None):
- if self.timer_id != None:
- gobject.source_remove(self.timer_id)
- self.timer_id = None
- if func != None:
- self.timer_id = gobject.timeout_add(delay, func,
- priority=gobject.PRIORITY_LOW)
-
-
- def switch_to_run_mode(self):
- # Set all widgets to their default states
- self.clear_state(clear_screendump=False)
- # Enable/disable some widgets
- self.button_break.set_sensitive(True)
- self.button_done.set_sensitive(False)
- self.data_vbox.set_sensitive(False)
- # Give focus to the Break button
- self.button_break.grab_focus()
- # Start the screendump timer
- self.redirect_timer(100, self.update)
- # Resume the VM
- self.vm.send_monitor_cmd("cont")
-
-
- def switch_to_step_mode(self):
- # Set all widgets to their default states
- self.clear_state(clear_screendump=False)
- # Enable/disable some widgets
- self.button_break.set_sensitive(False)
- self.button_done.set_sensitive(True)
- self.data_vbox.set_sensitive(True)
- # Give focus to the keystrokes entry widget
- self.entry_keys.grab_focus()
- # Start the screendump timer
- self.redirect_timer()
- # Stop the VM
- self.vm.send_monitor_cmd("stop")
-
-
- # Events in step mode
- def update(self):
- self.redirect_timer()
-
- if os.path.exists(self.screendump_filename):
- os.unlink(self.screendump_filename)
-
- (status, output) = self.vm.send_monitor_cmd("screendump " +
- self.screendump_filename)
- if status: # Failure
- logging.info("Could not fetch screendump")
- else:
- self.set_image_from_file(self.screendump_filename)
-
- self.redirect_timer(self.update_delay, self.update)
- return True
-
-
- def event_break_clicked(self, widget):
- if not self.vm.is_alive():
- self.message("The VM doesn't seem to be alive.", "Error")
- return
- # Switch to step mode
- self.switch_to_step_mode()
- # Compute time elapsed since last click on "Done" and add it
- # to self.run_time
- self.run_time += time.time() - self.time_when_done_clicked
- # Set recording time widget
- self.entry_time.set_text("%.2f" % self.run_time)
- # Update screendump ID
- self.update_screendump_id(self.steps_data_dir)
- # By default, check the barrier checkbox
- self.check_barrier.set_active(True)
- # Set default sleep and barrier timeout durations
- time_delta = time.time() - self.time_when_actions_completed
- if time_delta < 1.0: time_delta = 1.0
- self.spin_sleep.set_value(round(time_delta))
- self.spin_barrier_timeout.set_value(round(time_delta * 5))
- # Set window title
- self.window.set_title("Step Maker -- step %d at time %.2f" %
- (self.step_num, self.run_time))
-
-
- def event_done_clicked(self, widget):
- # Get step lines and screendump
- lines = self.get_step_lines(self.steps_data_dir)
- if lines == None:
- return
-
- # Get var values from user and write them to vars file
- vars = {}
- for line in lines.splitlines():
- words = line.split()
- if words and words[0] == "var":
- varname = words[1]
- if varname in self.vars.keys():
- val = self.vars[varname]
- elif varname in vars.keys():
- val = vars[varname]
- elif varname in self.params.keys():
- val = self.params[varname]
- vars[varname] = val
- else:
- val = self.inputdialog("$%s =" % varname, "Variable")
- if val == None:
- return
- vars[varname] = val
- for varname in vars.keys():
- self.vars_file.write("%s=%s\n" % (varname, vars[varname]))
- self.vars.update(vars)
-
- # Write step lines to file
- self.steps_file.write("# " + "-" * 32 + "\n")
- self.steps_file.write(lines)
-
- # Flush buffers of both files
- self.steps_file.flush()
- self.vars_file.flush()
-
- # Remember the current time
- self.time_when_done_clicked = time.time()
-
- # Switch to run mode
- self.switch_to_run_mode()
-
- # Send commands to VM
- for line in lines.splitlines():
- words = line.split()
- if not words:
- continue
- elif words[0] == "key":
- self.vm.send_key(words[1])
- elif words[0] == "var":
- val = self.vars.get(words[1])
- if not val:
- continue
- self.vm.send_string(val)
- elif words[0] == "mousemove":
- self.vm.send_monitor_cmd("mouse_move %d %d" % (-8000,-8000))
- time.sleep(0.5)
- self.vm.send_monitor_cmd("mouse_move %s %s" % (words[1],
- words[2]))
- time.sleep(0.5)
- elif words[0] == "mouseclick":
- self.vm.send_monitor_cmd("mouse_button %s" % words[1])
- time.sleep(0.1)
- self.vm.send_monitor_cmd("mouse_button 0")
-
- # Remember the current time
- self.time_when_actions_completed = time.time()
-
- # Move on to next step
- self.step_num += 1
-
- def event_capture_clicked(self, widget):
- self.message("Mouse actions disabled (for now).", "Sorry")
- return
-
- self.image_width_backup = self.image_width
- self.image_height_backup = self.image_height
- self.image_data_backup = self.image_data
-
- gtk.gdk.pointer_grab(self.event_box.window, False,
- gtk.gdk.BUTTON_PRESS_MASK |
- gtk.gdk.BUTTON_RELEASE_MASK)
- # Create empty cursor
- pix = gtk.gdk.Pixmap(self.event_box.window, 1, 1, 1)
- color = gtk.gdk.Color()
- cursor = gtk.gdk.Cursor(pix, pix, color, color, 0, 0)
- self.event_box.window.set_cursor(cursor)
- gtk.gdk.display_get_default().warp_pointer(gtk.gdk.screen_get_default(),
- self.prev_x, self.prev_y)
- self.redirect_event_box_input(
- self.event_capture_button_press,
- self.event_capture_button_release,
- self.event_capture_scroll)
- self.redirect_timer(10, self.update_capture)
- self.vm.send_monitor_cmd("cont")
-
- # Events in mouse capture mode
-
- def update_capture(self):
- self.redirect_timer()
-
- (screen, x, y, flags) = gtk.gdk.display_get_default().get_pointer()
- self.mouse_click_coords[0] = int(x * self.spin_sensitivity.get_value())
- self.mouse_click_coords[1] = int(y * self.spin_sensitivity.get_value())
-
- delay = self.spin_latency.get_value() / 1000
- if (x, y) != (self.prev_x, self.prev_y):
- self.vm.send_monitor_cmd("mouse_move %d %d" % (-8000, -8000))
- time.sleep(delay)
- self.vm.send_monitor_cmd("mouse_move %d %d" %
- (self.mouse_click_coords[0],
- self.mouse_click_coords[1]))
- time.sleep(delay)
-
- self.prev_x = x
- self.prev_y = y
-
- if os.path.exists(self.screendump_filename):
- os.unlink(self.screendump_filename)
-
- (status, output) = self.vm.send_monitor_cmd("screendump " +
- self.screendump_filename)
- if status: # Failure
- logging.info("Could not fetch screendump")
- else:
- self.set_image_from_file(self.screendump_filename)
-
- self.redirect_timer(int(self.spin_latency.get_value()),
- self.update_capture)
- return True
-
- def event_capture_button_press(self, widget,event):
- pass
-
- def event_capture_button_release(self, widget,event):
- gtk.gdk.pointer_ungrab()
- self.event_box.window.set_cursor(gtk.gdk.Cursor(gtk.gdk.CROSSHAIR))
- self.redirect_event_box_input(
- self.event_button_press,
- self.event_button_release,
- None,
- None,
- self.event_expose)
- self.redirect_timer()
- self.vm.send_monitor_cmd("stop")
- self.mouse_click_captured = True
- self.mouse_click_button = event.button
- self.set_image(self.image_width_backup, self.image_height_backup,
- self.image_data_backup)
- self.check_mousemove.set_sensitive(True)
- self.check_mouseclick.set_sensitive(True)
- self.check_mousemove.set_active(True)
- self.check_mouseclick.set_active(True)
- self.update_mouse_click_info()
-
- def event_capture_scroll(self, widget, event):
- if event.direction == gtk.gdk.SCROLL_UP:
- direction = 1
- else:
- direction = -1
- self.spin_sensitivity.set_value(self.spin_sensitivity.get_value() +
- direction)
- pass
-
-
-def run_stepmaker(test, params, env):
- vm = kvm_utils.env_get_vm(env, params.get("main_vm"))
- if not vm:
- raise error.TestError("VM object not found in environment")
- if not vm.is_alive():
- raise error.TestError("VM seems to be dead; Step Maker requires a"
- " living VM")
-
- steps_filename = params.get("steps")
- if not steps_filename:
- raise error.TestError("Steps filename not specified")
- steps_filename = kvm_utils.get_path(test.bindir, steps_filename)
- if os.path.exists(steps_filename):
- raise error.TestError("Steps file %s already exists" % steps_filename)
-
- StepMaker(vm, steps_filename, test.debugdir, params)
- gtk.main()
diff --git a/client/tests/kvm/steps.py b/client/tests/kvm/steps.py
deleted file mode 100644
index 8bc85f2..0000000
--- a/client/tests/kvm/steps.py
+++ /dev/null
@@ -1,249 +0,0 @@
-"""
-Utilities to perform automatic guest installation using step files.
-
-@copyright: Red Hat 2008-2009
-"""
-
-import os, time, md5, re, shutil, logging
-from autotest_lib.client.common_lib import utils, error
-import kvm_utils, ppm_utils, kvm_subprocess
-try:
- import PIL.Image
-except ImportError:
- logging.warning('No python imaging library installed. PPM image '
- 'conversion to JPEG disabled. In order to enable it, '
- 'please install python-imaging or the equivalent for your '
- 'distro.')
-
-
-def handle_var(vm, params, varname):
- var = params.get(varname)
- if not var:
- return False
- vm.send_string(var)
- return True
-
-
-def barrier_2(vm, words, params, debug_dir, data_scrdump_filename,
- current_step_num):
- if len(words) < 7:
- logging.error("Bad barrier_2 command line")
- return False
-
- cmd, dx, dy, x1, y1, md5sum, timeout = words[:7]
- dx, dy, x1, y1, timeout = map(int, [dx, dy, x1, y1, timeout])
-
- # Timeout/5 is the time it took stepmaker to complete this step.
- # Divide that number by 10 to poll 10 times, just in case
- # current machine is stronger then the "stepmaker machine".
- # Limit to 1 (min) and 10 (max) seconds between polls.
- sleep_duration = float(timeout) / 50.0
- if sleep_duration < 1.0: sleep_duration = 1.0
- if sleep_duration > 10.0: sleep_duration = 10.0
-
- scrdump_filename = os.path.join(debug_dir, "scrdump.ppm")
- cropped_scrdump_filename = os.path.join(debug_dir, "cropped_scrdump.ppm")
- expected_scrdump_filename = os.path.join(debug_dir, "scrdump_expected.ppm")
- expected_cropped_scrdump_filename = os.path.join(debug_dir,
- "cropped_scrdump_expected.ppm")
- comparison_filename = os.path.join(debug_dir, "comparison.ppm")
-
- fail_if_stuck_for = params.get("fail_if_stuck_for")
- if fail_if_stuck_for:
- fail_if_stuck_for = float(fail_if_stuck_for)
- else:
- fail_if_stuck_for = 1e308
-
- stuck_detection_history = params.get("stuck_detection_history")
- if stuck_detection_history:
- stuck_detection_history = int(stuck_detection_history)
- else:
- stuck_detection_history = 2
-
- keep_screendump_history = params.get("keep_screendump_history") == "yes"
- if keep_screendump_history:
- keep_all_history = params.get("keep_all_history") == "yes"
- history_dir = os.path.join(debug_dir, "barrier_history")
-
- end_time = time.time() + timeout
- end_time_stuck = time.time() + fail_if_stuck_for
- start_time = time.time()
-
- prev_whole_image_md5sums = []
-
- failure_message = None
-
- # Main loop
- while True:
- # Check for timeouts
- if time.time() > end_time:
- failure_message = "regular timeout"
- break
- if time.time() > end_time_stuck:
- failure_message = "guest is stuck"
- break
-
- # Make sure vm is alive
- if not vm.is_alive():
- failure_message = "VM is dead"
- break
-
- # Request screendump
- (status, output) = vm.send_monitor_cmd("screendump %s" %
- scrdump_filename)
- if status:
- logging.error("Could not fetch screendump")
- continue
-
- # Read image file
- (w, h, data) = ppm_utils.image_read_from_ppm_file(scrdump_filename)
-
- # Make sure image is valid
- if not ppm_utils.image_verify_ppm_file(scrdump_filename):
- logging.warn("Got invalid screendump: dimensions: %dx%d, "
- "data size: %d" % (w, h, len(data)))
- continue
-
- # Compute md5sum of whole image
- whole_image_md5sum = ppm_utils.image_md5sum(w, h, data)
-
- # Write screendump to history_dir (as JPG) if requested
- # and if the screendump differs from the previous one
- if (keep_screendump_history and
- whole_image_md5sum not in prev_whole_image_md5sums[:1]):
- try:
- os.makedirs(history_dir)
- except:
- pass
- history_scrdump_filename = os.path.join(history_dir,
- "scrdump-step_%s-%s.jpg" % (current_step_num,
- time.strftime("%Y%m%d-%H%M%S")))
- try:
- image = PIL.Image.open(scrdump_filename)
- image.save(history_scrdump_filename, format = 'JPEG',
- quality = 30)
- except NameError:
- pass
-
- # Compare md5sum of barrier region with the expected md5sum
- calced_md5sum = ppm_utils.get_region_md5sum(w, h, data, x1, y1, dx, dy,
- cropped_scrdump_filename)
- if calced_md5sum == md5sum:
- # Success -- remove screendump history unless requested not to
- if keep_screendump_history and not keep_all_history:
- shutil.rmtree(history_dir)
- # Report success
- return True
-
- # Insert image md5sum into queue of last seen images:
- # If md5sum is already in queue...
- if whole_image_md5sum in prev_whole_image_md5sums:
- # Remove md5sum from queue
- prev_whole_image_md5sums.remove(whole_image_md5sum)
- else:
- # Otherwise extend 'stuck' timeout
- end_time_stuck = time.time() + fail_if_stuck_for
- # Insert md5sum at beginning of queue
- prev_whole_image_md5sums.insert(0, whole_image_md5sum)
- # Limit queue length to stuck_detection_history
- prev_whole_image_md5sums = \
- prev_whole_image_md5sums[:stuck_detection_history]
-
- # Sleep for a while
- time.sleep(sleep_duration)
-
- # Failure
- message = ("Barrier failed at step %s after %.2f seconds (%s)" %
- (current_step_num, time.time() - start_time, failure_message))
-
- # What should we do with this failure?
- if words[-1] == "optional":
- logging.info(message)
- return False
- else:
- # Collect information and put it in debug_dir
- if data_scrdump_filename and os.path.exists(data_scrdump_filename):
- # Read expected screendump image
- (ew, eh, edata) = \
- ppm_utils.image_read_from_ppm_file(data_scrdump_filename)
- # Write it in debug_dir
- ppm_utils.image_write_to_ppm_file(expected_scrdump_filename,
- ew, eh, edata)
- # Write the cropped version as well
- ppm_utils.get_region_md5sum(ew, eh, edata, x1, y1, dx, dy,
- expected_cropped_scrdump_filename)
- # Perform comparison
- (w, h, data) = ppm_utils.image_read_from_ppm_file(scrdump_filename)
- if w == ew and h == eh:
- (w, h, data) = ppm_utils.image_comparison(w, h, data, edata)
- ppm_utils.image_write_to_ppm_file(comparison_filename, w, h,
- data)
- # Print error messages and fail the test
- long_message = message + "\n(see analysis at %s)" % debug_dir
- logging.error(long_message)
- raise error.TestFail, message
-
-
-def run_steps(test, params, env):
- vm = kvm_utils.env_get_vm(env, params.get("main_vm"))
- if not vm:
- raise error.TestError("VM object not found in environment")
- if not vm.is_alive():
- e_msg = "VM seems to be dead. Guestwizard requires a living VM"
- raise error.TestError(e_msg)
-
- steps_filename = params.get("steps")
- if not steps_filename:
- raise error.TestError("Steps filename not specified")
- steps_filename = kvm_utils.get_path(test.bindir, steps_filename)
- if not os.path.exists(steps_filename):
- raise error.TestError("Steps file not found: %s" % steps_filename)
-
- sf = open(steps_filename, "r")
- lines = sf.readlines()
- sf.close()
-
- vm.send_monitor_cmd("cont")
-
- current_step_num = 0
- current_screendump = None
- skip_current_step = False
-
- # Iterate over the lines in the file
- for line in lines:
- line = line.strip()
- if not line:
- continue
- logging.info(line)
-
- if line.startswith("#"):
- continue
-
- words = line.split()
- if words[0] == "step":
- current_step_num += 1
- current_screendump = None
- skip_current_step = False
- elif words[0] == "screendump":
- current_screendump = words[1]
- elif skip_current_step:
- continue
- elif words[0] == "sleep":
- time.sleep(float(words[1]))
- elif words[0] == "key":
- vm.send_key(words[1])
- elif words[0] == "var":
- if not handle_var(vm, params, words[1]):
- logging.error("Variable not defined: %s" % words[1])
- elif words[0] == "barrier_2":
- if current_screendump:
- scrdump_filename = os.path.join(
- ppm_utils.get_data_dir(steps_filename),
- current_screendump)
- else:
- scrdump_filename = None
- if not barrier_2(vm, words, params, test.debugdir,
- scrdump_filename, current_step_num):
- skip_current_step = True
- else:
- vm.send_key(words[0])
diff --git a/client/tests/kvm/tests/__init__.py b/client/tests/kvm/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/client/tests/kvm/tests/build.py b/client/tests/kvm/tests/build.py
new file mode 100644
index 0000000..f6856ad
--- /dev/null
+++ b/client/tests/kvm/tests/build.py
@@ -0,0 +1,335 @@
+import time, os, sys, urllib, re, signal, logging, datetime
+from autotest_lib.client.bin import utils, test
+from autotest_lib.client.common_lib import error
+import kvm_utils
+
+
+def load_kvm_modules(module_dir):
+ """
+ Unload previously loaded kvm modules, then load modules present on any
+ sub directory of module_dir. Function will walk through module_dir until
+ it finds the modules.
+
+ @param module_dir: Directory where the KVM modules are located.
+ """
+ vendor = "intel"
+ if os.system("grep vmx /proc/cpuinfo 1>/dev/null") != 0:
+ vendor = "amd"
+ logging.debug("Detected CPU vendor as '%s'" %(vendor))
+
+ logging.debug("Killing any qemu processes that might be left behind")
+ utils.system("pkill qemu", ignore_status=True)
+
+ logging.info("Unloading previously loaded KVM modules")
+ kvm_utils.unload_module("kvm")
+ if utils.module_is_loaded("kvm"):
+ message = "Failed to remove old KVM modules"
+ logging.error(message)
+ raise error.TestError(message)
+
+ logging.info("Loading new KVM modules...")
+ kvm_module_path = None
+ kvm_vendor_module_path = None
+ # Search for the built KVM modules
+ for folder, subdirs, files in os.walk(module_dir):
+ if "kvm.ko" in files:
+ kvm_module_path = os.path.join(folder, "kvm.ko")
+ kvm_vendor_module_path = os.path.join(folder, "kvm-%s.ko" % vendor)
+ abort = False
+ if not kvm_module_path:
+ logging.error("Need a directory containing both kernel module and "
+ "userspace sources.")
+ logging.error("If you are trying to build only KVM userspace and use "
+ "the KVM modules you have already loaded, put "
+ "'load_modules': 'no' on the control file 'params' "
+ "dictionary.")
+ raise error.TestError("Could not find a built kvm.ko module on the "
+ "source dir.")
+ elif not os.path.isfile(kvm_vendor_module_path):
+ logging.error("Could not find KVM (%s) module that was supposed to be"
+ " built on the source dir", vendor)
+ abort = True
+ if abort:
+ raise error.TestError("Could not load KVM modules.")
+ utils.system("/sbin/insmod %s" % kvm_module_path)
+ time.sleep(1)
+ utils.system("/sbin/insmod %s" % kvm_vendor_module_path)
+
+ if not utils.module_is_loaded("kvm"):
+ message = "Failed to load the KVM modules built for the test"
+ logging.error(message)
+ raise error.TestError(message)
+
+
+def create_symlinks(test_bindir, prefix):
+ """
+ Create symbolic links for the appropriate qemu and qemu-img commands on
+ the kvm test bindir.
+
+ @param test_bindir: KVM test bindir
+ @param prefix: KVM prefix path
+ """
+ qemu_path = os.path.join(test_bindir, "qemu")
+ qemu_img_path = os.path.join(test_bindir, "qemu-img")
+ if os.path.lexists(qemu_path):
+ os.unlink(qemu_path)
+ if os.path.lexists(qemu_img_path):
+ os.unlink(qemu_img_path)
+ kvm_qemu = os.path.join(prefix, "bin", "qemu-system-x86_64")
+ if not os.path.isfile(kvm_qemu):
+ raise error.TestError('Invalid qemu path')
+ kvm_qemu_img = os.path.join(prefix, "bin", "qemu-img")
+ if not os.path.isfile(kvm_qemu_img):
+ raise error.TestError('Invalid qemu-img path')
+ os.symlink(kvm_qemu, qemu_path)
+ os.symlink(kvm_qemu_img, qemu_img_path)
+
+
+class SourceDirInstaller:
+ """
+ Class that handles building/installing KVM directly from a tarball or
+ a single source code dir.
+ """
+ def __init__(self, test, params):
+ """
+ Initializes class attributes, and retrieves KVM code.
+
+ @param test: kvm test object
+ @param params: Dictionary with test arguments
+ """
+ install_mode = params["mode"]
+ srcdir = params.get("srcdir")
+ # KVM build prefix
+ self.test_bindir = test.bindir
+ prefix = os.path.join(test.bindir, 'build')
+ self.prefix = os.path.abspath(prefix)
+ # Are we going to load modules?
+ load_modules = params.get('load_modules')
+ if not load_modules:
+ self.load_modules = True
+ elif load_modules == 'yes':
+ self.load_modules = True
+ elif load_modules == 'no':
+ self.load_modules = False
+
+ if install_mode == 'localsrc':
+ if not srcdir:
+ raise error.TestError("Install from source directory specified"
+ "but no source directory provided on the"
+ "control file.")
+ else:
+ self.srcdir = srcdir
+ self.repo_type = kvm_utils.check_kvm_source_dir(self.srcdir)
+ return
+ else:
+ srcdir = test.srcdir
+ if not os.path.isdir(srcdir):
+ os.makedirs(srcdir)
+
+ if install_mode == 'release':
+ release_tag = params.get("release_tag")
+ release_dir = params.get("release_dir")
+ release_listing = params.get("release_listing")
+ logging.info("Installing KVM from release tarball")
+ if not release_tag:
+ release_tag = kvm_utils.get_latest_kvm_release_tag(
+ release_listing)
+ tarball = os.path.join(release_dir, "kvm-%s.tar.gz" % release_tag)
+ logging.info("Retrieving release kvm-%s" % release_tag)
+ tarball = utils.unmap_url("/", tarball, "/tmp")
+
+ elif install_mode == 'snapshot':
+ logging.info("Installing KVM from snapshot")
+ snapshot_dir = params.get("snapshot_dir")
+ if not snapshot_dir:
+ raise error.TestError("Snapshot dir not provided")
+ snapshot_date = params.get("snapshot_date")
+ if not snapshot_date:
+ # Take yesterday's snapshot
+ d = (datetime.date.today() -
+ datetime.timedelta(1)).strftime("%Y%m%d")
+ else:
+ d = snapshot_date
+ tarball = os.path.join(snapshot_dir, "kvm-snapshot-%s.tar.gz" % d)
+ logging.info("Retrieving kvm-snapshot-%s" % d)
+ tarball = utils.unmap_url("/", tarball, "/tmp")
+
+ elif install_mode == 'localtar':
+ tarball = params.get("tarball")
+ if not tarball:
+ raise error.TestError("KVM Tarball install specified but no"
+ " tarball provided on control file.")
+ logging.info("Installing KVM from a local tarball")
+ logging.info("Using tarball %s")
+ tarball = utils.unmap_url("/", params.get("tarball"), "/tmp")
+
+ os.chdir(srcdir)
+ self.srcdir = os.path.join(srcdir, utils.extract_tarball(tarball))
+ self.repo_type = kvm_utils.check_kvm_source_dir(self.srcdir)
+
+
+ def __build(self):
+ os.chdir(self.srcdir)
+ # For testing purposes, it's better to build qemu binaries with
+ # debugging symbols, so we can extract more meaningful stack traces.
+ cfg = "./configure --disable-strip --prefix=%s" % self.prefix
+ steps = [cfg, "make clean", "make -j %s" % (utils.count_cpus() + 1)]
+ logging.info("Building KVM")
+ for step in steps:
+ utils.system(step)
+
+
+ def __install(self):
+ os.chdir(self.srcdir)
+ logging.info("Installing KVM userspace")
+ if self.repo_type == 1:
+ utils.system("make -C qemu install")
+ elif self.repo_type == 2:
+ utils.system("make install")
+ create_symlinks(self.test_bindir, self.prefix)
+
+
+ def __load_modules(self):
+ load_kvm_modules(self.srcdir)
+
+
+ def install(self):
+ self.__build()
+ self.__install()
+ if self.load_modules:
+ self.__load_modules()
+
+
+class GitInstaller:
+ def __init__(self, test, params):
+ """
+ Initialize class parameters and retrieves code from git repositories.
+
+ @param test: kvm test object.
+ @param params: Dictionary with test parameters.
+ """
+ install_mode = params["mode"]
+ srcdir = params.get("srcdir", test.bindir)
+ if not srcdir:
+ os.makedirs(srcdir)
+ self.srcdir = srcdir
+ # KVM build prefix
+ self.test_bindir = test.bindir
+ prefix = os.path.join(test.bindir, 'build')
+ self.prefix = os.path.abspath(prefix)
+ # Are we going to load modules?
+ load_modules = params.get('load_modules')
+ if not load_modules:
+ self.load_modules = True
+ elif load_modules == 'yes':
+ self.load_modules = True
+ elif load_modules == 'no':
+ self.load_modules = False
+
+ kernel_repo = params.get("git_repo")
+ user_repo = params.get("user_git_repo")
+ kmod_repo = params.get("kmod_repo")
+
+ branch = params.get("git_branch", "master")
+ lbranch = params.get("lbranch")
+
+ tag = params.get("git_tag", "HEAD")
+ user_tag = params.get("user_git_tag", "HEAD")
+ kmod_tag = params.get("kmod_git_tag", "HEAD")
+
+ if not kernel_repo:
+ message = "KVM git repository path not specified"
+ logging.error(message)
+ raise error.TestError(message)
+ if not user_repo:
+ message = "KVM user git repository path not specified"
+ logging.error(message)
+ raise error.TestError(message)
+
+ kernel_srcdir = os.path.join(srcdir, "kvm")
+ kvm_utils.get_git_branch(kernel_repo, branch, kernel_srcdir, tag,
+ lbranch)
+ self.kernel_srcdir = kernel_srcdir
+
+ userspace_srcdir = os.path.join(srcdir, "kvm_userspace")
+ kvm_utils.get_git_branch(user_repo, branch, userspace_srcdir, user_tag,
+ lbranch)
+ self.userspace_srcdir = userspace_srcdir
+
+ if kmod_repo:
+ kmod_srcdir = os.path.join (srcdir, "kvm_kmod")
+ kvm_utils.get_git_branch(kmod_repo, branch, kmod_srcdir, user_tag,
+ lbranch)
+ self.kmod_srcdir = kmod_srcdir
+
+
+ def __build(self):
+ if self.kmod_srcdir:
+ logging.info('Building KVM modules')
+ os.chdir(self.kmod_srcdir)
+ utils.system('./configure')
+ utils.system('make clean')
+ utils.system('make sync LINUX=%s' % self.kernel_srcdir)
+ utils.system('make -j %s' % utils.count_cpus())
+ logging.info('Building KVM userspace code')
+ os.chdir(self.userspace_srcdir)
+ utils.system('./configure --disable-strip --prefix=%s' %
+ self.prefix)
+ utils.system('make clean')
+ utils.system('make -j %s' % utils.count_cpus())
+ else:
+ os.chdir(self.userspace_srcdir)
+ utils.system('./configure --disable-strip --prefix=%s' %
+ self.prefix)
+ logging.info('Building KVM modules')
+ utils.system('make clean')
+ utils.system('make -C kernel LINUX=%s sync' % self.kernel_srcdir)
+ logging.info('Building KVM userspace code')
+ utils.system('make -j %s' % utils.count_cpus())
+
+
+ def __install(self):
+ os.chdir(self.userspace_srcdir)
+ utils.system('make install')
+ create_symlinks(self.test_bindir, self.prefix)
+
+
+ def __load_modules(self):
+ if self.kmod_srcdir:
+ load_kvm_modules(self.kmod_srcdir)
+ else:
+ load_kvm_modules(self.userspace_srcdir)
+
+
+ def install(self):
+ self.__build()
+ self.__install()
+ if self.load_modules:
+ self.__load_modules()
+
+
+def run_build(test, params, env):
+ """
+ Installs KVM using the selected install mode. Most install methods will
+ take kvm source code, build it and install it to a given location.
+
+ @param test: kvm test object.
+ @param params: Dictionary with test parameters.
+ @param env: Test environment.
+ """
+ install_mode = params.get("mode")
+ srcdir = params.get("srcdir", test.srcdir)
+ params["srcdir"] = srcdir
+
+ if install_mode == 'noinstall':
+ logging.info("Skipping installation")
+ return
+ elif install_mode in ['localsrc', 'localtar', 'release', 'snapshot']:
+ installer = SourceDirInstaller(test, params)
+ elif install_mode == 'git':
+ installer = GitInstaller(test, params)
+ else:
+ raise error.TestError('Invalid or unsupported'
+ ' install mode: %s' % install_mode)
+
+ installer.install()
diff --git a/client/tests/kvm/tests/stepmaker.py b/client/tests/kvm/tests/stepmaker.py
new file mode 100755
index 0000000..24060db
--- /dev/null
+++ b/client/tests/kvm/tests/stepmaker.py
@@ -0,0 +1,357 @@
+#!/usr/bin/python
+"""
+Step file creator/editor.
+
+@copyright: Red Hat Inc 2009
+@author: mgoldish@xxxxxxxxxx (Michael Goldish)
+@version: "20090401"
+"""
+
+import pygtk, gtk, gobject, time, os, commands
+import common
+from autotest_lib.client.common_lib import error
+import kvm_utils, logging, ppm_utils, stepeditor
+pygtk.require('2.0')
+
+
+class StepMaker(stepeditor.StepMakerWindow):
+ """
+ Application used to create a step file. It will grab your input to the
+ virtual machine and record it on a 'step file', that can be played
+ making it possible to do unattended installs.
+ """
+ # Constructor
+ def __init__(self, vm, steps_filename, tempdir, params):
+ stepeditor.StepMakerWindow.__init__(self)
+
+ self.vm = vm
+ self.steps_filename = steps_filename
+ self.steps_data_dir = ppm_utils.get_data_dir(steps_filename)
+ self.tempdir = tempdir
+ self.screendump_filename = os.path.join(tempdir, "scrdump.ppm")
+ self.params = params
+
+ if not os.path.exists(self.steps_data_dir):
+ os.makedirs(self.steps_data_dir)
+
+ self.steps_file = open(self.steps_filename, "w")
+ self.vars_file = open(os.path.join(self.steps_data_dir, "vars"), "w")
+
+ self.step_num = 1
+ self.run_time = 0
+ self.update_delay = 1000
+ self.prev_x = 0
+ self.prev_y = 0
+ self.vars = {}
+ self.timer_id = None
+
+ self.time_when_done_clicked = time.time()
+ self.time_when_actions_completed = time.time()
+
+ self.steps_file.write("# Generated by Step Maker\n")
+ self.steps_file.write("# Generated on %s\n" % time.asctime())
+ self.steps_file.write("# uname -a: %s\n" %
+ commands.getoutput("uname -a"))
+ self.steps_file.flush()
+
+ self.vars_file.write("# This file lists the vars used during recording"
+ " with Step Maker\n")
+ self.vars_file.flush()
+
+ # Done/Break HBox
+ hbox = gtk.HBox(spacing=10)
+ self.user_vbox.pack_start(hbox)
+ hbox.show()
+
+ self.button_break = gtk.Button("Break")
+ self.button_break.connect("clicked", self.event_break_clicked)
+ hbox.pack_start(self.button_break)
+ self.button_break.show()
+
+ self.button_done = gtk.Button("Done")
+ self.button_done.connect("clicked", self.event_done_clicked)
+ hbox.pack_start(self.button_done)
+ self.button_done.show()
+
+ # Set window title
+ self.window.set_title("Step Maker")
+
+ # Connect "capture" button
+ self.button_capture.connect("clicked", self.event_capture_clicked)
+
+ # Switch to run mode
+ self.switch_to_run_mode()
+
+
+ def destroy(self, widget):
+ self.vm.send_monitor_cmd("cont")
+ self.steps_file.close()
+ self.vars_file.close()
+ stepeditor.StepMakerWindow.destroy(self, widget)
+
+
+ # Utilities
+ def redirect_timer(self, delay=0, func=None):
+ if self.timer_id != None:
+ gobject.source_remove(self.timer_id)
+ self.timer_id = None
+ if func != None:
+ self.timer_id = gobject.timeout_add(delay, func,
+ priority=gobject.PRIORITY_LOW)
+
+
+ def switch_to_run_mode(self):
+ # Set all widgets to their default states
+ self.clear_state(clear_screendump=False)
+ # Enable/disable some widgets
+ self.button_break.set_sensitive(True)
+ self.button_done.set_sensitive(False)
+ self.data_vbox.set_sensitive(False)
+ # Give focus to the Break button
+ self.button_break.grab_focus()
+ # Start the screendump timer
+ self.redirect_timer(100, self.update)
+ # Resume the VM
+ self.vm.send_monitor_cmd("cont")
+
+
+ def switch_to_step_mode(self):
+ # Set all widgets to their default states
+ self.clear_state(clear_screendump=False)
+ # Enable/disable some widgets
+ self.button_break.set_sensitive(False)
+ self.button_done.set_sensitive(True)
+ self.data_vbox.set_sensitive(True)
+ # Give focus to the keystrokes entry widget
+ self.entry_keys.grab_focus()
+ # Start the screendump timer
+ self.redirect_timer()
+ # Stop the VM
+ self.vm.send_monitor_cmd("stop")
+
+
+ # Events in step mode
+ def update(self):
+ self.redirect_timer()
+
+ if os.path.exists(self.screendump_filename):
+ os.unlink(self.screendump_filename)
+
+ (status, output) = self.vm.send_monitor_cmd("screendump " +
+ self.screendump_filename)
+ if status: # Failure
+ logging.info("Could not fetch screendump")
+ else:
+ self.set_image_from_file(self.screendump_filename)
+
+ self.redirect_timer(self.update_delay, self.update)
+ return True
+
+
+ def event_break_clicked(self, widget):
+ if not self.vm.is_alive():
+ self.message("The VM doesn't seem to be alive.", "Error")
+ return
+ # Switch to step mode
+ self.switch_to_step_mode()
+ # Compute time elapsed since last click on "Done" and add it
+ # to self.run_time
+ self.run_time += time.time() - self.time_when_done_clicked
+ # Set recording time widget
+ self.entry_time.set_text("%.2f" % self.run_time)
+ # Update screendump ID
+ self.update_screendump_id(self.steps_data_dir)
+ # By default, check the barrier checkbox
+ self.check_barrier.set_active(True)
+ # Set default sleep and barrier timeout durations
+ time_delta = time.time() - self.time_when_actions_completed
+ if time_delta < 1.0: time_delta = 1.0
+ self.spin_sleep.set_value(round(time_delta))
+ self.spin_barrier_timeout.set_value(round(time_delta * 5))
+ # Set window title
+ self.window.set_title("Step Maker -- step %d at time %.2f" %
+ (self.step_num, self.run_time))
+
+
+ def event_done_clicked(self, widget):
+ # Get step lines and screendump
+ lines = self.get_step_lines(self.steps_data_dir)
+ if lines == None:
+ return
+
+ # Get var values from user and write them to vars file
+ vars = {}
+ for line in lines.splitlines():
+ words = line.split()
+ if words and words[0] == "var":
+ varname = words[1]
+ if varname in self.vars.keys():
+ val = self.vars[varname]
+ elif varname in vars.keys():
+ val = vars[varname]
+ elif varname in self.params.keys():
+ val = self.params[varname]
+ vars[varname] = val
+ else:
+ val = self.inputdialog("$%s =" % varname, "Variable")
+ if val == None:
+ return
+ vars[varname] = val
+ for varname in vars.keys():
+ self.vars_file.write("%s=%s\n" % (varname, vars[varname]))
+ self.vars.update(vars)
+
+ # Write step lines to file
+ self.steps_file.write("# " + "-" * 32 + "\n")
+ self.steps_file.write(lines)
+
+ # Flush buffers of both files
+ self.steps_file.flush()
+ self.vars_file.flush()
+
+ # Remember the current time
+ self.time_when_done_clicked = time.time()
+
+ # Switch to run mode
+ self.switch_to_run_mode()
+
+ # Send commands to VM
+ for line in lines.splitlines():
+ words = line.split()
+ if not words:
+ continue
+ elif words[0] == "key":
+ self.vm.send_key(words[1])
+ elif words[0] == "var":
+ val = self.vars.get(words[1])
+ if not val:
+ continue
+ self.vm.send_string(val)
+ elif words[0] == "mousemove":
+ self.vm.send_monitor_cmd("mouse_move %d %d" % (-8000,-8000))
+ time.sleep(0.5)
+ self.vm.send_monitor_cmd("mouse_move %s %s" % (words[1],
+ words[2]))
+ time.sleep(0.5)
+ elif words[0] == "mouseclick":
+ self.vm.send_monitor_cmd("mouse_button %s" % words[1])
+ time.sleep(0.1)
+ self.vm.send_monitor_cmd("mouse_button 0")
+
+ # Remember the current time
+ self.time_when_actions_completed = time.time()
+
+ # Move on to next step
+ self.step_num += 1
+
+ def event_capture_clicked(self, widget):
+ self.message("Mouse actions disabled (for now).", "Sorry")
+ return
+
+ self.image_width_backup = self.image_width
+ self.image_height_backup = self.image_height
+ self.image_data_backup = self.image_data
+
+ gtk.gdk.pointer_grab(self.event_box.window, False,
+ gtk.gdk.BUTTON_PRESS_MASK |
+ gtk.gdk.BUTTON_RELEASE_MASK)
+ # Create empty cursor
+ pix = gtk.gdk.Pixmap(self.event_box.window, 1, 1, 1)
+ color = gtk.gdk.Color()
+ cursor = gtk.gdk.Cursor(pix, pix, color, color, 0, 0)
+ self.event_box.window.set_cursor(cursor)
+ gtk.gdk.display_get_default().warp_pointer(gtk.gdk.screen_get_default(),
+ self.prev_x, self.prev_y)
+ self.redirect_event_box_input(
+ self.event_capture_button_press,
+ self.event_capture_button_release,
+ self.event_capture_scroll)
+ self.redirect_timer(10, self.update_capture)
+ self.vm.send_monitor_cmd("cont")
+
+ # Events in mouse capture mode
+
+ def update_capture(self):
+ self.redirect_timer()
+
+ (screen, x, y, flags) = gtk.gdk.display_get_default().get_pointer()
+ self.mouse_click_coords[0] = int(x * self.spin_sensitivity.get_value())
+ self.mouse_click_coords[1] = int(y * self.spin_sensitivity.get_value())
+
+ delay = self.spin_latency.get_value() / 1000
+ if (x, y) != (self.prev_x, self.prev_y):
+ self.vm.send_monitor_cmd("mouse_move %d %d" % (-8000, -8000))
+ time.sleep(delay)
+ self.vm.send_monitor_cmd("mouse_move %d %d" %
+ (self.mouse_click_coords[0],
+ self.mouse_click_coords[1]))
+ time.sleep(delay)
+
+ self.prev_x = x
+ self.prev_y = y
+
+ if os.path.exists(self.screendump_filename):
+ os.unlink(self.screendump_filename)
+
+ (status, output) = self.vm.send_monitor_cmd("screendump " +
+ self.screendump_filename)
+ if status: # Failure
+ logging.info("Could not fetch screendump")
+ else:
+ self.set_image_from_file(self.screendump_filename)
+
+ self.redirect_timer(int(self.spin_latency.get_value()),
+ self.update_capture)
+ return True
+
+ def event_capture_button_press(self, widget,event):
+ pass
+
+ def event_capture_button_release(self, widget,event):
+ gtk.gdk.pointer_ungrab()
+ self.event_box.window.set_cursor(gtk.gdk.Cursor(gtk.gdk.CROSSHAIR))
+ self.redirect_event_box_input(
+ self.event_button_press,
+ self.event_button_release,
+ None,
+ None,
+ self.event_expose)
+ self.redirect_timer()
+ self.vm.send_monitor_cmd("stop")
+ self.mouse_click_captured = True
+ self.mouse_click_button = event.button
+ self.set_image(self.image_width_backup, self.image_height_backup,
+ self.image_data_backup)
+ self.check_mousemove.set_sensitive(True)
+ self.check_mouseclick.set_sensitive(True)
+ self.check_mousemove.set_active(True)
+ self.check_mouseclick.set_active(True)
+ self.update_mouse_click_info()
+
+ def event_capture_scroll(self, widget, event):
+ if event.direction == gtk.gdk.SCROLL_UP:
+ direction = 1
+ else:
+ direction = -1
+ self.spin_sensitivity.set_value(self.spin_sensitivity.get_value() +
+ direction)
+ pass
+
+
+def run_stepmaker(test, params, env):
+ vm = kvm_utils.env_get_vm(env, params.get("main_vm"))
+ if not vm:
+ raise error.TestError("VM object not found in environment")
+ if not vm.is_alive():
+ raise error.TestError("VM seems to be dead; Step Maker requires a"
+ " living VM")
+
+ steps_filename = params.get("steps")
+ if not steps_filename:
+ raise error.TestError("Steps filename not specified")
+ steps_filename = kvm_utils.get_path(test.bindir, steps_filename)
+ if os.path.exists(steps_filename):
+ raise error.TestError("Steps file %s already exists" % steps_filename)
+
+ StepMaker(vm, steps_filename, test.debugdir, params)
+ gtk.main()
diff --git a/client/tests/kvm/tests/steps.py b/client/tests/kvm/tests/steps.py
new file mode 100644
index 0000000..8bc85f2
--- /dev/null
+++ b/client/tests/kvm/tests/steps.py
@@ -0,0 +1,249 @@
+"""
+Utilities to perform automatic guest installation using step files.
+
+@copyright: Red Hat 2008-2009
+"""
+
+import os, time, md5, re, shutil, logging
+from autotest_lib.client.common_lib import utils, error
+import kvm_utils, ppm_utils, kvm_subprocess
+try:
+ import PIL.Image
+except ImportError:
+ logging.warning('No python imaging library installed. PPM image '
+ 'conversion to JPEG disabled. In order to enable it, '
+ 'please install python-imaging or the equivalent for your '
+ 'distro.')
+
+
+def handle_var(vm, params, varname):
+ var = params.get(varname)
+ if not var:
+ return False
+ vm.send_string(var)
+ return True
+
+
+def barrier_2(vm, words, params, debug_dir, data_scrdump_filename,
+ current_step_num):
+ if len(words) < 7:
+ logging.error("Bad barrier_2 command line")
+ return False
+
+ cmd, dx, dy, x1, y1, md5sum, timeout = words[:7]
+ dx, dy, x1, y1, timeout = map(int, [dx, dy, x1, y1, timeout])
+
+ # Timeout/5 is the time it took stepmaker to complete this step.
+ # Divide that number by 10 to poll 10 times, just in case
+ # current machine is stronger then the "stepmaker machine".
+ # Limit to 1 (min) and 10 (max) seconds between polls.
+ sleep_duration = float(timeout) / 50.0
+ if sleep_duration < 1.0: sleep_duration = 1.0
+ if sleep_duration > 10.0: sleep_duration = 10.0
+
+ scrdump_filename = os.path.join(debug_dir, "scrdump.ppm")
+ cropped_scrdump_filename = os.path.join(debug_dir, "cropped_scrdump.ppm")
+ expected_scrdump_filename = os.path.join(debug_dir, "scrdump_expected.ppm")
+ expected_cropped_scrdump_filename = os.path.join(debug_dir,
+ "cropped_scrdump_expected.ppm")
+ comparison_filename = os.path.join(debug_dir, "comparison.ppm")
+
+ fail_if_stuck_for = params.get("fail_if_stuck_for")
+ if fail_if_stuck_for:
+ fail_if_stuck_for = float(fail_if_stuck_for)
+ else:
+ fail_if_stuck_for = 1e308
+
+ stuck_detection_history = params.get("stuck_detection_history")
+ if stuck_detection_history:
+ stuck_detection_history = int(stuck_detection_history)
+ else:
+ stuck_detection_history = 2
+
+ keep_screendump_history = params.get("keep_screendump_history") == "yes"
+ if keep_screendump_history:
+ keep_all_history = params.get("keep_all_history") == "yes"
+ history_dir = os.path.join(debug_dir, "barrier_history")
+
+ end_time = time.time() + timeout
+ end_time_stuck = time.time() + fail_if_stuck_for
+ start_time = time.time()
+
+ prev_whole_image_md5sums = []
+
+ failure_message = None
+
+ # Main loop
+ while True:
+ # Check for timeouts
+ if time.time() > end_time:
+ failure_message = "regular timeout"
+ break
+ if time.time() > end_time_stuck:
+ failure_message = "guest is stuck"
+ break
+
+ # Make sure vm is alive
+ if not vm.is_alive():
+ failure_message = "VM is dead"
+ break
+
+ # Request screendump
+ (status, output) = vm.send_monitor_cmd("screendump %s" %
+ scrdump_filename)
+ if status:
+ logging.error("Could not fetch screendump")
+ continue
+
+ # Read image file
+ (w, h, data) = ppm_utils.image_read_from_ppm_file(scrdump_filename)
+
+ # Make sure image is valid
+ if not ppm_utils.image_verify_ppm_file(scrdump_filename):
+ logging.warn("Got invalid screendump: dimensions: %dx%d, "
+ "data size: %d" % (w, h, len(data)))
+ continue
+
+ # Compute md5sum of whole image
+ whole_image_md5sum = ppm_utils.image_md5sum(w, h, data)
+
+ # Write screendump to history_dir (as JPG) if requested
+ # and if the screendump differs from the previous one
+ if (keep_screendump_history and
+ whole_image_md5sum not in prev_whole_image_md5sums[:1]):
+ try:
+ os.makedirs(history_dir)
+ except:
+ pass
+ history_scrdump_filename = os.path.join(history_dir,
+ "scrdump-step_%s-%s.jpg" % (current_step_num,
+ time.strftime("%Y%m%d-%H%M%S")))
+ try:
+ image = PIL.Image.open(scrdump_filename)
+ image.save(history_scrdump_filename, format = 'JPEG',
+ quality = 30)
+ except NameError:
+ pass
+
+ # Compare md5sum of barrier region with the expected md5sum
+ calced_md5sum = ppm_utils.get_region_md5sum(w, h, data, x1, y1, dx, dy,
+ cropped_scrdump_filename)
+ if calced_md5sum == md5sum:
+ # Success -- remove screendump history unless requested not to
+ if keep_screendump_history and not keep_all_history:
+ shutil.rmtree(history_dir)
+ # Report success
+ return True
+
+ # Insert image md5sum into queue of last seen images:
+ # If md5sum is already in queue...
+ if whole_image_md5sum in prev_whole_image_md5sums:
+ # Remove md5sum from queue
+ prev_whole_image_md5sums.remove(whole_image_md5sum)
+ else:
+ # Otherwise extend 'stuck' timeout
+ end_time_stuck = time.time() + fail_if_stuck_for
+ # Insert md5sum at beginning of queue
+ prev_whole_image_md5sums.insert(0, whole_image_md5sum)
+ # Limit queue length to stuck_detection_history
+ prev_whole_image_md5sums = \
+ prev_whole_image_md5sums[:stuck_detection_history]
+
+ # Sleep for a while
+ time.sleep(sleep_duration)
+
+ # Failure
+ message = ("Barrier failed at step %s after %.2f seconds (%s)" %
+ (current_step_num, time.time() - start_time, failure_message))
+
+ # What should we do with this failure?
+ if words[-1] == "optional":
+ logging.info(message)
+ return False
+ else:
+ # Collect information and put it in debug_dir
+ if data_scrdump_filename and os.path.exists(data_scrdump_filename):
+ # Read expected screendump image
+ (ew, eh, edata) = \
+ ppm_utils.image_read_from_ppm_file(data_scrdump_filename)
+ # Write it in debug_dir
+ ppm_utils.image_write_to_ppm_file(expected_scrdump_filename,
+ ew, eh, edata)
+ # Write the cropped version as well
+ ppm_utils.get_region_md5sum(ew, eh, edata, x1, y1, dx, dy,
+ expected_cropped_scrdump_filename)
+ # Perform comparison
+ (w, h, data) = ppm_utils.image_read_from_ppm_file(scrdump_filename)
+ if w == ew and h == eh:
+ (w, h, data) = ppm_utils.image_comparison(w, h, data, edata)
+ ppm_utils.image_write_to_ppm_file(comparison_filename, w, h,
+ data)
+ # Print error messages and fail the test
+ long_message = message + "\n(see analysis at %s)" % debug_dir
+ logging.error(long_message)
+ raise error.TestFail, message
+
+
+def run_steps(test, params, env):
+ vm = kvm_utils.env_get_vm(env, params.get("main_vm"))
+ if not vm:
+ raise error.TestError("VM object not found in environment")
+ if not vm.is_alive():
+ e_msg = "VM seems to be dead. Guestwizard requires a living VM"
+ raise error.TestError(e_msg)
+
+ steps_filename = params.get("steps")
+ if not steps_filename:
+ raise error.TestError("Steps filename not specified")
+ steps_filename = kvm_utils.get_path(test.bindir, steps_filename)
+ if not os.path.exists(steps_filename):
+ raise error.TestError("Steps file not found: %s" % steps_filename)
+
+ sf = open(steps_filename, "r")
+ lines = sf.readlines()
+ sf.close()
+
+ vm.send_monitor_cmd("cont")
+
+ current_step_num = 0
+ current_screendump = None
+ skip_current_step = False
+
+ # Iterate over the lines in the file
+ for line in lines:
+ line = line.strip()
+ if not line:
+ continue
+ logging.info(line)
+
+ if line.startswith("#"):
+ continue
+
+ words = line.split()
+ if words[0] == "step":
+ current_step_num += 1
+ current_screendump = None
+ skip_current_step = False
+ elif words[0] == "screendump":
+ current_screendump = words[1]
+ elif skip_current_step:
+ continue
+ elif words[0] == "sleep":
+ time.sleep(float(words[1]))
+ elif words[0] == "key":
+ vm.send_key(words[1])
+ elif words[0] == "var":
+ if not handle_var(vm, params, words[1]):
+ logging.error("Variable not defined: %s" % words[1])
+ elif words[0] == "barrier_2":
+ if current_screendump:
+ scrdump_filename = os.path.join(
+ ppm_utils.get_data_dir(steps_filename),
+ current_screendump)
+ else:
+ scrdump_filename = None
+ if not barrier_2(vm, words, params, test.debugdir,
+ scrdump_filename, current_step_num):
+ skip_current_step = True
+ else:
+ vm.send_key(words[0])
--
1.6.2.5
--
To unsubscribe from this list: send the line "unsubscribe kvm" in
the body of a message to majordomo@xxxxxxxxxxxxxxx
More majordomo info at http://vger.kernel.org/majordomo-info.html
