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