--- a/tests/xmlrpc.py Fri Jan 27 21:14:18 2012 +0100
+++ b/tests/xmlrpc.py Fri Jan 27 21:21:41 2012 +0100
@@ -6,9 +6,11 @@
from tempfile import mkdtemp
import shutil
-from iro.model.utils import WithSession, POOL_SIZE as DB_POOL_SIZE
+from iro.model.utils import WithSession
+from iro.model import POOL_SIZE as DB_POOL_SIZE
-from iro.model.schema import User, Base
+from iro.model.schema import User, Base, Offer, Userright
+import iro.model.schema as schema
from ngdatabase.mysql import Server, createConfig, Database
@@ -46,6 +48,12 @@
self.__debug().stop()
time.sleep(.2)
self.s.join()
+ self.__cleanDB()
+
+ def __cleanDB(self):
+ with WithSession(md.engine, autocommit=True) as session:
+ for table in schema.__tables__:
+ session.query(getattr(schema,table)).delete()
def __debug(self):
return xServer('http://localhost:7080/debug')
@@ -64,6 +72,9 @@
self.failUnlessEqual(ret, ['listMethods', 'status', 'stop', 'sms', 'fax', 'mail', 'routes', 'defaultRoute', 'statistic'])
def testStatus(self):
+ ''' test the status function'''
+ with WithSession(md.engine, autocommit=True) as session:
+ session.add(User(name='test',apikey='abcdef123456789'))
ret = self.__rpc2().status('abcdef123456789')
self.failUnlessEqual(ret, "<User('test','abcdef123456789')>")
self.failUnlessEqual(self.__rpc2().status('abcdef123456789','abcde'), ["<User('test','abcdef123456789')>",'abcde',False])
@@ -73,12 +84,11 @@
self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 0), "<User('test','abcdef123456789')>")
self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 1), ["<User('test','abcdef123456789')>", '', True])
-
def testNoSuchUser(self):
'''a unknown user should raise a UserNotNound Exception
bewcause xmlrpc only has a Fault exception this Exception has to be deliverd through a xmlrpclib.Fault Exception'''
with self.assertRaises(Fault) as fault:
- self.__rpc2().status('abcdef123456788')
+ self.__rpc2().status('abcdef123456789')
exc = fault.exception
unf=IroError.UserNotFound()
self.failUnlessEqual(exc.faultCode, unf.code)
@@ -100,6 +110,28 @@
self.failUnlessEqual(exc.faultCode, 700)
self.failUnlessEqual(exc.faultString, "Validation of 'apikey' failed.")
+ def testRoutes(self):
+ '''test the route function'''
+ with WithSession(md.engine, autocommit=True) as session:
+ u=User(name='test',apikey='abcdef123456789')
+ o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o))
+ session.add(u)
+ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic'])
+
+ with self.assertRaises(Fault) as fault:
+ self.__rpc2().routes('abcdef123456789','fax')
+ exc = fault.exception
+ self.failUnlessEqual(exc.faultCode, 700)
+ self.failUnlessEqual(exc.faultString, "Typ is not valid.")
+
+ with WithSession(md.engine, autocommit=True) as session:
+ o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms")
+ u = session.query(User).filter_by(name="test").first()
+ u.rights.append(Userright(o))
+ session.commit()
+ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic','sipgate_plus'])
+
def startReactor(engine):
"""starts the Rector with a special debug Clild, so that the reactor can be stopped remotly. """
from twisted.internet import reactor
@@ -136,8 +168,6 @@
self.server.start()
self.db.create()
Base.metadata.create_all(self.engine)
- with WithSession(self.engine, autocommit=True) as session:
- session.add(User(name='test',apikey='abcdef123456789'))
def tearDown(self):
self.server.stop()