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 import re
  33
  34 from htest.utils import retries
  35
  36 class ScenarioTask:
  37     """
  38     Base class for individual tasks that are executed in a scenario.
  39     """
  40
  41     def __init__(self, name):
  42         """
  43         Set @name to the task name (call from subclass).
  44         """
  45
  46         self.name = name
  47         self.machine = None
  48         self.fail_message = ''
  49         self.logger = logging.getLogger(name)
  50
  51     def check_required_argument(self, args, name):
  52         """
  53         To be used by subclasses to check that arguments
  54         were specified.
  55         """
  56         if not name in args:
  57             raise Exception("Required argument {} missing.".format(name))
  58
  59     def is_vm_launcher(self):
  60         """
  61         Whether this task starts a new VM.
  62         """
  63         return False
  64
  65     def get_name(self):
  66         return self.name
  67
  68     def set_machine(self, machine):
  69         """
  70         Set machine responsible for executing this task.
  71         """
  72         self.machine = machine
  73
  74     def run(self):
  75         """
  76         Actually execute this task.
  77         """
  78         pass
  79
  80 class ScenarioTaskBoot(ScenarioTask):
  81     """
  82     Brings the machine up.
  83     """
  84     def __init__(self, args):
  85         ScenarioTask.__init__(self, 'boot')
  86         self.args = args
  87
  88     def is_vm_launcher(self):
  89         return True
  90
  91     def run(self):
  92         self.machine.boot()
  93
  94 class ScenarioTaskCommand(ScenarioTask):
  95     """
  96     Run a command in vterm.
  97     """
  98
  99     def __init__(self, args):
 100         ScenarioTask.__init__(self, 'command')
 101         if type(args) is str:
 102             args = { 'args': args}
 103         self.check_required_argument(args, 'args')
 104         self.command = args['args']
 105         self.ignore_abort = False
 106         if 'ignoreabort' in args:
 107             self.ignore_abort = args['ignoreabort']
 108         self.args = args
 109
 110     def _find_in_lines(self, text, lines):
 111         for l in lines:
 112             if l.find(text) != -1:
 113                 return True
 114         return False
 115
 116     def _grep_in_lines(self, comp_re, lines):
 117         for l in lines:
 118             if comp_re.search(l) is not None:
 119                 return True
 120         return False
 121
 122     def run(self):
 123         self.logger.info("Typing '{}' into {}.".format(self.command, self.machine.name))
 124
 125         cursor_symbol = self.machine.get_vterm_cursor_symbol()
 126         cursor_symbol_spaced = '' if cursor_symbol == '' else ' ' + cursor_symbol
 127         prompt_re = re.compile('^/[^ ]* #' + re.escape(cursor_symbol_spaced) + '[\t ]*$')
 128         self.logger.debug("RE for prompt matching: {}".format(prompt_re))
 129
 130         # Capture the screen before typing the command.
 131         self.machine.capture_vterm()
 132
 133         self.machine.type(self.command)
 134
 135         # Wait until the command is fully displayed on the screen.
 136         # That is needed to properly detect the newly displayed lines.
 137         # FIXME: this will not work for long commands spanning multiple lines
 138         for xxx in retries(timeout=60, interval=2, name="vterm-type", message="Failed to type command"):
 139             self.machine.vterm = []
 140             self.machine.capture_vterm()
 141             lines = self.machine.vterm
 142
 143             if len(lines) > 0:
 144                 line = lines[0].strip()
 145
 146                 if (cursor_symbol != '') and line.endswith(cursor_symbol):
 147                     line = line[0:-(len(cursor_symbol))]
 148                 if line.endswith(self.command.strip()):
 149                     break
 150
 151         self.machine.vterm = []
 152         self.machine.type('\n')
 153
 154         # Read output of the command.
 155         # We wait until prompt reappears or we find some text that is not
 156         # supposed to be there. Meanwhile we check that the text that is
 157         # supposed to be there appears.
 158         asserted_text_found = not 'assert' in self.args
 159         for xxx in retries(timeout=60, interval=2, name="vterm-run", message="Failed to run command"):
 160             self.logger.debug("self.vterm = {}".format(self.machine.vterm))
 161             self.machine.capture_vterm()
 162             lines = self.machine.vterm
 163             self.logger.debug("Read lines {}".format(lines))
 164             self.machine.vterm = []
 165             if not self.ignore_abort:
 166                 if self._find_in_lines('Cannot spawn', lines) or self._find_in_lines('Command failed', lines):
 167                     raise Exception('Failed to run command')
 168             if 'negassert' in self.args:
 169                 if self._find_in_lines(self.args['negassert'], lines):
 170                     raise Exception('Found forbidden text {} ...'.format(self.args['negassert']))
 171             if ('assert' in self.args) and (not asserted_text_found):
 172                 asserted_text_found = self._find_in_lines(self.args['assert'], lines)
 173             if self._grep_in_lines(prompt_re, lines):
 174                 if not asserted_text_found:
 175                     raise Exception('Missing expected text {} ...'.format(self.args['assert']))
 176                 break
 177         self.logger.info("Command '{}' done.".format(self.command))
 178
 179 class ScenarioTaskCls(ScenarioTask):
 180     """
 181     Clear vterm screen.
 182     """
 183
 184     def __init__(self, args):
 185         ScenarioTask.__init__(self, 'vterm-cls')
 186
 187     def run(self):
 188         self.logger.info("Clearing the screen.")
 189
 190         for i in range(30):
 191             self.machine.type('\n')