--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/tests/viewinterface.py Sun Feb 26 18:45:17 2012 +0100
@@ -0,0 +1,200 @@
+from twisted.internet.defer import inlineCallbacks
+from datetime import datetime
+
+from iro.model.schema import User, Offer, Userright, Job, Message
+from iro.controller.viewinterface import Interface
+from iro.model.pool import data
+
+import iro.error as IroError
+
+from .dbtestcase import DBTestCase
+
+class DummyPool():
+ def run(self, f,*a,**k):
+ return f(*a,**k)
+
+class ViewInterfaceTest(DBTestCase):
+ """tests for the xmlrpc interface"""
+ def setUp(self):
+ DBTestCase.setUp(self)
+ self.pool = data.pool
+ data.pool = DummyPool()
+
+ def tearDown(self):
+ data.pool = self.pool
+ self.pool = None
+ DBTestCase.tearDown(self)
+
+ @inlineCallbacks
+ 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'))
+ st = yield Interface().status('abcdef123456789')
+ self.assertEqual(st, {})
+
+ 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"}}
+ st = yield Interface().status('abcdef123456789',jid)
+ self.assertEqual(st, status)
+ st = yield Interface().status('abcdef123456789')
+ self.assertEqual(st, status)
+ st = yield Interface().status('abcdef123456789', '', 'false')
+ self.assertEqual(st, status)
+ st = yield Interface().status('abcdef123456789', '', 0)
+ self.assertEqual(st, status)
+
+ #JobNotFound
+ d = Interface().status('abcdef123456789',jid+1)
+ self.assertFailure(d, IroError.JobNotFound)
+ yield d
+
+ #self.assertEqual(self.__rpc2().status('abcdef123456789','abcde', True), ["<User('test','abcdef123456789')>",'abcde', True])
+ #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 'true'), ["<User('test','abcdef123456789')>", '', True])
+ #self.assertEqual(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'''
+ d = Interface().status('abcdef123456789')
+ self.assertFailure(d, IroError.UserNotFound)
+ return d
+
+
+ def testValidationFault(self):
+ '''a validate Exception should be translated to a xmlrpclib.Fault.'''
+ d = Interface().status('xxx')
+ self.assertFailure(d, IroError.ValidateException)
+
+ @inlineCallbacks
+ 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)
+ r = yield Interface().routes('abcdef123456789','sms')
+ self.assertEqual(r, ['sipgate_basic'])
+
+ d = Interface().routes('abcdef123456789','fax')
+ self.assertFailure(d,IroError.ValidateException)
+ yield d
+
+ 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()
+
+ r = yield Interface().routes('abcdef123456789','sms')
+ self.assertEqual(r, ['sipgate_basic','sipgate_plus'])
+ r = yield Interface().routes('abcdef123456789','fax')
+ self.assertEqual(r, [])
+
+
+ with self.session() as session:
+ u = session.query(User).filter_by(name="test").first()
+ u.rights.append(Userright(o))
+
+ r = yield Interface().routes('abcdef123456789','sms')
+ self.assertEqual(r, ['sipgate_basic','sipgate_plus'])
+ r = yield Interface().routes('abcdef123456789','fax')
+ self.assertEqual(r, ['faxde'])
+
+ @inlineCallbacks
+ 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)
+ r = yield Interface().defaultRoute('abcdef123456789','sms')
+ self.assertEqual(r, ['sipgate_basic'])
+
+ @inlineCallbacks
+ def testTelnumbers(self):
+ '''test the telefon validator'''
+ r = yield Interface().telnumber(["0123/456(78)","+4912346785433","00123435456-658"])
+ self.assertEqual(r, True)
+
+ invalid=['xa','+1','1-23',';:+0','0123']
+
+ d = Interface().telnumber(['01234']+invalid)
+ def x(failure):
+ self.assertEqual(failure.getErrorMessage(),"701:No valid telnumber: '%s'"%invalid[0])
+ failure.raiseException()
+ d.addErrback(x)
+ self.assertFailure(d, IroError.InvalidTel)
+ yield d
+
+ @inlineCallbacks
+ def testVaildEmail(self):
+ '''test vaild email adresses (got from wikipedia)'''
+ validmails=["niceandsimple@example.com"]
+ r = yield Interface().email(validmails)
+ self.assertEqual(r,True)
+
+ def testInvaildEmail(self):
+ '''test invaild email adresses (got from wikipedia)'''
+ invalid=["Abc.example.com",]
+ d = Interface().email(invalid)
+ self.assertFailure(d, IroError.InvalidMail)
+ return d
+
+ @inlineCallbacks
+ def testBill(self):
+ '''test bill function'''
+ apikey='abcdef123456789'
+ with self.session() as session:
+ u=User(name='test',apikey=apikey)
+ session.add(u)
+
+ r = yield Interface().bill(apikey)
+ self.assertEqual(r,{'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=yield Interface().bill(apikey)
+ self.assertEqual(ret['total'],{'price':0.8,'anz':2})
+ self.assertEqual(ret['sipgate_basic'],
+ {'price':0.8,'anz':2,
+ 'info':{'i':{'price':0.4,'anz':1},
+ 'a':{'price':0.4,'anz':1},
+ }
+ })
+
+ def testSMS(self):
+ pass
+ testSMS.todo = "To implement"
+
+
+ def testMail(self):
+ pass
+ testMail.todo = "To implement"
+
+ def testFax(self):
+ pass
+ testFax.todo = "To implement"