1 # -*- coding: utf-8 -*- |
1 # -*- coding: utf-8 -*- |
2 |
2 |
3 import unittest |
3 import unittest, ConfigParser |
4 import server |
4 import iro |
5 import threading, xmlrpclib |
5 import threading, xmlrpclib |
6 from multiprocessing import Queue |
6 from multiprocessing import Queue |
7 from multiprocessing.managers import BaseManager |
7 from multiprocessing.managers import BaseManager |
8 |
8 |
9 class StoppableXMLRPCServer(server.SecureUserDBXMLRPCServer, threading.Thread): |
9 class StoppableXMLRPCServer(iro.SecureUserDBXMLRPCServer, threading.Thread): |
10 running=True |
10 running=True |
11 def __init__(self, *args, **kwargs): |
11 def __init__(self, *args, **kwargs): |
12 server.SecureUserDBXMLRPCServer.__init__(self, *args, **kwargs) |
12 iro.SecureUserDBXMLRPCServer.__init__(self, *args, **kwargs) |
13 threading.Thread.__init__(self) |
13 threading.Thread.__init__(self) |
14 |
14 |
15 |
15 |
16 def run(self): |
16 def run(self): |
17 # *serve_forever* muss in einem eigenen Thread laufen, damit man es |
17 # *serve_forever* muss in einem eigenen Thread laufen, damit man es |
24 self.running=False |
24 self.running=False |
25 self.server_close() |
25 self.server_close() |
26 |
26 |
27 |
27 |
28 def init_server(): |
28 def init_server(): |
29 userlist=[{"name":"test","password":"test", "class":server.User}, |
29 userlist=[{"name":"test","password":"test", "class":iro.User}, |
30 {"name":"test2","password":"test2", "class": server.User}, |
30 {"name":"test2","password":"test2", "class": iro.User}, |
31 {"name":"admin","password":"admin", "class": server.Admin}] |
31 {"name":"admin","password":"admin", "class": iro.Admin}] |
32 |
32 |
33 |
33 |
34 |
34 |
35 class MyManager(BaseManager): |
35 class MyManager(BaseManager): |
36 pass |
36 pass |
37 |
37 |
38 MyManager.register('MessageJob', server.MessageJob) |
38 MyManager.register('MessageJob', iro.MessageJob) |
39 manager = MyManager() |
39 manager = MyManager() |
40 manager.start() |
40 manager.start() |
41 |
41 |
42 |
42 |
43 #anbieter erzeugen und konfigurieren |
43 #anbieter erzeugen und konfigurieren |
44 import anbieter |
44 import anbieter |
45 sip=anbieter.sipgate() |
45 sip=anbieter.sipgate() |
46 sip.read_basic_config("iro.conf") |
46 sip.read_basic_config("iro.conf") |
47 localhost="" |
47 |
|
48 localhost=iro.MySMTP() |
|
49 localhost.read_basic_config("iro.conf") |
|
50 |
|
51 smstrade=iro.MySmstrade() |
|
52 smstrade.read_basic_config("iro.conf") |
48 |
53 |
49 #Benutzerdatenbank erstellen |
54 #Benutzerdatenbank erstellen |
50 queue = Queue() |
55 queue = Queue() |
51 provider={"sms":sip, "fax":sip, "mail":localhost} |
56 provider={"sms":sip, "fax":sip, "mail":localhost} |
52 jobqueue=server.Jobs(manager, queue, provider) |
57 jobqueue=iro.Jobs(manager, queue, provider) |
53 userdb=server.MyUserDB(userlist,jobqueue) |
58 userdb=iro.MyUserDB(userlist,jobqueue) |
54 |
59 |
55 |
60 |
56 #Server starten |
61 #Server starten |
|
62 cp = ConfigParser.ConfigParser() |
|
63 cp.read(["iro.conf"]) |
|
64 cert=cp.get('server', 'cert') |
|
65 key=cp.get('server', 'key') |
57 serv = StoppableXMLRPCServer(addr=("localhost", 8000), |
66 serv = StoppableXMLRPCServer(addr=("localhost", 8000), |
58 userdb=userdb, |
67 userdb=userdb, |
59 certificate="./certs/test.cert.pem",privatekey="./certs/test.key.pem") |
68 certificate=cert,privatekey=key) |
60 serv.relam="xmlrpc" |
69 serv.relam="xmlrpc" |
61 return serv |
70 return serv |
62 |
71 |
63 |
72 |
64 class TestServer(unittest.TestCase): |
73 class TestServer(unittest.TestCase): |
76 self.assertEqual(xmlrpclib.Server("https://test2:test2@localhost:8000").status(), {}) |
85 self.assertEqual(xmlrpclib.Server("https://test2:test2@localhost:8000").status(), {}) |
77 self.assertRaises(xmlrpclib.ProtocolError, xmlrpclib.Server("https://test2:test@localhost:8000").status) |
86 self.assertRaises(xmlrpclib.ProtocolError, xmlrpclib.Server("https://test2:test@localhost:8000").status) |
78 self.assertRaises(xmlrpclib.ProtocolError,xmlrpclib.Server ("https://test:test2@localhost:8000").status) |
87 self.assertRaises(xmlrpclib.ProtocolError,xmlrpclib.Server ("https://test:test2@localhost:8000").status) |
79 |
88 |
80 def testsendSMS(self): |
89 def testsendSMS(self): |
81 servstr="https://test:test@localhost:8000" |
90 servstr="https://test:test@localhost:8000" |
82 client=xmlrpclib.Server(servstr) |
91 client=xmlrpclib.Server(servstr) |
83 id=client.startSMS("test",["01234"] ) |
92 id=client.startSMS("test",["01234"] ) |
84 self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': {'content': 'test'}}} ) |
93 self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': 'test'}} ) |
85 |
94 |
86 def testTwoUser(self): |
95 def testTwoUser(self): |
87 u1="https://test:test@localhost:8000" |
96 u1="https://test:test@localhost:8000" |
88 u2="https://test2:test2@localhost:8000" |
97 u2="https://test2:test2@localhost:8000" |
89 admin="https://admin:admin@localhost:8000" |
98 admin="https://admin:admin@localhost:8000" |
90 client1=xmlrpclib.Server(u1) |
99 client1=xmlrpclib.Server(u1) |
91 client2=xmlrpclib.Server(u2) |
100 client2=xmlrpclib.Server(u2) |
92 admin=xmlrpclib.Server(admin) |
101 admin=xmlrpclib.Server(admin) |
93 id1=client1.startSMS("test",["01234"] ) |
102 id1=client1.startSMS("test",["01234"] ) |
94 self.assertEqual(client2.status(),{} ) |
103 self.assertEqual(client2.status(),{} ) |
95 self.assertEqual(admin.status(id1),{id1: {'status': ['init',{}], 'name': {'content': 'test'}}} ) |
104 self.assertEqual(admin.status(id1),{id1: {'status': ['init',{}], 'name': 'test'}} ) |
96 id2=client2.startSMS("test2",["01234"] ) |
105 id2=client2.startSMS("test2",["01234"] ) |
97 self.assertNotEqual(id1, id2) |
106 self.assertNotEqual(id1, id2) |
98 self.assertEqual(client1.status(),{id1: {'status': ['init',{}], 'name': {'content': 'test'}}}) |
107 self.assertEqual(client1.status(),{id1: {'status': ['init',{}], 'name': 'test'}}) |
99 self.assertEqual(client2.status(),{id2: {'status': ['init',{}], 'name': {'content': 'test2'}}}) |
108 self.assertEqual(client2.status(),{id2: {'status': ['init',{}], 'name': 'test2'}}) |
100 self.assertEqual(admin.status(),{id1: {'status': ['init',{}], 'name': {'content': 'test'}}, |
109 self.assertEqual(admin.status(),{id1: {'status': ['init',{}], 'name': 'test'}, |
101 id2: {'status': ['init',{}], 'name': {'content': 'test2'}}} ) |
110 id2: {'status': ['init',{}], 'name': 'test2'}} ) |
102 |
111 |
103 self.assertRaises(xmlrpclib.Fault, client2.status, id1) |
112 self.assertRaises(xmlrpclib.Fault, client2.status, id1) |
104 self.assertRaises(xmlrpclib.Fault, client1.status, id2) |
113 self.assertRaises(xmlrpclib.Fault, client1.status, id2) |
|
114 |
|
115 def testGetProvider(self): |
|
116 servstr="https://test:test@localhost:8000" |
|
117 client=xmlrpclib.Server(servstr) |
|
118 self.assertEqual(client.getProvider("sms").sort(), ["fax.de","geonet", "sipgate", "smstrade"]) |
|
119 self.assertEqual(client.getProvider("fax").sort(), ["fax.de","geonet", "sipgate"]) |
|
120 self.assertEqual(client.getProvider("mail").sort(), ["localhost"]) |
105 |
121 |
106 if __name__ == "__main__": |
122 if __name__ == "__main__": |
107 unittest.main() |
123 unittest.main() |