tests/xmlrpc.py
branchdevel
changeset 117 351a02310dd8
parent 114 1ed072cc6793
child 119 3462c2c90f95
--- 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()