1 from multiprocessing import Process |
1 from multiprocessing import Process |
2 from sqlalchemy import create_engine, pool |
|
3 import unittest |
2 import unittest |
4 |
3 |
5 from tempfile import mkdtemp |
|
6 import shutil |
|
7 |
|
8 from datetime import datetime |
4 from datetime import datetime |
9 |
5 |
10 import time |
6 import time |
11 |
7 |
12 from xmlrpclib import Server as xServer, ServerProxy, Fault |
8 from xmlrpclib import Server as xServer, ServerProxy, Fault |
13 |
9 |
14 from iro.controller.pool import dbPool |
10 from iro.model.schema import User, Offer, Userright, Job, Message |
15 |
|
16 from iro.model.schema import User, Base, Offer, Userright, Job, Message |
|
17 |
11 |
18 from iro.main import runReactor |
12 from iro.main import runReactor |
19 |
13 |
20 import iro.error as IroError |
14 import iro.error as IroError |
21 |
15 |
22 from ngdatabase.mysql import Server, createConfig, Database |
16 from .dbtestcase import DBTestCase, setUpModule, tearDownModule |
23 |
|
24 from .dbtestcase import DBTestCase |
|
25 |
|
26 class SampleDatabase(Database): |
|
27 def createPassword(self): |
|
28 self.password="test" |
|
29 return self.password |
|
30 |
17 |
31 |
18 |
32 #activates all logging we can get. |
19 #activates all logging we can get. |
33 |
20 |
34 from twisted.python import log |
21 from twisted.python import log |
38 observer.start() |
25 observer.start() |
39 |
26 |
40 class XMLRPCTest(DBTestCase): |
27 class XMLRPCTest(DBTestCase): |
41 """tests for the xmlrpc interface""" |
28 """tests for the xmlrpc interface""" |
42 def setUp(self): |
29 def setUp(self): |
43 if not self.engine: |
30 self.s = Process(target=startReactor, args=(self.engine,)) |
44 self.engine = md.engine |
|
45 self.s = Process(target=startReactor, args=(md.engine,)) |
|
46 self.s.start() |
31 self.s.start() |
47 #the new process needs time to get stated, so this process has to sleep |
32 #the new process needs time to get stated, so this process has to sleep |
48 time.sleep(.2) |
33 time.sleep(.2) |
49 |
34 |
50 def tearDown(self): |
35 def tearDown(self): |
245 root = resource.Resource() |
230 root = resource.Resource() |
246 root = appendResource(root) |
231 root = appendResource(root) |
247 root.putChild('debug', XMLRPCDebug()) |
232 root.putChild('debug', XMLRPCDebug()) |
248 runReactor(reactor, engine, root) |
233 runReactor(reactor, engine, root) |
249 |
234 |
250 |
|
251 class ModuleData: |
|
252 def __init__(self): |
|
253 self.tdir = mkdtemp(prefix='iro-mysql-') |
|
254 self.server = Server('%s/my.cnf'%self.tdir) |
|
255 self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir) |
|
256 self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir, |
|
257 poolclass = pool.SingletonThreadPool, pool_size=dbPool.maxthreads, )#echo=True) |
|
258 |
|
259 def setUp(self): |
|
260 with open('%s/my.cnf'%self.tdir,'w') as cnf: |
|
261 cnf.write(createConfig(self.tdir)) |
|
262 self.server.create() |
|
263 self.server.start() |
|
264 self.db.create() |
|
265 Base.metadata.create_all(self.engine) |
|
266 |
|
267 def tearDown(self): |
|
268 self.server.stop() |
|
269 shutil.rmtree(self.tdir) |
|
270 |
|
271 |
|
272 md=ModuleData() |
|
273 |
|
274 def setUpModule(): |
|
275 md.setUp() |
|
276 |
|
277 def tearDownModule(): |
|
278 md.tearDown() |
|
279 |
|
280 |
|
281 if __name__ == '__main__': |
235 if __name__ == '__main__': |
282 unittest.main() |
236 unittest.main() |