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.trial import unittest 
 28   
 29  from flumotion.common import log 
 30  from flumotion.configure import configure 
 31   
 32  __version__ = "$Rev: 7541 $" 
 33   
 34   
35 -class TestCase(unittest.TestCase, log.Loggable):
36 37 # A sequence of reactors classes that this test supports, can be 38 # overridden in subclasses. You can also set this to an empty 39 # sequence, which means "any reactor" 40 supportedReactors = [selectreactor.SelectReactor] 41 42 # TestCase in Twisted 2.0 doesn't define failUnlessFailure method. 43 if not hasattr(unittest.TestCase, 'failUnlessFailure'): 44
45 - def failUnlessFailure(self, deferred, *expectedFailures):
46 47 def _cb(result): 48 self.fail("did not catch an error, instead got %r" % 49 (result, ))
50 51 def _eb(failure): 52 failure.trap(*expectedFailures) 53 return failure.value
54 return deferred.addCallbacks(_cb, _eb) 55 assertFailure = failUnlessFailure 56 57 # notice the two spaces and read the following comment 58
59 - def __init__(self, methodName=' impossible-name '):
60 # Skip the test if the class specifies supportedReactors and 61 # the current reactor is not among them. Use 62 # reactor.__class__ rather than type(reactor), because in old 63 # Twisted the reactor was not a new-style class and 64 # type(reactor) returns 'instance' 65 if (self.supportedReactors and 66 reactor.__class__ not in self.supportedReactors): 67 # Set the 'skip' attribute on the class rather than on the 68 # instance, because otherwise Twisted 2.0.1 refuses to 69 # ignore the testcase 70 self.__class__.skip = "this test case does not support " \ 71 "running with %s as the reactor" % reactor 72 73 # Twisted changed the TestCase.__init__ signature several 74 # times. 75 # 76 # In versions older than 2.1.0 there was no __init__ method. 77 # 78 # In versions 2.1.0 up to 2.4.0 there is a __init__ method 79 # with a methodName kwarg that has a default value of None. 80 # 81 # In version 2.5.0 the default value of the kwarg was changed 82 # to "runTest". 83 # 84 # In versions above 2.5.0 God only knows what's the default 85 # value, as we do not currently support them. 86 import inspect 87 if not inspect.ismethod(unittest.TestCase.__init__): 88 # it's Twisted < 2.1.0 89 unittest.TestCase.__init__(self) 90 else: 91 # it's Twisted >= 2.1.0 92 if methodName == ' impossible-name ': 93 # we've been called with no parameters, use the 94 # default parameter value from the superclass 95 defaults = inspect.getargspec(unittest.TestCase.__init__)[3] 96 methodName = defaults[0] 97 unittest.TestCase.__init__(self, methodName=methodName)
98 99 # Loggable and TestCase both have a debug method; prefer ours 100
101 - def debug(self, *args, **kwargs):
102 log.Loggable.debug(self, *args, **kwargs)
103 104 105 # test objects to be used in unittests to simulate the processes 106 # subclass them to add your own methods 107 108
109 -class TestClient(pb.Referenceable):
110 111 type = "client" # override in subclass 112 remoteRoot = None # RemoteReference to the server-side root 113
114 - def run(self, port):
115 """ 116 Start the client by connecting to the server on the given port. 117 118 @type port: int 119 120 @rtype: L{twisted.internet.defer.Deferred} 121 """ 122 self._f = pb.PBClientFactory() 123 self._p = reactor.connectTCP("127.0.0.1", port, self._f) 124 d = self._f.getRootObject() 125 d.addCallback(self._gotRootObject) 126 return d
127
128 - def stop(self):
129 """ 130 Stop the client. 131 132 @rtype: L{twisted.internet.defer.Deferred} 133 """ 134 self._p.disconnect() 135 return self._dDisconnect
136
137 - def _gotRootObject(self, remoteReference):
138 self.remoteRoot = remoteReference 139 140 # make sure we will get a deferred fired on disconnect 141 # so that the broker gets cleaned up from the reactor as well 142 self._dDisconnect = defer.Deferred() 143 self.remoteRoot.notifyOnDisconnect( 144 lambda r: self._dDisconnect.callback(None)) 145 return self.remoteRoot.callRemote('identify', self.type, self)
146
147 - def remote_receive(self, object):
148 # called by the server to send us an object 149 self.object = object
150 151
152 -class TestAdmin(TestClient):
153 type = 'admin'
154 155
156 -class TestWorker(TestClient):
157 type = 'worker'
158 159
160 -class TestManagerRoot(pb.Root, log.Loggable):
161 logCategory = "testmanagerroot" 162
163 - def remote_identify(self, who, reference):
164 """ 165 Called by a TestClient to announce the type of client, and give 166 a reference. 167 """ 168 self.debug('remote_identify: who %r, ref %r' % (who, reference)) 169 key = who + 'Reference' 170 setattr(self, key, reference)
171
172 - def remote_receive(self, object):
173 # called by the client to send us an object 174 self.object = object
175 176
177 -class TestManager:
178
179 - def run(self, rootClass):
180 """ 181 Run the test manager. Return port it is listening on. 182 183 @type rootClass: subclass of L{TestManagerRoot} 184 185 @rtype: int 186 """ 187 self.root = rootClass() 188 factory = pb.PBServerFactory(self.root) 189 factory.unsafeTracebacks = 1 190 self._p = reactor.listenTCP(0, factory, interface="127.0.0.1") 191 port = self._p.getHost().port 192 return port
193
194 - def stop(self):
195 """ 196 Stop the server. 197 """ 198 return self._p.stopListening()
199 200
201 -class TestPB(log.Loggable):
202 """ 203 I combine a manager and a client to test passing back and forth objects. 204 """ 205 logCategory = "testpb" 206
207 - def __init__(self):
208 self.manager = TestManager() 209 self.client = TestClient()
210
211 - def start(self):
212 port = self.manager.run(TestManagerRoot) 213 return self.client.run(port)
214
215 - def stop(self):
216 d = self.manager.stop() 217 d.addCallback(lambda r: self.client.stop()) 218 return d
219
220 - def send(self, object):
221 """ 222 Send the object from client to server. 223 Return the server's idea of the object. 224 """ 225 self.debug('sending object %r from broker %r' % ( 226 object, self.client.remoteRoot.broker)) 227 d = self.client.remoteRoot.callRemote('receive', object) 228 d.addCallback(lambda r: self.manager.root.object) 229 return d
230
231 - def receive(self, object):
232 """ 233 Receive the object from server to client. 234 Return the client's idea of the object. 235 """ 236 self.debug('receiving object %r' % object) 237 d = self.manager.root.clientReference.callRemote('receive', object) 238 d.addCallback(lambda r: self.client.object) 239 return d
240 241
242 -class TestCaseWithManager(TestCase):
243
244 - def setUp(self):
245 from flumotion.twisted import pb 246 from flumotion.common import server, connection 247 from flumotion.manager import manager, config 248 from StringIO import StringIO 249 250 managerConf = """ 251 <planet> 252 <manager name="planet"> 253 <host>localhost</host> 254 <port>0</port> 255 <transport>tcp</transport> 256 <component name="manager-bouncer" type="htpasswdcrypt-bouncer"> 257 <property name="data"><![CDATA[ 258 user:PSfNpHTkpTx1M 259 ]]></property> 260 </component> 261 </manager> 262 </planet> 263 """ 264 265 conf = config.ManagerConfigParser(StringIO(managerConf)).manager 266 self.vishnu = manager.Vishnu(conf.name, 267 unsafeTracebacks=True) 268 self.vishnu.loadManagerConfigurationXML(StringIO(managerConf)) 269 s = server.Server(self.vishnu) 270 if conf.transport == "ssl": 271 p = s.startSSL(conf.host, conf.port, conf.certificate, 272 configure.configdir) 273 elif conf.transport == "tcp": 274 p = s.startTCP(conf.host, conf.port) 275 self.tport = p 276 self.port = p.getHost().port 277 i = connection.PBConnectionInfo('localhost', self.port, 278 conf.transport == 'ssl', 279 pb.Authenticator(username='user', 280 password='test')) 281 self.connectionInfo = i
282
283 - def _flushErrors(self, *types):
284 # This bit about log flushing seems to be necessary with twisted < 2.5. 285 try: 286 self.flushLoggedErrors(*types) 287 except AttributeError: 288 from twisted.python import log as tlog 289 tlog.flushErrors(*types)
290
291 - def tearDown(self):
292 from flumotion.common import errors 293 self._flushErrors(errors.NotAuthenticatedError) 294 295 d = self.vishnu.shutdown() 296 d.addCallback(lambda _: self.tport.stopListening()) 297 return d
298 299
300 -def _diff(old, new, desc):
301 import difflib 302 lines = difflib.unified_diff(old, new) 303 lines = list(lines) 304 if not lines: 305 return 306 output = '' 307 for line in lines: 308 output += '%s: %s\n' % (desc, line[:-1]) 309 310 raise AssertionError( 311 ("\nError while comparing strings:\n" 312 "%s") % (output, ))
313 314
315 -def diffStrings(orig, new, desc='input'):
316 317 def _tolines(s): 318 return [line + '\n' for line in s.split('\n')]
319 320 return _diff(_tolines(orig), 321 _tolines(new), 322 desc=desc) 323