# HG changeset patch # User Sandro Knauß # Date 1330278317 -3600 # Node ID 71221fd1f7c16e36f70f4c816d57345d5ab64b44 # Parent ae2767918faafeda97d0965c6d00502c94aa6137 tests for viewinterface diff -r ae2767918faa -r 71221fd1f7c1 tests/viewinterface.py --- /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), ["",'abcde', True]) + #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 'true'), ["", '', True]) + #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 1), ["", '', 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"