|
1 from twisted.internet.defer import inlineCallbacks |
|
2 from datetime import datetime |
|
3 |
|
4 from iro.model.schema import User, Offer, Userright, Job, Message |
|
5 from iro.controller.viewinterface import Interface |
|
6 from iro.model.pool import data |
|
7 |
|
8 import iro.error as IroError |
|
9 |
|
10 from .dbtestcase import DBTestCase |
|
11 |
|
12 class DummyPool(): |
|
13 def run(self, f,*a,**k): |
|
14 return f(*a,**k) |
|
15 |
|
16 class ViewInterfaceTest(DBTestCase): |
|
17 """tests for the xmlrpc interface""" |
|
18 def setUp(self): |
|
19 DBTestCase.setUp(self) |
|
20 self.pool = data.pool |
|
21 data.pool = DummyPool() |
|
22 |
|
23 def tearDown(self): |
|
24 data.pool = self.pool |
|
25 self.pool = None |
|
26 DBTestCase.tearDown(self) |
|
27 |
|
28 @inlineCallbacks |
|
29 def testStatus(self): |
|
30 ''' test the status function''' |
|
31 with self.session() as session: |
|
32 u = User(name='test',apikey='abcdef123456789') |
|
33 session.add(User(name='test',apikey='abcdef123456789')) |
|
34 st = yield Interface().status('abcdef123456789') |
|
35 self.assertEqual(st, {}) |
|
36 |
|
37 with self.session() as session: |
|
38 u = session.merge(u) |
|
39 j = Job(info='info', status="started") |
|
40 j.user=u |
|
41 session.add(j) |
|
42 session.commit() |
|
43 jid=j.id |
|
44 status = {str(jid):{"status":"started"}} |
|
45 st = yield Interface().status('abcdef123456789',jid) |
|
46 self.assertEqual(st, status) |
|
47 st = yield Interface().status('abcdef123456789') |
|
48 self.assertEqual(st, status) |
|
49 st = yield Interface().status('abcdef123456789', '', 'false') |
|
50 self.assertEqual(st, status) |
|
51 st = yield Interface().status('abcdef123456789', '', 0) |
|
52 self.assertEqual(st, status) |
|
53 |
|
54 #JobNotFound |
|
55 d = Interface().status('abcdef123456789',jid+1) |
|
56 self.assertFailure(d, IroError.JobNotFound) |
|
57 yield d |
|
58 |
|
59 #self.assertEqual(self.__rpc2().status('abcdef123456789','abcde', True), ["<User('test','abcdef123456789')>",'abcde', True]) |
|
60 #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 'true'), ["<User('test','abcdef123456789')>", '', True]) |
|
61 #self.assertEqual(self.__rpc2().status('abcdef123456789', '', 1), ["<User('test','abcdef123456789')>", '', True]) |
|
62 |
|
63 def testNoSuchUser(self): |
|
64 '''a unknown user should raise a UserNotNound Exception |
|
65 bewcause xmlrpc only has a Fault exception this Exception has to be deliverd through a xmlrpclib.Fault Exception''' |
|
66 d = Interface().status('abcdef123456789') |
|
67 self.assertFailure(d, IroError.UserNotFound) |
|
68 return d |
|
69 |
|
70 |
|
71 def testValidationFault(self): |
|
72 '''a validate Exception should be translated to a xmlrpclib.Fault.''' |
|
73 d = Interface().status('xxx') |
|
74 self.assertFailure(d, IroError.ValidateException) |
|
75 |
|
76 @inlineCallbacks |
|
77 def testRoutes(self): |
|
78 '''test the route function''' |
|
79 with self.session() as session: |
|
80 u=User(name='test',apikey='abcdef123456789') |
|
81 o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms") |
|
82 u.rights.append(Userright(o)) |
|
83 session.add(u) |
|
84 r = yield Interface().routes('abcdef123456789','sms') |
|
85 self.assertEqual(r, ['sipgate_basic']) |
|
86 |
|
87 d = Interface().routes('abcdef123456789','fax') |
|
88 self.assertFailure(d,IroError.ValidateException) |
|
89 yield d |
|
90 |
|
91 with self.session() as session: |
|
92 o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms") |
|
93 u = session.query(User).filter_by(name="test").first() |
|
94 u.rights.append(Userright(o)) |
|
95 o=Offer(name="faxde", provider="faxde", route="", typ="fax") |
|
96 session.add(o) |
|
97 session.commit() |
|
98 |
|
99 r = yield Interface().routes('abcdef123456789','sms') |
|
100 self.assertEqual(r, ['sipgate_basic','sipgate_plus']) |
|
101 r = yield Interface().routes('abcdef123456789','fax') |
|
102 self.assertEqual(r, []) |
|
103 |
|
104 |
|
105 with self.session() as session: |
|
106 u = session.query(User).filter_by(name="test").first() |
|
107 u.rights.append(Userright(o)) |
|
108 |
|
109 r = yield Interface().routes('abcdef123456789','sms') |
|
110 self.assertEqual(r, ['sipgate_basic','sipgate_plus']) |
|
111 r = yield Interface().routes('abcdef123456789','fax') |
|
112 self.assertEqual(r, ['faxde']) |
|
113 |
|
114 @inlineCallbacks |
|
115 def testDefaultRoutes(self): |
|
116 '''test the defaultRoute function''' |
|
117 with self.session() as session: |
|
118 u=User(name='test',apikey='abcdef123456789') |
|
119 o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms") |
|
120 u.rights.append(Userright(o,True)) |
|
121 o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms") |
|
122 u.rights.append(Userright(o)) |
|
123 session.add(u) |
|
124 r = yield Interface().defaultRoute('abcdef123456789','sms') |
|
125 self.assertEqual(r, ['sipgate_basic']) |
|
126 |
|
127 @inlineCallbacks |
|
128 def testTelnumbers(self): |
|
129 '''test the telefon validator''' |
|
130 r = yield Interface().telnumber(["0123/456(78)","+4912346785433","00123435456-658"]) |
|
131 self.assertEqual(r, True) |
|
132 |
|
133 invalid=['xa','+1','1-23',';:+0','0123'] |
|
134 |
|
135 d = Interface().telnumber(['01234']+invalid) |
|
136 def x(failure): |
|
137 self.assertEqual(failure.getErrorMessage(),"701:No valid telnumber: '%s'"%invalid[0]) |
|
138 failure.raiseException() |
|
139 d.addErrback(x) |
|
140 self.assertFailure(d, IroError.InvalidTel) |
|
141 yield d |
|
142 |
|
143 @inlineCallbacks |
|
144 def testVaildEmail(self): |
|
145 '''test vaild email adresses (got from wikipedia)''' |
|
146 validmails=["niceandsimple@example.com"] |
|
147 r = yield Interface().email(validmails) |
|
148 self.assertEqual(r,True) |
|
149 |
|
150 def testInvaildEmail(self): |
|
151 '''test invaild email adresses (got from wikipedia)''' |
|
152 invalid=["Abc.example.com",] |
|
153 d = Interface().email(invalid) |
|
154 self.assertFailure(d, IroError.InvalidMail) |
|
155 return d |
|
156 |
|
157 @inlineCallbacks |
|
158 def testBill(self): |
|
159 '''test bill function''' |
|
160 apikey='abcdef123456789' |
|
161 with self.session() as session: |
|
162 u=User(name='test',apikey=apikey) |
|
163 session.add(u) |
|
164 |
|
165 r = yield Interface().bill(apikey) |
|
166 self.assertEqual(r,{'total':{'price':0.0,'anz':0}}) |
|
167 |
|
168 with self.session() as session: |
|
169 u = session.merge(u) |
|
170 o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") |
|
171 u.rights.append(Userright(o)) |
|
172 j = Job(info='i',status='sended') |
|
173 j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o)) |
|
174 u.jobs.append(j) |
|
175 |
|
176 j = Job(info='a',status='sended') |
|
177 j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o)) |
|
178 u.jobs.append(j) |
|
179 |
|
180 ret=yield Interface().bill(apikey) |
|
181 self.assertEqual(ret['total'],{'price':0.8,'anz':2}) |
|
182 self.assertEqual(ret['sipgate_basic'], |
|
183 {'price':0.8,'anz':2, |
|
184 'info':{'i':{'price':0.4,'anz':1}, |
|
185 'a':{'price':0.4,'anz':1}, |
|
186 } |
|
187 }) |
|
188 |
|
189 def testSMS(self): |
|
190 pass |
|
191 testSMS.todo = "To implement" |
|
192 |
|
193 |
|
194 def testMail(self): |
|
195 pass |
|
196 testMail.todo = "To implement" |
|
197 |
|
198 def testFax(self): |
|
199 pass |
|
200 testFax.todo = "To implement" |