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.args = args
 105
 106     def _grep(self, text, lines):
 107         for l in lines:
 108             if l.find(text) != -1:
 109                 return True
 110         return False
 111
 112     def run(self):
 113         self.logger.info("Typing '{}' into {}.".format(self.command, self.machine.name))
 114
 115         self.machine.type(self.command)
 116         self.machine.type('\n')
 117
 118         for xxx in retries(timeout=60, interval=2, name="vterm-run", message="Failed to run command"):
 119             lines = self.machine.capture_vterm()
 120             if 'negassert' in self.args:
 121                 if self._grep(self.args['negassert'], lines):
 122                     raise Exception('Found forbidden text {} ...'.format(self.args['negassert']))
 123             if 'assert' in self.args:
 124                 if self._grep(self.args['assert'], lines):
 125                     break
 126             if self._grep('Cannot spawn', lines) or self._grep('Command failed', lines):
 127                 raise Exception('Failed to run command')
 128             if self._grep('# _', lines):
 129                 if 'assert' in self.args:
 130                     raise Exception('Missing expected text {} ...'.format(self.args['assert']))
 131                 break
 132         self.logger.info("Command '{}' done.".format(self.command))
 133
 134 class ScenarioTaskCls(ScenarioTask):
 135     """
 136     Clear vterm screen.
 137     """
 138
 139     def __init__(self, args):
 140         ScenarioTask.__init__(self, 'vterm-cls')
 141
 142     def run(self):
 143         self.logger.info("Clearing the screen.")
 144
 145         for i in range(30):
 146             self.machine.type('\n')