htest/tasks.py

   1 #!/usr/bin/env python3
   2
   3 #
   4 # Copyright (c) 2018 Vojtech Horky
   5 # All rights reserved.
   6 #
   7 # Redistribution and use in source and binary forms, with or without
   8 # modification, are permitted provided that the following conditions
   9 # are met:
  10 #
  11 # - Redistributions of source code must retain the above copyright
  12 #   notice, this list of conditions and the following disclaimer.
  13 # - Redistributions in binary form must reproduce the above copyright
  14 #   notice, this list of conditions and the following disclaimer in the
  15 #   documentation and/or other materials provided with the distribution.
  16 # - The name of the author may not be used to endorse or promote products
  17 #   derived from this software without specific prior written permission.
  18 #
  19 # THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  20 # IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  21 # OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  22 # IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  23 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  24 # NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  25 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  26 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  27 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  28 # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29 #
  30
  31 import logging
  32
  33 from htest.utils import retries
  34
  35 class ScenarioTask:
  36     """
  37     Base class for individual tasks that are executed in a scenario.
  38     """
  39
  40     def __init__(self, name):
  41         """
  42         Set @name to the task name (call from subclass).
  43         """
  44
  45         self.name = name
  46         self.machine = None
  47         self.fail_message = ''
  48         self.logger = logging.getLogger(name)
  49
  50     def check_required_argument(self, args, name):
  51         """
  52         To be used by subclasses to check that arguments
  53         were specified.
  54         """
  55         if not name in args:
  56             raise Exception("Required argument {} missing.".format(name))
  57
  58     def is_vm_launcher(self):
  59         """
  60         Whether this task starts a new VM.
  61         """
  62         return False
  63
  64     def get_name(self):
  65         return self.name
  66
  67     def set_machine(self, machine):
  68         """
  69         Set machine responsible for executing this task.
  70         """
  71         self.machine = machine
  72
  73     def run(self):
  74         """
  75         Actually execute this task.
  76         """
  77         pass
  78
  79 class ScenarioTaskBoot(ScenarioTask):
  80     """
  81     Brings the machine up.
  82     """
  83     def __init__(self, args):
  84         ScenarioTask.__init__(self, 'boot')
  85         self.args = args
  86
  87     def is_vm_launcher(self):
  88         return True
  89
  90     def run(self):
  91         self.machine.boot()
  92
  93 class ScenarioTaskCommand(ScenarioTask):
  94     """
  95     Run a command in vterm.
  96     """
  97
  98     def __init__(self, args):
  99         ScenarioTask.__init__(self, 'command')
 100         if type(args) is str:
 101             args = { 'args': args}
 102         self.check_required_argument(args, 'args')
 103         self.command = args['args']
 104         self.ignore_abort = False
 105         if 'ignoreabort' in args:
 106             self.ignore_abort = args['ignoreabort']
 107         self.args = args
 108
 109     def _find_in_lines(self, text, lines):
 110         for l in lines:
 111             if l.find(text) != -1:
 112                 return True
 113         return False
 114
 115     def run(self):
 116         self.logger.info("Typing '{}' into {}.".format(self.command, self.machine.name))
 117
 118         # Capture the screen before typing the command.
 119         self.machine.capture_vterm()
 120
 121         self.machine.type(self.command)
 122
 123         # Wait until the command is fully displayed on the screen.
 124         # That is needed to properly detect the newly displayed lines.
 125         # FIXME: this will not work for long commands spanning multiple lines
 126         for xxx in retries(timeout=60, interval=2, name="vterm-type", message="Failed to type command"):
 127             self.machine.vterm = []
 128             self.machine.capture_vterm()
 129             lines = self.machine.vterm
 130
 131             if len(lines) > 0:
 132                 line = lines[0].strip()
 133                 if line.endswith("_"):
 134                     line = line[0:-1]
 135                 if line.endswith(self.command.strip()):
 136                     break
 137
 138         self.machine.vterm = []
 139         self.machine.type('\n')
 140
 141         # Read output of the command.
 142         # We wait until prompt reappears or we find some text that is not
 143         # supposed to be there.
 144         for xxx in retries(timeout=60, interval=2, name="vterm-run", message="Failed to run command"):
 145             self.logger.debug("self.vterm = {}".format(self.machine.vterm))
 146             self.machine.capture_vterm()
 147             lines = self.machine.vterm
 148             self.logger.debug("Read lines {}".format(lines))
 149             self.machine.vterm = []
 150             if not self.ignore_abort:
 151                 if self._find_in_lines('Cannot spawn', lines) or self._find_in_lines('Command failed', lines):
 152                     raise Exception('Failed to run command')
 153             if 'negassert' in self.args:
 154                 if self._find_in_lines(self.args['negassert'], lines):
 155                     raise Exception('Found forbidden text {} ...'.format(self.args['negassert']))
 156             if self._find_in_lines('# _', lines):
 157                 if 'assert' in self.args:
 158                     if not self._find_in_lines(self.args['assert'], lines):
 159                         raise Exception('Missing expected text {} ...'.format(self.args['assert']))
 160                 break
 161         self.logger.info("Command '{}' done.".format(self.command))
 162
 163 class ScenarioTaskCls(ScenarioTask):
 164     """
 165     Clear vterm screen.
 166     """
 167
 168     def __init__(self, args):
 169         ScenarioTask.__init__(self, 'vterm-cls')
 170
 171     def run(self):
 172         self.logger.info("Clearing the screen.")
 173
 174         for i in range(30):
 175             self.machine.type('\n')