--- a/tests/xmlrpc.py Mon Jan 30 06:52:46 2012 +0100
+++ b/tests/xmlrpc.py Mon Jan 30 21:02:03 2012 +0100
@@ -2,25 +2,26 @@
from sqlalchemy import create_engine, pool
import unittest
-
from tempfile import mkdtemp
import shutil
+from datetime import datetime
+
+import time
+
+from xmlrpclib import Server as xServer, ServerProxy, Fault
+
from iro.model.utils import WithSession
from iro.model import POOL_SIZE as DB_POOL_SIZE
-from iro.model.schema import User, Base, Offer, Userright, Job
+from iro.model.schema import User, Base, Offer, Userright, Job, Message
import iro.model.schema as schema
-from ngdatabase.mysql import Server, createConfig, Database
-
from iro.main import runReactor
import iro.error as IroError
-import time
-
-from xmlrpclib import Server as xServer, ServerProxy, Fault
+from ngdatabase.mysql import Server, createConfig, Database
class SampleDatabase(Database):
def createPassword(self):
@@ -69,7 +70,7 @@
def testListMethods(self):
'''list of all offical Methods, that can be executed'''
ret=self.__rpc2().listMethods()
- self.failUnlessEqual(ret, ['listMethods', 'status', 'stop', 'sms', 'fax', 'mail', 'routes', 'defaultRoute', 'statistic', 'telnumber','email'])
+ self.failUnlessEqual(ret, ['listMethods', 'status', 'stop', 'sms', 'fax', 'mail', 'routes', 'defaultRoute', 'bill', 'telnumber','email'])
def testStatus(self):
''' test the status function'''
@@ -227,8 +228,72 @@
exc = fault.exception
self.failUnlessEqual(exc.faultCode, 702)
self.failUnlessEqual(exc.faultString, "No valid email: '%s'"%number)
+
+ def testBill(self):
+ '''test bill function'''
+ apikey='abcdef123456789'
+ with WithSession(md.engine, autocommit=True) as session:
+ u=User(name='test',apikey=apikey)
+ session.add(u)
+ self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.0,'anz':0}})
+
+ with WithSession(md.engine, autocommit=True) as session:
+ u = session.merge(u)
+ o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")
+ j = Job(hash='a1',info='i',status='sended')
+ m = Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.30, offer=o, job=j)
+ u.rights.append(Userright(o))
+ u.jobs.append(j)
+ session.add(m)
+
+ self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.3,'anz':1},
+ 'sipgate_basic':{'price':0.3,'anz':1,'info':{'i':{'price':0.3,'anz':1}}}})
+
+ with WithSession(md.engine, autocommit=True) as session:
+ j = session.merge(j)
+ j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o))
+
+ self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.7,'anz':2},
+ 'sipgate_basic':{'price':0.7,'anz':2,'info':{'i':{'price':0.7,'anz':2}}}})
+
+ with WithSession(md.engine, autocommit=True) as session:
+ m = session.merge(m)
+ m.isBilled=True
+
+ self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.4,'anz':1},
+ 'sipgate_basic':{'price':0.4,'anz':1,'info':{'i':{'price':0.4,'anz':1}}}})
-
+ with WithSession(md.engine, autocommit=True) as session:
+ u = session.merge(u)
+ j = Job(hash='a2',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=self.__rpc2().bill(apikey)
+ self.failUnlessEqual(ret['total'],{'price':0.8,'anz':2})
+ self.failUnlessEqual(ret['sipgate_basic'],
+ {'price':0.8,'anz':2,
+ 'info':{'i':{'price':0.4,'anz':1},
+ 'a':{'price':0.4,'anz':1},
+ }
+ })
+
+ with WithSession(md.engine, autocommit=True) as session:
+ u = session.merge(u)
+ j = Job(hash='a3',info='a',status='sended')
+ o = Offer(name='sipgate_gold',provider="sipgate",route="gold",typ="sms")
+ j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.5, offer=o))
+ u.rights.append(Userright(offer=o))
+ u.jobs.append(j)
+
+ ret=self.__rpc2().bill(apikey)
+ self.failUnlessEqual(ret['total'],{'price':1.3,'anz':3})
+ self.failUnlessEqual(ret['sipgate_gold'],
+ {'price':0.5,'anz':1,
+ 'info':{
+ 'a':{'price':0.5,'anz':1},
+ }
+ })
+
def startReactor(engine):
"""starts the Rector with a special debug Clild, so that the reactor can be stopped remotly. """
from twisted.internet import reactor
@@ -256,7 +321,7 @@
self.server = Server('%s/my.cnf'%self.tdir)
self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)
self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir,
- poolclass = pool.SingletonThreadPool, pool_size=DB_POOL_SIZE, )
+ poolclass = pool.SingletonThreadPool, pool_size=DB_POOL_SIZE, )#echo=True)
def setUp(self):
with open('%s/my.cnf'%self.tdir,'w') as cnf: