Package flumotion :: Package common :: Module testsuite
[hide private]

Source Code for Module flumotion.common.testsuite

  1  # -*- Mode: Python -*- 
  2  # vi:si:et:sw=4:sts=4:ts=4 
  3  # 
  4  # Flumotion - a streaming media server 
  5  # Copyright (C) 2004,2005,2006,2007,2008 Fluendo, S.L. (www.fluendo.com). 
  6  # All rights reserved. 
  7   
  8  # This file may be distributed and/or modified under the terms of 
  9  # the GNU General Public License version 2 as published by 
 10  # the Free Software Foundation. 
 11  # This file is distributed without any warranty; without even the implied 
 12  # warranty of merchantability or fitness for a particular purpose. 
 13  # See "LICENSE.GPL" in the source distribution for more information. 
 14   
 15  # Licensees having purchased or holding a valid Flumotion Advanced 
 16  # Streaming Server license may use this file in accordance with the 
 17  # Flumotion Advanced Streaming Server Commercial License Agreement. 
 18  # See "LICENSE.Flumotion" in the source distribution for more information. 
 19   
 20  # Headers in this file shall remain intact. 
 21   
 22  """testsuite base classes and helpers for diffing strings 
 23  """ 
 24   
 25  from twisted.spread import pb 
 26  from twisted.internet import reactor, defer, selectreactor 
 27  from twisted.scripts import trial 
 28  from twisted.trial import unittest, util 
 29   
 30  from flumotion.common import log 
 31  from flumotion.configure import configure 
 32   
 33  __version__ = "$Rev$" 
 34   
 35   
 36  try: 
 37      _getConfig = trial.getConfig 
 38  except AttributeError: 
 39      # trial.getConfig() is only available when using flumotion-trial 
 40      _getConfig = dict 
 41   
 42   
