from twisted.internet.defer import inlineCallbacks
from datetime import datetime
from Queue import deque
from iro.model.schema import User, Offer, Userright, Job, Message
from iro.controller.viewinterface import Interface
from iro.controller.pool import taskPool
from iro.model.message import SMS, Fax, Mail
from iro.model.pool import data
from iro.model.offer import offers
from iro.model.job import exJobs
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):
exJobs.clear()
offers.clear()
taskPool.pool.q.queue = deque()
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},
}
})
@inlineCallbacks
def testSMS(self):
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)
jobid = yield Interface().sms('abcdef123456789','message',['0123325456'],['basic'])
with self.session() as session:
u = session.merge(u)
job = u.job(jobid)
exJob = job.extend
self.assertEqual(exJob.message,SMS("message",None))
self.assertEqual(taskPool.pool.q.qsize(),1)
@inlineCallbacks
def testMail(self):
with self.session() as session:
u = User(name='test',apikey='abcdef123456789')
o = Offer(name='loc',provider="localhost",route="",typ="mail")
u.rights.append(Userright(o))
session.add(u)
jobid = yield Interface().mail('abcdef123456789','sub', "hey body!", ['t@te.de'], "frm@t.de" ,['loc'])
with self.session() as session:
u = session.merge(u)
job = u.job(jobid)
exJob = job.extend
self.assertEqual(exJob.message,Mail("sub",'hey body!','frm@t.de'))
self.assertEqual(taskPool.pool.q.qsize(),1)
def testFax(self):
pass
testFax.todo = "To implement"