--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/iro/tests/xmlrpc.py Sun Mar 18 14:06:27 2012 +0100
@@ -0,0 +1,216 @@
+from multiprocessing import Process
+import unittest
+
+from datetime import datetime
+
+import time
+
+from xmlrpclib import Server as xServer, ServerProxy, Fault
+
+from iro.model.schema import User, Offer, Userright, Job, Message
+
+from iro.main import runReactor
+
+import iro.error as IroError
+
+from ..test_helpers.dbtestcase import DBTestCase
+
+
+class XMLRPCTest(DBTestCase):
+ """tests for the xmlrpc interface"""
+ def setUp(self):
+ DBTestCase.setUp(self)
+ self.s = Process(target=startReactor, args=(self.engine,))
+ self.s.start()
+ #the new process needs time to get stated, so this process has to sleep
+ time.sleep(.2)
+
+ def tearDown(self):
+ self.__debug().stop()
+ time.sleep(.2)
+ self.s.join()
+ DBTestCase.tearDown(self)
+
+ def __debug(self):
+ return xServer('http://localhost:7080/debug')
+
+ def __rpc2(self):
+ return ServerProxy('http://localhost:7080/RPC2')
+
+ def testDebugHello(self):
+ '''simple test for the connection to xmlrpc server'''
+ ret=self.__debug().hello()
+ self.failUnlessEqual(ret,'hello')
+
+ def testListMethods(self):
+ '''list of all offical Methods, that can be executed'''
+ ret=self.__rpc2().listMethods()
+ self.failUnlessEqual(ret, ['listMethods', 'status', 'stop', 'sms', 'fax', 'mail', 'routes', 'defaultRoute', 'bill', 'telnumber','email'])
+
+ def testStatus(self):
+ ''' test the status function'''
+ with self.session() as session:
+ u = User(name='test',apikey='abcdef123456789')
+ session.add(User(name='test',apikey='abcdef123456789'))
+ self.failUnlessEqual(self.__rpc2().status('abcdef123456789'), {})
+
+ with self.session() as session:
+ u = session.merge(u)
+ j = Job(info='info', status="started")
+ j.user=u
+ session.add(j)
+ session.commit()
+ jid=j.id
+ status = {str(jid):{"status":"started"}}
+ self.failUnlessEqual(self.__rpc2().status('abcdef123456789',jid), status)
+ self.failUnlessEqual(self.__rpc2().status('abcdef123456789'), status)
+ self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 'false'), status)
+ self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 0), status)
+
+ #JobNotFound
+ exc = self.assertRaises(Fault, self.__rpc2().status, 'abcdef123456789',jid+1)
+ unf = IroError.JobNotFound()
+ self.failUnlessEqual(exc.faultCode, unf.code)
+ self.failUnlessEqual(exc.faultString, unf.msg)
+
+ #self.failUnlessEqual(self.__rpc2().status('abcdef123456789','abcde', True), ["<User('test','abcdef123456789')>",'abcde', True])
+ #self.failUnlessEqual(self.__rpc2().status('abcdef123456789', '', 'true'), ["<User('test','abcdef123456789')>", '', True])
+ #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'''
+ exc = self.assertRaises(Fault, self.__rpc2().status, 'abcdef123456789')
+ unf=IroError.UserNotFound()
+ self.failUnlessEqual(exc.faultCode, unf.code)
+ self.failUnlessEqual(exc.faultString, unf.msg)
+
+ def testNoSuchMethod(self):
+ '''a unknown mothod should raise a Exception '''
+ exc = self.assertRaises(Fault, self.__rpc2().nosuchmethod)
+ self.failUnlessEqual(exc.faultCode, 8001)
+ self.failUnlessEqual(exc.faultString, "procedure nosuchmethod not found")
+
+ def testValidationFault(self):
+ '''a validate Exception should be translated to a xmlrpclib.Fault.'''
+ exc = self.assertRaises(Fault, self.__rpc2().status,'xxx')
+ self.failUnlessEqual(exc.faultCode, 700)
+ self.failUnlessEqual(exc.faultString, "Validation of 'apikey' failed.")
+
+ def testRoutes(self):
+ '''test the route function'''
+ with self.session() 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'])
+
+ exc = self.assertRaises(Fault, self.__rpc2().routes,'abcdef123456789','fax')
+ self.failUnlessEqual(exc.faultCode, 700)
+ self.failUnlessEqual(exc.faultString, "Typ is not valid.")
+
+ with self.session() 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))
+ o=Offer(name="faxde", provider="faxde", route="", typ="fax")
+ session.add(o)
+ session.commit()
+ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic','sipgate_plus'])
+ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','fax'),[])
+
+ with self.session() as session:
+ u = session.query(User).filter_by(name="test").first()
+ u.rights.append(Userright(o))
+
+ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic','sipgate_plus'])
+ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','fax'),['faxde'])
+
+ def testDefaultRoutes(self):
+ '''test the defaultRoute function'''
+ with self.session() as session:
+ u=User(name='test',apikey='abcdef123456789')
+ o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o,True))
+ o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms")
+ u.rights.append(Userright(o))
+ session.add(u)
+ self.failUnlessEqual(self.__rpc2().defaultRoute('abcdef123456789','sms'),['sipgate_basic'])
+
+ def testTelnumbers(self):
+ '''test the telefon validator'''
+ self.failUnlessEqual(self.__rpc2().telnumber(["0123/456(78)","+4912346785433","00123435456-658"]),True)
+
+ invalid=['xa','+1','1-23',';:+0','0123']
+
+ exc = self.assertRaises(Fault, self.__rpc2().telnumber,['01234']+invalid)
+ self.failUnlessEqual(exc.faultCode, 701)
+ self.failUnlessEqual(exc.faultString, "No valid telnumber: '%s'" % invalid[0])
+
+
+ def testVaildEmail(self):
+ '''test vaild email adresses (got from wikipedia)'''
+ validmails=["niceandsimple@example.com"]
+ self.failUnlessEqual(self.__rpc2().email(validmails),True)
+
+ def testInvaildEmail(self):
+ '''test invaild email adresses (got from wikipedia)'''
+ invalid=["Abc.example.com",]
+ exc= self.assertRaises(Fault, self.__rpc2().email, invalid)
+ self.failUnlessEqual(exc.faultCode, 702)
+ self.failUnlessEqual(exc.faultString, "No valid email: '%s'" % invalid[0])
+
+ def testBill(self):
+ '''test bill function'''
+ apikey='abcdef123456789'
+ with self.session() as session:
+ u=User(name='test',apikey=apikey)
+ session.add(u)
+
+ self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.0,'anz':0}})
+
+ with self.session() as session:
+ u = session.merge(u)
+ o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")
+ u.rights.append(Userright(o))
+ j = Job(info='i',status='sended')
+ j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o))
+ u.jobs.append(j)
+
+ j = Job(info='a',status='sended')
+ j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o))
+ u.jobs.append(j)
+
+ ret=self.__rpc2().bill(apikey)
+ self.failUnlessEqual(ret['total'],{'price':0.8,'anz':2})
+ self.failUnlessEqual(ret['sipgate_basic'],
+ {'price':0.8,'anz':2,
+ 'info':{'i':{'price':0.4,'anz':1},
+ 'a':{'price':0.4,'anz':1},
+ }
+ })
+
+
+def startReactor(engine):
+ """starts the Rector with a special debug Clild, so that the reactor can be stopped remotly. """
+ from twisted.internet import reactor
+ from twisted.web import xmlrpc, resource
+
+ from iro.view.xmlrpc import appendResource
+
+ class XMLRPCDebug(xmlrpc.XMLRPC):
+ def xmlrpc_stop(self):
+ reactor.callLater(0.1,reactor.stop)
+ return ""
+
+ def xmlrpc_hello(self):
+ return "hello"
+
+ root = resource.Resource()
+ root = appendResource(root)
+ root.putChild('debug', XMLRPCDebug())
+ runReactor(reactor, engine, root)
+
+if __name__ == '__main__':
+ unittest.main()