1 from twisted.internet.defer import inlineCallbacks

     2 from datetime import datetime

     3 from Queue import deque

     4 

     5 from iro.model.schema import User, Offer, Userright, Job, Message

     6 from iro.controller.viewinterface import Interface

     7 from iro.controller.pool  import taskPool

     8 

     9 from iro.model.message import SMS, Fax, Mail

    10 from iro.model.pool import data

    11 from iro.model.offer import offers

    12 from iro.model.job import exJobs

    13 

    14 import iro.error as IroError

    15 

    16 from ..test_helpers.dbtestcase import DBTestCase

    17 

    18 class DummyPool():

    19     def run(self, f,*a,**k):

    20         return f(*a,**k)

    21 

    22 class ViewInterfaceTest(DBTestCase):

    23     """tests for the xmlrpc interface"""

    24     def setUp(self):

    25         DBTestCase.setUp(self)

    26         self.pool = data.pool

    27         data.pool = DummyPool()

    28 

    29     def tearDown(self):

    30         exJobs.clear()

    31         offers.clear()

    32         taskPool.pool.q.queue = deque()

    33         data.pool = self.pool

    34         self.pool = None

    35         DBTestCase.tearDown(self)

    36     

    37     @inlineCallbacks    

    38     def testStatus(self):

    39         ''' test the status function'''

    40         with self.session() as session:

    41             u = User(name='test',apikey='abcdef123456789')

    42             session.add(User(name='test',apikey='abcdef123456789'))

    43         st = yield Interface().status('abcdef123456789')

    44         self.assertEqual(st, {})

    45         

    46         with self.session() as session:

    47             u = session.merge(u)

    48             j = Job(info='info', status="started")

    49             j.user=u

    50             session.add(j)

    51             session.commit()

    52             jid=j.id

    53         status = {str(jid):{"status":"started"}}

    54         st = yield Interface().status('abcdef123456789',jid)

    55         self.assertEqual(st, status)

    56         st = yield Interface().status('abcdef123456789')

    57         self.assertEqual(st, status)

    58         st = yield Interface().status('abcdef123456789', '', 'false')

    59         self.assertEqual(st, status)

    60         st = yield Interface().status('abcdef123456789', '', 0)

    61         self.assertEqual(st, status)

    62 

    63         #JobNotFound

    64         d = Interface().status('abcdef123456789',jid+1)

    65         self.assertFailure(d, IroError.JobNotFound)

    66         yield d

    67 

    68         #self.assertEqual(self.__rpc2().status('abcdef123456789','abcde', True), ["<User('test','abcdef123456789')>",'abcde', True])

    69         #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 'true'), ["<User('test','abcdef123456789')>", '', True])

    70         #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 1), ["<User('test','abcdef123456789')>", '', True])

    71     

    72     def testNoSuchUser(self):

    73         '''a unknown user should raise a UserNotNound Exception

    74         bewcause xmlrpc only has a Fault exception this Exception has to be deliverd through a xmlrpclib.Fault Exception'''

    75         d = Interface().status('abcdef123456789')

    76         self.assertFailure(d, IroError.UserNotFound)

    77         return d

    78 

    79     

    80     def testValidationFault(self):

    81         '''a validate Exception should be translated to a xmlrpclib.Fault.'''

    82         d = Interface().status('xxx')

    83         self.assertFailure(d, IroError.ValidateException)

    84 

    85     @inlineCallbacks

    86     def testRoutes(self):

    87         '''test the route function'''

    88         with self.session() as session:

    89             u=User(name='test',apikey='abcdef123456789')

    90             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    91             u.rights.append(Userright(o)) 

    92             session.add(u)

    93         r = yield Interface().routes('abcdef123456789','sms')

    94         self.assertEqual(r, ['sipgate_basic'])

    95 

    96         d = Interface().routes('abcdef123456789','fax')

    97         self.assertFailure(d,IroError.ValidateException)

    98         yield d

    99 

   100         with self.session() as session:

   101             o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms")

   102             u = session.query(User).filter_by(name="test").first()

   103             u.rights.append(Userright(o)) 

   104             o=Offer(name="faxde", provider="faxde", route="", typ="fax")

   105             session.add(o)

   106             session.commit()

   107         

   108         r = yield Interface().routes('abcdef123456789','sms')

   109         self.assertEqual(r, ['sipgate_basic','sipgate_plus'])

   110         r = yield Interface().routes('abcdef123456789','fax')

   111         self.assertEqual(r, [])

   112         

   113 

   114         with self.session() as session:

   115             u = session.query(User).filter_by(name="test").first()

   116             u.rights.append(Userright(o)) 

   117         

   118         r = yield Interface().routes('abcdef123456789','sms')

   119         self.assertEqual(r, ['sipgate_basic','sipgate_plus'])

   120         r = yield Interface().routes('abcdef123456789','fax')

   121         self.assertEqual(r, ['faxde'])

   122 

   123     @inlineCallbacks

   124     def testDefaultRoutes(self):

   125         '''test the defaultRoute function'''

   126         with self.session() as session:

   127             u=User(name='test',apikey='abcdef123456789')

   128             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

   129             u.rights.append(Userright(o,True)) 

   130             o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms")

   131             u.rights.append(Userright(o))

   132             session.add(u)

   133         r = yield Interface().defaultRoute('abcdef123456789','sms')

   134         self.assertEqual(r, ['sipgate_basic'])

   135 

   136     @inlineCallbacks

   137     def testTelnumbers(self):

   138         '''test the telefon validator'''

   139         r = yield Interface().telnumber(["0123/456(78)","+4912346785433","00123435456-658"])

   140         self.assertEqual(r, True)

   141 

   142         invalid=['xa','+1','1-23',';:+0','0123']

   143 

   144         d = Interface().telnumber(['01234']+invalid) 

   145         def x(failure):

   146             self.assertEqual(failure.getErrorMessage(),"701: No valid telnumber: '%s'"%invalid[0])

   147             failure.raiseException()

   148         d.addErrback(x)

   149         self.assertFailure(d, IroError.InvalidTel)

   150         yield d

   151 

   152     @inlineCallbacks

   153     def testVaildEmail(self):

   154         '''test vaild email adresses (got from wikipedia)'''

   155         validmails=["niceandsimple@example.com"]

   156         r = yield Interface().email(validmails)

   157         self.assertEqual(r,True)

   158     

   159     def testInvaildEmail(self):

   160         '''test invaild email adresses (got from wikipedia)'''

   161         invalid=["Abc.example.com",]

   162         d = Interface().email(invalid)

   163         self.assertFailure(d, IroError.InvalidMail)

   164         return d

   165 

   166     @inlineCallbacks

   167     def testBill(self):

   168         '''test bill function'''

   169         apikey='abcdef123456789'

   170         with self.session() as session:

   171             u=User(name='test',apikey=apikey)

   172             session.add(u)

   173         

   174         r = yield  Interface().bill(apikey)

   175         self.assertEqual(r,{'total':{'price':0.0,'anz':0}})

   176         

   177         with self.session() as session:

   178             u = session.merge(u)

   179             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   180             u.rights.append(Userright(o))

   181             j = Job(info='i',status='sended')

   182             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o))

   183             u.jobs.append(j)

   184  

   185             j = Job(info='a',status='sended')

   186             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o))

   187             u.jobs.append(j) 

   188         

   189         ret=yield Interface().bill(apikey) 

   190         self.assertEqual(ret['total'],{'price':0.8,'anz':2})

   191         self.assertEqual(ret['sipgate_basic'],

   192                 {'price':0.8,'anz':2,

   193                     'info':{'i':{'price':0.4,'anz':1},

   194                         'a':{'price':0.4,'anz':1},

   195                         }

   196                     })

   197 

   198     @inlineCallbacks

   199     def testSMS(self):

   200         with self.session() as session:

   201             u = User(name='test',apikey='abcdef123456789')

   202             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   203             u.rights.append(Userright(o))

   204             session.add(u)

   205         

   206         jobid = yield Interface().sms('abcdef123456789','message',['0123325456'],['sipgate_basic'])

   207        

   208         with self.session() as session:

   209             u = session.merge(u)

   210             job = u.job(jobid)

   211             exJob = job.extend

   212         

   213         self.assertEqual(exJob.message,SMS("message",None))

   214         self.assertEqual(taskPool.pool.q.qsize(),1)

   215 

   216 

   217     @inlineCallbacks

   218     def testMail(self):

   219         with self.session() as session:

   220             u = User(name='test',apikey='abcdef123456789')

   221             o = Offer(name='loc',provider="localhost",route="",typ="mail")

   222             u.rights.append(Userright(o))

   223             session.add(u)

   224         

   225         jobid = yield Interface().mail('abcdef123456789','sub', "hey body!", ['t@te.de'], "frm@t.de" ,['loc'])

   226         with self.session() as session:

   227             u = session.merge(u)

   228             job = u.job(jobid)

   229             exJob = job.extend

   230         

   231         self.assertEqual(exJob.message,Mail("sub",'hey body!','frm@t.de'))

   232         self.assertEqual(taskPool.pool.q.qsize(),1)

   233 

   234     @inlineCallbacks

   235     def testMailFrmNone(self):

   236         with self.session() as session:

   237             u = User(name='test',apikey='abcdef123456789')

   238             o = Offer(name='loc',provider="localhost",route="",typ="mail")

   239             u.rights.append(Userright(o))

   240             session.add(u)

   241         

   242         jobid = yield Interface().mail('abcdef123456789','sub', "hey body!", ['t@te.de'], None,['loc'])

   243         with self.session() as session:

   244             u = session.merge(u)

   245             job = u.job(jobid)

   246             exJob = job.extend

   247         

   248         self.assertEqual(exJob.message,Mail("sub",'hey body!',None))

   249         self.assertEqual(taskPool.pool.q.qsize(),1)

   250 

   251     

   252 

   253     @inlineCallbacks

   254     def testFax(self):

   255         with self.session() as session:

   256             u = User(name='test',apikey='abcdef123456789')

   257             o = Offer(name='b',provider="sipgate",route="b",typ="fax")

   258             u.rights.append(Userright(o))

   259             session.add(u)

   260         

   261         jobid = yield Interface().fax('abcdef123456789','subject', 'blublbubblu',['0123325456'],['b'])

   262        

   263         with self.session() as session:

   264             u = session.merge(u)

   265             job = u.job(jobid)

   266             exJob = job.extend

   267         

   268         self.assertEqual(exJob.message,Fax("subject","blublbubblu"))

   269         self.assertEqual(taskPool.pool.q.qsize(),1)