# HG changeset patch # User Sandro Knauß # Date 1335304740 -7200 # Node ID 97826c8248f93ed2e3cc272aa6c8e7d5a26cead0 # Parent b218238e76b90b83a0e3e92c92443e518ac80c5f adding sipgate backend diff -r b218238e76b9 -r 97826c8248f9 iro/offer/sipgate.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/offer/sipgate.py Tue Apr 24 23:59:00 2012 +0200 @@ -0,0 +1,118 @@ +from functools import partial +from twisted.web.xmlrpc import Proxy +import xmlrpclib +from decimal import Decimal + +from .provider import providers, Provider +from ..config import Option +from ..validate import vFloat +from ..model.status import Status +from ..error import NoTyp +from .. import __version__ + + +class Sipgate(Provider): + """ + s. auch http.tuxad.com/sipgate.html + und http://lists.sipgate.net/pipermail/sipgate-developers/2007-September/000016.html + + http://www.sipgate.de/beta/public/static/downloads/basic/api/sipgate_api_documentation.pdf + + See :doc:`provider` for a tutorial, how to create a new provider backend. This turorial implements this provider. + """ + + url="https://%s:%s@samurai.sipgate.net/RPC2" + """XML-RPC url for sipgate""" + + def __init__(self,name): + options =[("username", Option(lambda x,y: x,long="Loginname for sipgate", must=True)), + ("password", Option(lambda x,y: x,long="Password for sipgate", must=True)), + ("price_sms", Option(vFloat, long="price for one sms", default="0.079")), + ("price_fax", Option(vFloat, long="price for one fax", default="0.03")), + ] + Provider.__init__(self, name, {"sms" : [None], "fax":[None]}, options) + + def proxy(self): + """returns a XML-RPC proxy object to sipgate API.""" + return Proxy(self.url%(self.username, self.password)) + + def load(self,cfg): + """Loads configuration into object. + + :param dict cfg: The Configuration dict. Normally you use ``configParser.items("section")``. + """ + + Provider.load(self,cfg) + #return self.clientIdentify() + + def clientIdentify(self): + """identificaton of client to sipgate server.""" + args_identify = { + "ClientName" : "Iro", + "ClientVersion" : __version__, + "ClientVendor" : "netzguerilla.net" + } + + return self.proxy().callRemote("samurai.ClientIdentify", args_identify) + + def sms(self, recipient, sms): + """send one SMS to recimpient. + + :param `iro.telnumber.Telnumber` recipient: mobilenumber of recipient + :param `iro.model.message.SMS` sms: the sms to send + :return: a deferrer + """ + args={ + "TOS" : "text", + "Content" : sms.getContent(), + "RemoteUri" : "sip:%s%s@sipgate.net"%(recipient.land, recipient.number), + } + return self.proxy().callRemote("samurai.SessionInitiate",args) + + def fax(self, recipient, fax): + """send one fax to recimpient. + + :param `iro.telnumber.Telnumber` recipient: faxnumber of recipient + :param `iro.model.message.Fax` fax: the fax to send + :return: a deferrer. + """ + + args={ + "TOS" : "fax", + "Content" : xmlrpclib.Binary(fax.getAttachment(0)), + "RemoteUri" : "sip:%s%s@sipgate.net"%(recipient.land, recipient.number), + } + return self.proxy().callRemote("samurai.SessionInitiate",args) + + def _status(self,value,typ): + """returns a :class:`~iro.model.status.Status` object. + :raises: :exc:`iro.error.NoTyp` + """ + if typ not in self.typs.keys(): + raise NoTyp(typ) + return Status(self, None, Decimal(getattr(self,"price_"+typ)), 1, value["SessionID"]) + + def send(self, typ, recipient, msg): + """send msg to recipient. + + :param string typ: Typ of message. + :param `iro.telnumber.Telnumber` recipient: telnumber of recipient + :param msg: the msg to send + :type msg: :class:`iro.model.message.Fax` or :class:`iro.model.message.SMS` + :return: a deferrer, that returns a :class:`~iro.model.status.Status` object + :raises: :exc:`iro.error.NoTyp` + """ + if typ not in self.typs.keys(): + raise NoTyp(typ) + d = getattr(self,typ)(recipient, msg) + d.addCallback(self._status, typ) + return d + + def getSendFunc(self, typ, route): + """returns :meth:`send` method with bound typ, if typ and route is valid.""" + + Provider.getSendFunc(self, typ, route) + return partial(self.send, typ) + +providers["sipgate"] = Sipgate + diff -r b218238e76b9 -r 97826c8248f9 iro/tests/sipgate.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/tests/sipgate.py Tue Apr 24 23:59:00 2012 +0200 @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +from twisted.trial import unittest +from twisted.web import xmlrpc, server +from twisted.internet import reactor +from decimal import Decimal + +from iro.error import NoRoute, NoTyp, NeededOption, RejectRecipient +from iro.telnumber import Telnumber +from iro.model.message import SMS, Fax +from iro.offer.sipgate import Sipgate + + +class XMLRPCServer(xmlrpc.XMLRPC): + def lookupProcedure(self, procedurePath): + return self.l[procedurePath] + +class TestSipgateProvider(unittest.TestCase): + + def setUp(self): + self.server = XMLRPCServer() + self.p = reactor.listenTCP(0, server.Site(self.server), + interface="127.0.0.1") + self.port = self.p.getHost().port + + def tearDown(self): + self.server = None + return self.p.stopListening() + + def getProvider(self, c=None): + _c={"username":"XXXXXXXX", + "password":"PPPPPPPP", + "typ": "sipgate" + } + + if c: + _c.update(c) + + ret = Sipgate("test") + ret.url="http://%%s:%%s@127.0.0.1:%d/" % self.port + ret.load(_c.items()) + return ret + + def testSendSMS(self): + p=self.getProvider() + content = u"Hello World - äüöß'\"" + + @xmlrpc.withRequest + def f(request, args): + self.assertEqual(request.getUser(),"XXXXXXXX") + self.assertEqual(request.getPassword(),"PPPPPPPP") + self.assertEqual(args,{'Content': u'Hello World - äüöß\'"', + 'TOS': 'text', + 'RemoteUri': 'sip:491701234567@sipgate.net'}) + return {"SessionID":"", "StatusCode":200, "StatusString":"Method success"} + + self.server.l = {"samurai.SessionInitiate": f} + + def s(r): + self.assertEqual(r.provider, p) + self.assertEqual(r.route, None) + self.assertEqual(r.costs, Decimal('0.079')) + self.assertEqual(r.exID, '') + self.assertEqual(r.count, 1) + + d = p.send("sms", Telnumber("01701234567"), SMS(content,None)) + d.addCallback(s) + return d + + + def testNeededOption(self): + s= self.getProvider() + self.assertEqual(s.username, "XXXXXXXX") + self.assertEqual(s.password, "PPPPPPPP") + self.assertEqual(s.price_sms, '0.079') + self.assertEqual(s.price_fax, '0.03') + + self.assertRaises(NeededOption, s.load,[]) + + def testSendFunc(self): + s = self.getProvider() + p = s.getSendFunc("sms",None) + self.assertEqual(p.func, s.send) + self.assertEqual(p.args, ("sms",)) + self.assertRaises(NoRoute,s.getSendFunc,"sms","foo") + self.assertRaises(NoTyp,s.getSendFunc,"mail2",None)