indico/tests/base.py

   1 # -*- coding: utf-8 -*-
   2 ##
   3 ##
   4 ## This file is part of Indico.
   5 ## Copyright (C) 2002 - 2012 European Organization for Nuclear Research (CERN).
   6 ##
   7 ## Indico is free software; you can redistribute it and/or
   8 ## modify it under the terms of the GNU General Public License as
   9 ## published by the Free Software Foundation; either version 3 of the
  10 ## License, or (at your option) any later version.
  11 ##
  12 ## Indico is distributed in the hope that it will be useful, but
  13 ## WITHOUT ANY WARRANTY; without even the implied warranty of
  14 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15 ## General Public License for more details.
  16 ##
  17 ## You should have received a copy of the GNU General Public License
  18 ## along with Indico;if not, see <http://www.gnu.org/licenses/>.
  19
  20 """
  21 This module provides a skeleton of a test runner (BaseTestRunner) that all the
  22 other TestRunners should inherit from.
  23 """
  24
  25 # System modules
  26 import os, sys
  27 from threading import Thread
  28
  29 # Python stdlib
  30 import StringIO
  31 import pkg_resources
  32 from smtpd import SMTPServer
  33 import asyncore
  34 import logging
  35
  36 # Indico
  37 from indico.util.console import colored
  38 from indico.tests.util import TeeStringIO
  39 from indico.tests.config import TestConfig
  40
  41
  42 class TestOptionException(Exception):
  43     """
  44     Raised when a particular runner doesn't support an option
  45     """
  46
  47 class IOMixin(object):
  48     """
  49     Mixin class that provides some simple utility functions
  50     for error/info messages in the console
  51     """
  52
  53     @classmethod
  54     def _info(cls, message):
  55         """
  56         Prints an info message
  57         """
  58         print colored("** %s" % message, 'blue')
  59
  60     @classmethod
  61     def _success(cls, message):
  62         """
  63         Prints an info message
  64         """
  65         print colored("** %s" % message, 'green')
  66
  67     @classmethod
  68     def _error(cls, message):
  69         """
  70         Prints an error message
  71         """
  72         print colored("** %s" % message, 'red')
  73
  74
  75 class OptionProxy(object):
  76     """
  77     Encapsulates all the options present in a TestRunner,
  78     providing a common access point, and controlling some
  79     "hot spots" as well
  80     """
  81
  82     def __init__(self, allowedOptions):
  83         self._optionTable = allowedOptions
  84         self._options = {}
  85
  86     def call(self, runner, event, *args):
  87         """
  88         Invoked from a code hot spot, so that the option can
  89         perform operations
  90         """
  91
  92         for option in self._options.values():
  93             if hasattr(option, event) and option.shouldExecute():
  94                 getattr(option, event)(runner, *args)
  95
  96     def configure(self, **kwargs):
  97         """
  98         Initializes the options based on command line parameters
  99         """
 100
 101         for optName, optClass  in self._optionTable.iteritems():
 102             if optName in kwargs:
 103                 self._options[optName] = optClass(kwargs[optName])
 104             else:
 105                 self._options[optName] = optClass(None)
 106
 107         for optName in kwargs:
 108             if optName not in self._optionTable:
 109                 raise TestOptionException("Option '%s' not allowed here!" %
 110                                           optName)
 111
 112
 113     def valueOf(self, optName, default=None):
 114         """
 115         Returns the direct value of an option
 116         """
 117         if optName in self._options:
 118             return self._options[optName].value
 119         else:
 120             return default
 121
 122
 123 class Option(IOMixin):
 124     """
 125     Represents an option for a TestRunner
 126     """
 127
 128     def __init__(self, value):
 129         self.value = value
 130
 131     def shouldExecute(self):
 132         """
 133         Determines if the Option should be taken into account (hot spots),
 134         depending on the context
 135         """
 136         return True
 137
 138
 139 class BaseTestRunner(IOMixin):
 140     """
 141     Base class for all other TestRunners.
 142     A TestRunner runs a specific kind of test (i.e. UnitTestRunner)
 143     """
 144
 145     # overloaded for each runner, contains allowed options for each runner
 146
 147     # for this case:
 148     #   * silent - True if the output shouldn't be redirected to the console
 149
 150     _runnerOptions = {'silent': Option}
 151
 152     # path to this current file
 153     setupDir = os.path.dirname(__file__)
 154
 155     def __init__(self, **kwargs):
 156         """
 157         Options can be passed as kwargs, currently the following is supported:
 158
 159         """
 160
 161         self.err = None
 162         self.out = None
 163
 164         # make a TestConfig instance available everywhere
 165         self.config = TestConfig.getInstance()
 166
 167         # initialize allowed options
 168         self.options = OptionProxy(self._runnerOptions)
 169         self.options.configure(**kwargs)
 170         self._logger = logging.getLogger('test')
 171
 172     def _run(self):
 173         """
 174         This method should be overloaded by inheriting classes.
 175         It should provide the code that executes the actual tests,
 176         returning output information.
 177         """
 178         pass
 179
 180     def run(self):
 181         """
 182         Executes the actual test code
 183         """
 184         # get the description from the first lines
 185         # of the docstring
 186         description =  self.__doc__.strip().split('\n')[0]
 187
 188         self._startIOCapture()
 189
 190         self._info("Running %s" % description)
 191
 192         self._callOptions('pre_run')
 193         result = self._run()
 194         self._callOptions('post_run')
 195
 196         if result:
 197             self._success("%s successful!" % description)
 198         else:
 199             self._error("%s failed!" % description)
 200
 201         # ask the option handlers to compute a final message
 202         self._callOptions('final_message')
 203         self._writeReport(self.__class__.__name__,
 204                           self._finishIOCapture())
 205
 206         return result
 207
 208     def _startIOCapture(self):
 209         """
 210         Start capturing stdout and stderr to StringIOs
 211         If options['verbose'] has been set, the data will be output to the
 212         stdout/stderr as well
 213         """
 214
 215         if self.options.valueOf('silent'):
 216             # just capture it
 217             self.err = StringIO.StringIO()
 218             self.out = self.err
 219         else:
 220             # capture I/O but display it as well
 221             self.out = TeeStringIO(sys.stdout)
 222             self.err = TeeStringIO(sys.stderr, targetStream = self.out)
 223         sys.stderr = self.err
 224         sys.stdout = self.out
 225
 226
 227
 228     def _finishIOCapture(self):
 229         """
 230         Restore stdout/stderr and return the captured data
 231         """
 232         sys.stderr = sys.__stderr__
 233         sys.stdout = sys.__stdout__
 234
 235         return self.out.getvalue()
 236
 237     @staticmethod
 238     def findPlugins():
 239         """
 240         Goes throught the plugin directories, and adds
 241         existing unit test dirs
 242         """
 243
 244         dirs = []
 245
 246         for epoint in pkg_resources.iter_entry_points('indico.ext_types'):
 247             dirs.append(os.path.dirname(epoint.load().__file__))
 248
 249         for epoint in pkg_resources.iter_entry_points('indico.ext'):
 250             dirs.append(os.path.dirname(epoint.load().__file__))
 251
 252         return dirs
 253
 254     @staticmethod
 255     def _redirectPipeToStdout(pipe):
 256         """
 257         Redirect a given pipe to stdout
 258         """
 259         while True:
 260             data = pipe.readline()
 261             if not data:
 262                 break
 263             print data,
 264
 265     def _writeReport(self, filename, content):
 266         """
 267         Write the test report, using the filename and content that are passed
 268         """
 269         filePath = os.path.join(self.setupDir, 'report', filename + ".txt")
 270         try:
 271             f = open(filePath, 'w')
 272             f.write(content)
 273             f.close()
 274         except IOError:
 275             return "Unable to write in %s, check your file permissions." % \
 276                     os.path.join(self.setupDir, 'report', filename + ".txt")
 277
 278         self._info("report in %s" % filePath)
 279
 280     @staticmethod
 281     def walkThroughFolders(rootPath, foldersPattern):
 282         """
 283         Scan a directory and return folders which match the pattern
 284         """
 285
 286         rootPluginsPath = os.path.join(rootPath)
 287         foldersArray = []
 288
 289         for root, __, __ in os.walk(rootPluginsPath):
 290             if root.endswith(foldersPattern) > 0:
 291                 foldersArray.append(root)
 292
 293         return foldersArray
 294
 295     def _callOptions(self, method, *args):
 296         """
 297         Invokes the option proxy, providing the hot spot with name 'method',
 298         that options should have extended
 299         """
 300
 301         # invoke the option proxy
 302         self.options.call(self, method, *args)
 303
 304
 305 # Some utils
 306
 307 class FakeMailServer(SMTPServer):
 308     def process_message(self, peer, mailfrom, rcpttos, data):
 309          logging.getLogger('indico.test.fake_smtp').info("mail from %s" % mailfrom)
 310
 311
 312 class FakeMailThread(Thread):
 313     def __init__(self, addr):
 314         super(FakeMailThread, self).__init__()
 315         self.addr = addr
 316         self.server = FakeMailServer(self.addr, '')
 317
 318     def run(self):
 319         asyncore.loop()
 320
 321     def close(self):
 322         if self.server:
 323             self.server.close()
 324
 325     def get_addr(self):
 326         return self.server.socket.getsockname()