1 # Copyright (c) 2012 netzguerilla.net <iro@netzguerilla.net>

     2 #

     3 # This file is part of Iro.

     4 #

     5 # Permission is hereby granted, free of charge, to any person obtaining a copy of

     6 # this software and associated documentation files (the "Software"), to deal in

     7 # the Software without restriction, including without limitation the rights to use,

     8 # copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the

     9 # #Software, and to permit persons to whom the Software is furnished to do so,

    10 # subject to the following conditions:

    11 #

    12 # The above copyright notice and this permission notice shall be included in

    13 # all copies or substantial portions of the Software.

    14 #

    15 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,

    16 # INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A

    17 # PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT

    18 # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION

    19 # OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE

    20 # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

    21 

    22 import unittest

    23 from twisted.internet import reactor, defer

    24 from twisted.web import server

    25 from txjsonrpc.web import jsonrpc as txjsonrpc

    26 from txjsonrpc.web.jsonrpc import Fault

    27 from  txjsonrpc import jsonrpclib

    28 

    29 from datetime import datetime

    30 

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

    32 

    33 from iro.view import jsonrpc

    34 import iro.error as IroError

    35 

    36 from iro.test_helpers.dbtestcase import DBTestCase

    37 

    38 from iro.model import setEngine, setPool

    39 from iro.controller.pool import startPool, dbPool

    40 

    41 

    42 class JSONRPCTest(DBTestCase):

    43     """tests for the jsonrpc interface"""

    44     def setUp(self):

    45         DBTestCase.setUp(self)

    46 

    47         setEngine(self.engine)

    48         startPool(reactor)

    49         setPool(dbPool)

    50         self.p = reactor.listenTCP(0, server.Site(jsonrpc.JSONRPCInterface()),

    51                 interface="127.0.0.1")

    52         self.port = self.p.getHost().port

    53 

    54     def tearDown(self):

    55         DBTestCase.tearDown(self)

    56         return self.p.stopListening()

    57 

    58     def proxy(self):

    59         return txjsonrpc.Proxy("http://127.0.0.1:%d/" % self.port,version=jsonrpclib.VERSION_2)

    60 

    61     def testListMethods(self):

    62         '''list of all offical Methods, that can be executed'''

    63 

    64         def cbMethods(ret):

    65             ret.sort()

    66             self.failUnlessEqual(ret, ['bill', 'defaultRoute', 'email', 'fax', 'listMethods', 'mail', 'routes', 'sms', 'status', 'telnumber'])

    67 

    68         d=self.proxy().callRemote("listMethods")

    69         d.addCallback(cbMethods)

    70         return d

    71 

    72     def testApikey(self):

    73         ''' test apikey'''

    74         with self.session() as session:

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

    76 

    77         d = self.proxy().callRemote('status', 'abcdef123456789')

    78         d.addCallback(lambda ret: self.failUnlessEqual(ret,{}))

    79 

    80         return d

    81 

    82     def testStatus(self):

    83         ''' test the status function'''

    84         with self.session() as session:

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

    86             session.add(u)

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

    88             j.user=u

    89             session.add(j)

    90             session.commit()

    91             jid=j.id

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

    93         args=[('abcdef123456789',jid),

    94                 ('abcdef123456789',),

    95                 ('abcdef123456789', '', 'false'),

    96                 ('abcdef123456789', '', 0),

    97                 ]

    98         dl=[]

    99         for a in args:

   100             d = self.proxy().callRemote('status',*a)

   101             d.addCallback(lambda ret: self.failUnlessEqual(ret,status))

   102             dl.append(d)

   103 

   104         def f(exc):

   105             jnf = IroError.JobNotFound()

   106             self.failUnlessEqual(exc.faultCode, jnf.code)

   107             self.failUnlessEqual(exc.faultString, jnf.msg)

   108         d = self.proxy().callRemote('status', 'abcdef123456789', jid+1)

   109         d = self.assertFailure(d, Fault)

   110         d.addCallback(f)

   111         dl.append(d)

   112 

   113         return defer.DeferredList(dl, fireOnOneErrback=True)

   114 

   115     def testNoSuchUser(self):

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

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

   118         def f(exc):

   119             unf = IroError.UserNotFound()

   120             self.failUnlessEqual(exc.faultCode, unf.code)

   121             self.failUnlessEqual(exc.faultString, unf.msg)

   122         d = self.proxy().callRemote('status', 'abcdef123456789')

   123         d = self.assertFailure(d, Fault)

   124         d.addCallback(f)

   125         return d

   126 

   127     def testNoSuchMethod(self):

   128         '''a unknown mothod should raise a Exception '''

   129         def f(exc):

   130             self.failUnlessEqual(exc.faultCode, -32601)

   131             self.failUnlessEqual(exc.faultString, 'procedure nosuchmethod not found')

   132         d = self.proxy().callRemote('nosuchmethod')

   133         d = self.assertFailure(d, Fault)

   134         d.addCallback(f)

   135         return d

   136 

   137     def testValidationFault(self):

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

   139         def f(exc):

   140             self.failUnlessEqual(exc.faultCode, 700)

   141             self.failUnlessEqual(exc.faultString, "Validation of 'apikey' failed.")

   142         d = self.proxy().callRemote('status','xxx')

   143         d = self.assertFailure(d, Fault)

   144         d.addCallback(f)

   145         return d

   146 

   147     @defer.inlineCallbacks

   148     def testRoutes(self):

   149         '''test the route function'''

   150         with self.session() as session:

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

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

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

   154             session.add(u)

   155 

   156         x = yield self.proxy().callRemote('routes','abcdef123456789','sms')

   157         self.failUnlessEqual(x,['sipgate_basic'])

   158 

   159         def f(exc):

   160             self.failUnlessEqual(exc.faultCode, 700)

   161             self.failUnlessEqual(exc.faultString, "Typ fax is not valid.")

   162         d = self.proxy().callRemote('routes','abcdef123456789','fax')

   163         d = self.assertFailure(d, Fault)

   164         d.addCallback(f)

   165         yield d

   166 

   167         with self.session() as session:

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

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

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

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

   172             session.add(o)

   173             session.commit()

   174 

   175         x = yield self.proxy().callRemote('routes','abcdef123456789','sms')

   176         self.failUnlessEqual(x, ['sipgate_basic','sipgate_plus'])

   177         x = yield self.proxy().callRemote('routes','abcdef123456789','fax')

   178         self.failUnlessEqual(x, [])

   179 

   180         with self.session() as session:

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

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

   183 

   184         x = yield self.proxy().callRemote('routes','abcdef123456789','sms')

   185         self.failUnlessEqual(x, ['sipgate_basic','sipgate_plus'])

   186         x = yield self.proxy().callRemote('routes','abcdef123456789','fax')

   187         self.failUnlessEqual(x, ['faxde'])

   188 

   189     def testDefaultRoutes(self):

   190         '''test the defaultRoute function'''

   191         with self.session() as session:

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

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

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

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

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

   197             session.add(u)

   198         d = self.proxy().callRemote('defaultRoute','abcdef123456789','sms')

   199         d.addCallback(lambda x: self.failUnlessEqual(x,['sipgate_basic']))

   200 

   201         return d

   202 

   203     def testTelnumbers(self):

   204         '''test the telefon validator'''

   205         dl = []

   206         d = self.proxy().callRemote('telnumber',["0123/456(78)","+4912346785433","00123435456-658"])

   207         d.addCallback(lambda x: self.failUnlessEqual(x,True))

   208         dl.append(d)

   209 

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

   211         def f(exc):

   212             self.failUnlessEqual(exc.faultCode, 701)

   213             self.failUnlessEqual(exc.faultString, "No valid telnumber: '%s'" % invalid[0])

   214         d = self.proxy().callRemote('telnumber', ['01234']+invalid)

   215         d = self.assertFailure(d, Fault)

   216         d.addCallback(f)

   217         dl.append(d)

   218 

   219         return defer.DeferredList(dl, fireOnOneErrback=True)

   220 

   221     def testVaildEmail(self):

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

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

   224         d = self.proxy().callRemote('email', validmails)

   225         d.addCallback(lambda x: self.failUnlessEqual(x,True))

   226         return d

   227 

   228     def testInvaildEmail(self):

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

   230         invalid=["Abc.example.com",'foo@t.de']

   231         def f(exc):

   232             self.failUnlessEqual(exc.faultCode, 702)

   233             self.failUnlessEqual(exc.faultString, "No valid email: '%s'" % invalid[0])

   234         d = self.proxy().callRemote('email', invalid)

   235         d = self.assertFailure(d, Fault)

   236         d.addCallback(f)

   237         return d

   238 

   239     def testBill(self):

   240         '''test bill function'''

   241         apikey='abcdef123456789'

   242         with self.session() as session:

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

   244             session.add(u)

   245         d = self.proxy().callRemote('bill', apikey)

   246         d.addCallback(lambda x: self.failUnlessEqual(x,{'total':{'price':0.0,'anz':0}}))

   247 

   248         return d

   249 

   250     def testBillWithPrice(self):

   251         apikey='abcdef123456789'

   252         with self.session() as session:

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

   254             session.add(u)

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

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

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

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

   259             u.jobs.append(j)

   260 

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

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

   263             u.jobs.append(j)

   264 

   265         def f(ret):

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

   267             self.failUnlessEqual(ret['sipgate_basic'],

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

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

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

   271                             }

   272                         })

   273 

   274         d = self.proxy().callRemote('bill', apikey)

   275         d.addCallback(f)

   276         return d

   277 

   278 if __name__ == '__main__':

   279     unittest.main()