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