43 -def attr(*args, **kwargs):
44 """Decorator that adds attributes to objects. 45 46 It can be used to set the 'slow', 'skip', or 'todo' flags in test cases. 47 """ 48 49 def wrap(func): 50 for name in args: 51 # these are just True flags: 52 setattr(func, name, True) 53 for name, value in kwargs.items(): 54 setattr(func, name, value) 55 return func
56 return wrap 57 58
59 -class TestCase(unittest.TestCase, log.Loggable):
60 61 # A sequence of reactors classes that this test supports, can be 62 # overridden in subclasses. You can also set this to an empty 63 # sequence, which means "any reactor" 64 supportedReactors = [selectreactor.SelectReactor] 65 66 # TestCase in Twisted 2.0 doesn't define failUnlessFailure method. 67 if not hasattr(unittest.TestCase, 'failUnlessFailure'): 68
69 - def failUnlessFailure(self, deferred, *expectedFailures):
70 71 def _cb(result): 72 self.fail("did not catch an error, instead got %r" % 73 (result, ))
74 75 def _eb(failure): 76 failure.trap(*expectedFailures) 77 return failure.value
78 return deferred.addCallbacks(_cb, _eb) 79 assertFailure = failUnlessFailure 80 81 # notice the two spaces and read the following comment 82
83 - def __init__(self, methodName=' impossible-name '):
84 # Skip the test if the class specifies supportedReactors and 85 # the current reactor is not among them. Use 86 # reactor.__class__ rather than type(reactor), because in old 87 # Twisted the reactor was not a new-style class and 88 # type(reactor) returns 'instance' 89 if (self.supportedReactors and 90 reactor.__class__ not in self.supportedReactors): 91 # Set the 'skip' attribute on the class rather than on the 92 # instance, because otherwise Twisted 2.0.1 refuses to 93 # ignore the testcase 94 self.__class__.skip = "this test case does not support " \ 95 "running with %s as the reactor" % reactor 96 97 # Twisted changed the TestCase.__init__ signature several 98 # times. 99 # 100 # In versions older than 2.1.0 there was no __init__ method. 101 # 102 # In versions 2.1.0 up to 2.4.0 there is a __init__ method 103 # with a methodName kwarg that has a default value of None. 104 # 105 # In version 2.5.0 the default value of the kwarg was changed 106 # to "runTest". 107 # 108 # In versions above 2.5.0 God only knows what's the default 109 # value, as we do not currently support them. 110 import inspect 111 if not inspect.ismethod(unittest.TestCase.__init__): 112 # it's Twisted < 2.1.0 113 unittest.TestCase.__init__(self) 114 else: 115 # it's Twisted >= 2.1.0 116 if methodName == ' impossible-name ': 117 # we've been called with no parameters, use the 118 # default parameter value from the superclass 119 defaults = inspect.getargspec(unittest.TestCase.__init__)[3] 120 methodName = defaults[0] 121 unittest.TestCase.__init__(self, methodName=methodName) 122 123 # Skip slow tests if '--skip-slow' option is enabled 124 if _getConfig().get('skip-slow'): 125 if self.getSlow() and not self.getSkip(): 126 self.skip = 'slow test'
127
128 - def getSlow(self):
129 """ 130 Return whether this test has been marked as slow. Checks on the 131 instance first, then the class, then the module, then packages. As 132 soon as it finds something with a C{slow} attribute, returns that. 133 Returns C{False} if it cannot find anything. 134 """ 135 return util.acquireAttribute(self._parents, 'slow', False)
136 137 # Loggable and TestCase both have a debug method; prefer ours 138
139 - def debug(self, *args, **kwargs):
140 log.Loggable.debug(self, *args, **kwargs)
141 142 143 # test objects to be used in unittests to simulate the processes 144 # subclass them to add your own methods 145 146
147 -class TestClient(pb.Referenceable):
148 149 type = "client" # override in subclass 150 remoteRoot = None # RemoteReference to the server-side root 151
152 - def run(self, port):
153 """ 154 Start the client by connecting to the server on the given port. 155 156 @type port: int 157 158 @rtype: L{twisted.internet.defer.Deferred} 159 """ 160 self._f = pb.PBClientFactory() 161 self._p = reactor.connectTCP("127.0.0.1", port, self._f) 162 d = self._f.getRootObject() 163 d.addCallback(self._gotRootObject) 164 return d
165
166 - def stop(self):
167 """ 168 Stop the client. 169 170 @rtype: L{twisted.internet.defer.Deferred} 171 """ 172 self._p.disconnect() 173 return self._dDisconnect
174
175 - def _gotRootObject(self, remoteReference):
176 self.remoteRoot = remoteReference 177 178 # make sure we will get a deferred fired on disconnect 179 # so that the broker gets cleaned up from the reactor as well 180 self._dDisconnect = defer.Deferred() 181 self.remoteRoot.notifyOnDisconnect( 182 lambda r: self._dDisconnect.callback(None)) 183 return self.remoteRoot.callRemote('identify', self.type, self)
184
185 - def remote_receive(self, object):
186 # called by the server to send us an object 187 self.object = object
188 189
190 -class TestAdmin(TestClient):
191 type = 'admin'
192 193
194 -class TestWorker(TestClient):
195 type = 'worker'
196 197
198 -class TestManagerRoot(pb.Root, log.Loggable):
199 logCategory = "testmanagerroot" 200
201 - def remote_identify(self, who, reference):
202 """ 203 Called by a TestClient to announce the type of client, and give 204 a reference. 205 """ 206 self.debug('remote_identify: who %r, ref %r' % (who, reference)) 207 key = who + 'Reference' 208 setattr(self, key, reference)
209
210 - def remote_receive(self, object):
211 # called by the client to send us an object 212 self.object = object
213 214
215 -class TestManager:
216
217 - def run(self, rootClass):
218 """ 219 Run the test manager. Return port it is listening on. 220 221 @type rootClass: subclass of L{TestManagerRoot} 222 223 @rtype: int 224 """ 225 self.root = rootClass() 226 factory = pb.PBServerFactory(self.root) 227 factory.unsafeTracebacks = 1 228 self._p = reactor.listenTCP(0, factory, interface="127.0.0.1") 229 port = self._p.getHost().port 230 return port
231
232 - def stop(self):
233 """ 234 Stop the server. 235 """ 236 return self._p.stopListening()
237 238
239 -class TestPB(log.Loggable):
240 """ 241 I combine a manager and a client to test passing back and forth objects. 242 """ 243 logCategory = "testpb" 244
245 - def __init__(self):
246 self.manager = TestManager() 247 self.client = TestClient()
248
249 - def start(self):
250 port = self.manager.run(TestManagerRoot) 251 return self.client.run(port)
252
253 - def stop(self):
254 d = self.manager.stop() 255 d.addCallback(lambda r: self.client.stop()) 256 return d
257
258 - def send(self, object):
259 """ 260 Send the object from client to server. 261 Return the server's idea of the object. 262 """ 263 self.debug('sending object %r from broker %r' % ( 264 object, self.client.remoteRoot.broker)) 265 d = self.client.remoteRoot.callRemote('receive', object) 266 d.addCallback(lambda r: self.manager.root.object) 267 return d
268
269 - def receive(self, object):
270 """ 271 Receive the object from server to client. 272 Return the client's idea of the object. 273 """ 274 self.debug('receiving object %r' % object) 275 d = self.manager.root.clientReference.callRemote('receive', object) 276 d.addCallback(lambda r: self.client.object) 277 return d
278 279
280 -class TestCaseWithManager(TestCase):
281
282 - def setUp(self):
283 from flumotion.twisted import pb 284 from flumotion.common import server, connection 285 from flumotion.manager import manager, config 286 from StringIO import StringIO 287 288 managerConf = """ 289 <planet> 290 <manager name="planet"> 291 <host>localhost</host> 292 <port>0</port> 293 <transport>tcp</transport> 294 <component name="manager-bouncer" type="htpasswdcrypt-bouncer"> 295 <property name="data"><![CDATA[ 296 user:PSfNpHTkpTx1M 297 ]]></property> 298 </component> 299 </manager> 300 </planet> 301 """ 302 303 conf = config.ManagerConfigParser(StringIO(managerConf)).manager 304 self.vishnu = manager.Vishnu(conf.name, 305 unsafeTracebacks=True) 306 self.vishnu.loadManagerConfigurationXML(StringIO(managerConf)) 307 s = server.Server(self.vishnu) 308 if conf.transport == "ssl": 309 p = s.startSSL(conf.host, conf.port, conf.certificate, 310 configure.configdir) 311 elif conf.transport == "tcp": 312 p = s.startTCP(conf.host, conf.port) 313 self.tport = p 314 self.port = p.getHost().port 315 i = connection.PBConnectionInfo('localhost', self.port, 316 conf.transport == 'ssl', 317 pb.Authenticator(username='user', 318 password='test')) 319 self.connectionInfo = i
320
321 - def _flushErrors(self, *types):
322 # This bit about log flushing seems to be necessary with twisted < 2.5. 323 try: 324 self.flushLoggedErrors(*types) 325 except AttributeError: 326 from twisted.python import log as tlog 327 tlog.flushErrors(*types)
328
329 - def tearDown(self):
330 from flumotion.common import errors 331 self._flushErrors(errors.NotAuthenticatedError) 332 333 d = self.vishnu.shutdown() 334 d.addCallback(lambda _: self.tport.stopListening()) 335 return d
336 337
338 -def _diff(old, new, desc):
339 import difflib 340 lines = difflib.unified_diff(old, new) 341 lines = list(lines) 342 if not lines: 343 return 344 output = '' 345 for line in lines: 346 output += '%s: %s\n' % (desc, line[:-1]) 347 348 raise AssertionError( 349 ("\nError while comparing strings:\n" 350 "%s") % (output, ))
351 352
353 -def diffStrings(orig, new, desc='input'):
354 355 def _tolines(s): 356 return [line + '\n' for line in s.split('\n')]
357 358 return _diff(_tolines(orig), 359 _tolines(new), 360 desc=desc) 361