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 
 27  from twisted.trial import unittest 
 28   
 29  from flumotion.common import log 
 30  from flumotion.configure import configure 
 31   
 32  __version__ = "$Rev: 6964 $" 
 33   
 34   
35 -class TestCase(unittest.TestCase):
36 37 # TestCase in Twisted 2.0 doesn't define failUnlessFailure method. 38 if not hasattr(unittest.TestCase, 'failUnlessFailure'):
39 - def failUnlessFailure(self, deferred, *expectedFailures):
40 def _cb(result): 41 self.fail("did not catch an error, instead got %r" % 42 (result,))
43 def _eb(failure): 44 failure.trap(*expectedFailures) 45 return failure.value
46 return deferred.addCallbacks(_cb, _eb) 47 assertFailure = failUnlessFailure 48 49 # test objects to be used in unittests to simulate the processes 50 # subclass them to add your own methods 51
52 -class TestClient(pb.Referenceable):
53 54 type = "client" # override in subclass 55 remoteRoot = None # RemoteReference to the server-side root 56
57 - def run(self, port):
58 """ 59 Start the client by connecting to the server on the given port. 60 61 @type port: int 62 63 @rtype: L{twisted.internet.defer.Deferred} 64 """ 65 self._f = pb.PBClientFactory() 66 self._p = reactor.connectTCP("127.0.0.1", port, self._f) 67 d = self._f.getRootObject() 68 d.addCallback(self._gotRootObject) 69 return d
70
71 - def stop(self):
72 """ 73 Stop the client. 74 75 @rtype: L{twisted.internet.defer.Deferred} 76 """ 77 self._p.disconnect() 78 return self._dDisconnect
79
80 - def _gotRootObject(self, remoteReference):
81 self.remoteRoot = remoteReference 82 83 # make sure we will get a deferred fired on disconnect 84 # so that the broker gets cleaned up from the reactor as well 85 self._dDisconnect = defer.Deferred() 86 self.remoteRoot.notifyOnDisconnect( 87 lambda r: self._dDisconnect.callback(None)) 88 return self.remoteRoot.callRemote('identify', self.type, self)
89
90 - def remote_receive(self, object):
91 # called by the server to send us an object 92 self.object = object
93 94
95 -class TestAdmin(TestClient):
96 type = 'admin'
97 98
99 -class TestWorker(TestClient):
100 type = 'worker'
101 102
103 -class TestManagerRoot(pb.Root, log.Loggable):
104 logCategory = "testmanagerroot"
105 - def remote_identify(self, who, reference):
106 """ 107 Called by a TestClient to announce the type of client, and give 108 a reference. 109 """ 110 self.debug('remote_identify: who %r, ref %r' % (who, reference)) 111 key = who + 'Reference' 112 setattr(self, key, reference)
113
114 - def remote_receive(self, object):
115 # called by the client to send us an object 116 self.object = object
117 118
119 -class TestManager:
120 - def run(self, rootClass):
121 """ 122 Run the test manager. Return port it is listening on. 123 124 @type rootClass: subclass of L{TestManagerRoot} 125 126 @rtype: int 127 """ 128 self.root = rootClass() 129 factory = pb.PBServerFactory(self.root) 130 factory.unsafeTracebacks = 1 131 self._p = reactor.listenTCP(0, factory, interface="127.0.0.1") 132 port = self._p.getHost().port 133 return port
134
135 - def stop(self):
136 """ 137 Stop the server. 138 """ 139 return self._p.stopListening()
140 141
142 -class TestPB(log.Loggable):
143 """ 144 I combine a manager and a client to test passing back and forth objects. 145 """ 146 logCategory = "testpb" 147
148 - def __init__(self):
149 self.manager = TestManager() 150 self.client = TestClient()
151
152 - def start(self):
153 port = self.manager.run(TestManagerRoot) 154 return self.client.run(port)
155
156 - def stop(self):
157 d = self.manager.stop() 158 d.addCallback(lambda r: self.client.stop()) 159 return d
160
161 - def send(self, object):
162 """ 163 Send the object from client to server. 164 Return the server's idea of the object. 165 """ 166 self.debug('sending object %r from broker %r' % ( 167 object, self.client.remoteRoot.broker)) 168 d = self.client.remoteRoot.callRemote('receive', object) 169 d.addCallback(lambda r: self.manager.root.object) 170 return d
171
172 - def receive(self, object):
173 """ 174 Receive the object from server to client. 175 Return the client's idea of the object. 176 """ 177 self.debug('receiving object %r' % object) 178 d = self.manager.root.clientReference.callRemote('receive', object) 179 d.addCallback(lambda r: self.client.object) 180 return d
181 182
183 -class TestCaseWithManager(TestCase):
184 - def setUp(self):
185 from flumotion.twisted import pb 186 from flumotion.common import server, connection 187 from flumotion.manager import manager, config 188 from StringIO import StringIO 189 190 managerConf = """ 191 <planet> 192 <manager name="planet"> 193 <host>localhost</host> 194 <port>0</port> 195 <transport>tcp</transport> 196 <component name="manager-bouncer" type="htpasswdcrypt-bouncer"> 197 <property name="data"><![CDATA[ 198 user:PSfNpHTkpTx1M 199 ]]></property> 200 </component> 201 </manager> 202 </planet> 203 """ 204 205 conf = config.ManagerConfigParser(StringIO(managerConf)).manager 206 self.vishnu = manager.Vishnu(conf.name, 207 unsafeTracebacks=True) 208 self.vishnu.loadManagerConfigurationXML(StringIO(managerConf)) 209 s = server.Server(self.vishnu) 210 if conf.transport == "ssl": 211 p = s.startSSL(conf.host, conf.port, conf.certificate, 212 configure.configdir) 213 elif conf.transport == "tcp": 214 p = s.startTCP(conf.host, conf.port) 215 self.tport = p 216 self.port = p.getHost().port 217 i = connection.PBConnectionInfo('localhost', self.port, 218 conf.transport == 'ssl', 219 pb.Authenticator(username='user', 220 password='test')) 221 self.connectionInfo = i
222
223 - def _flushErrors(self, *types):
224 # This bit about log flushing seems to be necessary with twisted < 2.5. 225 try: 226 self.flushLoggedErrors(*types) 227 except AttributeError: 228 from twisted.python import log as tlog 229 tlog.flushErrors(*types)
230
231 - def tearDown(self):
232 from flumotion.common import errors 233 self._flushErrors(errors.NotAuthenticatedError) 234 235 d = self.vishnu.shutdown() 236 d.addCallback(lambda _: self.tport.stopListening()) 237 return d
238
239 -def _diff(old, new, desc):
240 import difflib 241 lines = difflib.unified_diff(old, new) 242 lines = list(lines) 243 if not lines: 244 return 245 output = '' 246 for line in lines: 247 output += '%s: %s\n' % (desc, line[:-1]) 248 249 raise AssertionError( 250 ("\nError while comparing strings:\n" 251 "%s") % (output,))
252
253 -def diffStrings(orig, new, desc='input'):
254 def _tolines(s): 255 return [line + '\n' for line in s.split('\n')]
256 257 return _diff(_tolines(orig), 258 _tolines(new), 259 desc=desc) 260