indico/tests/util.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 defines some utility classes for the testing framework,
  22 such as a wrapper for ZEOServer that allows a "test" server to be created
  23 """
  24
  25 import os, time, logging
  26 from multiprocessing import Process
  27 from StringIO import StringIO
  28
  29 from ZEO.runzeo import ZEOOptions, ZEOServer
  30
  31
  32 class SilentZEOServer(ZEOServer):
  33     """
  34     A ZEO Server that doesn't write on the console
  35     """
  36     def setup_default_logging(self):
  37         # do nothing, just use whatever handler is available
  38         pass
  39
  40
  41 class TestZEOServer(Process):
  42     """
  43     Creates a standalone ZEO server for tests
  44     """
  45     def __init__(self, port, fd, hostname="localhost"):
  46         Process.__init__(self)
  47         self.addr = (hostname, port)
  48         self.fd = fd
  49
  50     def run(self):
  51         """
  52         Actually starts the server
  53         """
  54         options = ZEOOptions()
  55         options.realize(['-f', self.fd, '-a', '%s:%d' % self.addr])
  56         self.server = SilentZEOServer(options)
  57         self.server.main()
  58
  59     def shutdown(self, killself=False):
  60         """
  61         This is basically a 'blocking' terminate()
  62         """
  63
  64         self.terminate()
  65
  66         if killself:
  67             # Do not wait for shutdown if we are going to kill ourselves
  68             return
  69
  70         # wait till i'm dead
  71         while self.is_alive():
  72             time.sleep(1)
  73
  74
  75 class TeeStringIO(StringIO):
  76     """
  77     Wrapper for StringIO that writes to an output stream as well
  78     """
  79     def __init__(self, out, targetStream = None):
  80         self.__outStream =  out
  81         self.__targetStream = targetStream
  82         StringIO.__init__(self)
  83
  84     def write(self, string, echo=True):
  85
  86         if echo:
  87             self.__outStream.write(string)
  88         if self.__targetStream:
  89             self.__targetStream.write(string, echo=False)
  90         else:
  91             StringIO.write(self, string)
  92
  93     def read(self, n=-1):
  94         self.seek(0)
  95         self.__outStream.write(StringIO.read(self.__targetStream or self, n=n))
  96
  97
  98 def openBrowser(browserPath, filePath):
  99     """
 100     Open a browser window with an HTML document
 101     """
 102     os.system("%s %s" % (browserPath, filePath))
 103
 104
 105 ###
 106 # path manipulation functions
 107 # adapted from path.py (Jason Orendorff <jason at jorendorff com>)
 108 #
 109 # (Public Domain code)
 110 ###
 111
 112 def splitall(loc):
 113     """
 114     Return a list of the path components in this path.
 115
 116     The first item in the list will be a path.  Its value will be
 117     either os.curdir, os.pardir, empty, or the root directory of
 118     this path (for example, '/' or 'C:\\').  The other items in
 119     the list will be strings.
 120
 121     path.path.joinpath(*result) will yield the original path.
 122     """
 123
 124     parts = []
 125     while loc != os.curdir and loc != os.pardir:
 126         prev = loc
 127         loc, child = os.path.split(prev)
 128         if loc == prev:
 129             break
 130         parts.append(child)
 131     parts.append(loc)
 132     parts.reverse()
 133     return parts
 134
 135
 136 def relpathto(origin, dest):
 137     """
 138     Return a relative path from self to dest.
 139
 140     If there is no relative path from self to dest, for example if
 141     they reside on different drives in Windows, then this returns
 142     dest.abspath().
 143     """
 144
 145     orig_list = splitall(os.path.normcase(origin))
 146     # Don't normcase dest!  We want to preserve the case.
 147     dest_list = splitall(dest)
 148
 149     if orig_list[0] != os.path.normcase(dest_list[0]):
 150         # Can't get here from there.
 151         return dest
 152
 153     # Find the location where the two paths start to differ.
 154     i = 0
 155     for start_seg, dest_seg in zip(orig_list, dest_list):
 156         if start_seg != os.path.normcase(dest_seg):
 157             break
 158         i += 1
 159
 160     # Now i is the point where the two paths diverge.
 161     # Need a certain number of "os.pardir"s to work up
 162     # from the origin to the point of divergence.
 163     segments = [os.pardir] * (len(orig_list) - i)
 164     # Need to add the diverging part of dest_list.
 165     segments += dest_list[i:]
 166     if len(segments) == 0:
 167         # If they happen to be identical, use os.curdir.
 168         relpath = os.curdir
 169     else:
 170         relpath = os.path.join(*segments)
 171     return relpath
 172
 173 ### END ###