1 # -*- coding: utf-8 -*- |
|
2 |
|
3 import unittest, ConfigParser |
|
4 import iro |
|
5 from job import SMSJob, FAXJob, MailJob |
|
6 from joblist import Joblist |
|
7 from providerlist import Providerlist |
|
8 import threading, xmlrpclib |
|
9 from multiprocessing import Queue |
|
10 from multiprocessing.managers import BaseManager |
|
11 |
|
12 class StoppableXMLRPCServer(iro.SecureUserDBXMLRPCServer, threading.Thread): |
|
13 running=False |
|
14 def __init__(self, *args, **kwargs): |
|
15 iro.SecureUserDBXMLRPCServer.__init__(self, *args, **kwargs) |
|
16 threading.Thread.__init__(self) |
|
17 self.timeout=.5 |
|
18 |
|
19 def start(self): |
|
20 self.running=True |
|
21 threading.Thread.start(self) |
|
22 |
|
23 |
|
24 def run(self): |
|
25 # *serve_forever* muss in einem eigenen Thread laufen, damit man es |
|
26 # unterbrochen werden kann! |
|
27 while (self.running): |
|
28 self.handle_request() |
|
29 |
|
30 def stop(self): |
|
31 if (self.running): |
|
32 self.running=False |
|
33 self.server_close() |
|
34 self.join() |
|
35 |
|
36 def __del__(self): |
|
37 self.stop() |
|
38 |
|
39 def __enter__(self): |
|
40 self.start() |
|
41 return self |
|
42 |
|
43 def __exit__(self,type,value,traceback): |
|
44 self.stop() |
|
45 |
|
46 |
|
47 |
|
48 def init_server(): |
|
49 userlist=[{"name":"test","password":"test", "class":iro.User}, |
|
50 {"name":"test2","password":"test2", "class": iro.User}, |
|
51 {"name":"admin","password":"admin", "class": iro.Admin}] |
|
52 |
|
53 |
|
54 |
|
55 class MyManager(BaseManager): |
|
56 pass |
|
57 |
|
58 MyManager.register('SMSJob', SMSJob) |
|
59 MyManager.register('FAXob', FAXJob) |
|
60 MyManager.register('MailJob',MailJob) |
|
61 MyManager.register('Providerlist',Providerlist) |
|
62 manager = MyManager() |
|
63 manager.start() |
|
64 |
|
65 |
|
66 #anbieter erzeugen und konfigurieren |
|
67 import anbieter |
|
68 sip=anbieter.sipgate() |
|
69 sip.read_basic_config("iro.conf") |
|
70 |
|
71 localhost=iro.MySMTP() |
|
72 localhost.read_basic_config("iro.conf") |
|
73 |
|
74 smstrade=iro.MySmstrade() |
|
75 smstrade.read_basic_config("iro.conf") |
|
76 |
|
77 #Benutzerdatenbank erstellen |
|
78 queue = Queue() |
|
79 provider=Providerlist() |
|
80 provider.add("sipgate", sip, ["sms", "fax", ]) |
|
81 provider.add("smstrade", smstrade, ["sms", ]) |
|
82 provider.add("geonet", None, ["sms", "fax", ]) |
|
83 provider.add("fax.de", None, ["sms", "fax", ]) |
|
84 provider.add("localhost", localhost, ["mail", ]) |
|
85 provider.setDefault("sms","smstrade") |
|
86 provider.setDefault("fax","sipgate") |
|
87 provider.setDefault("mail","localhost") |
|
88 jobqueue=Joblist(manager, queue, provider) |
|
89 userdb=iro.MyUserDB(userlist,jobqueue) |
|
90 |
|
91 |
|
92 #Server starten |
|
93 cp = ConfigParser.ConfigParser() |
|
94 cp.read(["iro.conf"]) |
|
95 cert=cp.get('server', 'cert') |
|
96 key=cp.get('server', 'key') |
|
97 serv = StoppableXMLRPCServer(addr=("localhost", 8000), |
|
98 userdb=userdb, |
|
99 certificate=cert,privatekey=key, |
|
100 logRequests=False) |
|
101 serv.relam="xmlrpc" |
|
102 return serv |
|
103 |
|
104 |
|
105 class TestServer(unittest.TestCase): |
|
106 |
|
107 def setUp(self): |
|
108 self.serv = init_server() |
|
109 self.serv.start() |
|
110 |
|
111 def tearDown(self): |
|
112 self.serv.stop() |
|
113 |
|
114 def testLogin(self): |
|
115 self.assertEqual(xmlrpclib.Server("https://test:test@localhost:8000").status(), {}) |
|
116 self.assertEqual(xmlrpclib.Server("https://test2:test2@localhost:8000").status(), {}) |
|
117 self.assertRaises(xmlrpclib.ProtocolError, xmlrpclib.Server("https://test2:test@localhost:8000").status) |
|
118 self.assertRaises(xmlrpclib.ProtocolError,xmlrpclib.Server ("https://test:test2@localhost:8000").status) |
|
119 |
|
120 def testsendSMS(self): |
|
121 servstr="https://test:test@localhost:8000" |
|
122 client=xmlrpclib.Server(servstr) |
|
123 id=client.startSMS("test",["01234", ] ) |
|
124 self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': 'test'}} ) |
|
125 |
|
126 def testTwoUser(self): |
|
127 u1="https://test:test@localhost:8000" |
|
128 u2="https://test2:test2@localhost:8000" |
|
129 admin="https://admin:admin@localhost:8000" |
|
130 client1=xmlrpclib.Server(u1) |
|
131 client2=xmlrpclib.Server(u2) |
|
132 admin=xmlrpclib.Server(admin) |
|
133 id1=client1.startSMS("test",["01234"] ) |
|
134 self.assertEqual(client2.status(),{} ) |
|
135 self.assertEqual(admin.status(id1),{id1: {'status': ['init',{}], 'name': 'test'}} ) |
|
136 id2=client2.startSMS("test2",["01234"] ) |
|
137 self.assertNotEqual(id1, id2) |
|
138 self.assertEqual(client1.status(),{id1: {'status': ['init',{}], 'name': 'test'}}) |
|
139 self.assertEqual(client2.status(),{id2: {'status': ['init',{}], 'name': 'test2'}}) |
|
140 self.assertEqual(admin.status(),{id1: {'status': ['init',{}], 'name': 'test'}, |
|
141 id2: {'status': ['init',{}], 'name': 'test2'}} ) |
|
142 |
|
143 self.assertEqual(client2.status(id1), {}) |
|
144 self.assertEqual(client1.status(id2), {}) |
|
145 |
|
146 def testGetProvider(self): |
|
147 servstr="https://test:test@localhost:8000" |
|
148 client=xmlrpclib.Server(servstr) |
|
149 self.assertEqual(client.getProvider("sms"), ["fax.de","geonet", "sipgate", "smstrade"]) |
|
150 self.assertEqual(client.getProvider("fax"), ["fax.de","geonet", "sipgate"]) |
|
151 self.assertEqual(client.getProvider("mail"), ["localhost"]) |
|
152 |
|
153 self.assertRaises(xmlrpclib.ProtocolError,client.getProvider, "temp") |
|
154 |
|
155 def testGetDefault(self): |
|
156 servstr="https://test:test@localhost:8000" |
|
157 client=xmlrpclib.Server(servstr) |
|
158 self.assertEqual(client.getDefaultProvider("sms"), "smstrade") |
|
159 self.assertEqual(client.getDefaultProvider("fax"),"sipgate") |
|
160 self.assertEqual(client.getDefaultProvider("mail"), "localhost") |
|
161 |
|
162 self.assertRaises(xmlrpclib.ProtocolError,client.getDefaultProvider, "temp") |
|
163 |
|
164 |
|
165 if __name__ == "__main__": |
|
166 unittest.main() |
|