|
4
|
1 #!/usr/bin/env python
|
|
|
2 import unittest
|
|
|
3 import hashlib
|
|
|
4 from threading import Thread
|
|
|
5 import time
|
|
|
6 from wsgiref import simple_server
|
|
|
7 import platform
|
|
|
8
|
|
|
9 ##### server vvv #####
|
|
|
10 class api(object):
|
|
|
11 def mymethod(self):
|
|
|
12 return 'wibbler woz ere'
|
|
|
13
|
|
7
|
14 def echo(self, mystring):
|
|
|
15 return 'ECHO: ' + mystring
|
|
|
16
|
|
|
17 def raiseexception(self):
|
|
|
18 dividebyzeroerror = 1/0
|
|
5
|
19
|
|
4
|
20 def myauth(username, password, useragent=None):
|
|
5
|
21 return username == 'testuser' and \
|
|
|
22 hashlib.md5('s3cr3t').hexdigest() == password and \
|
|
|
23 useragent == 'wibble_unittest'
|
|
4
|
24
|
|
|
25 def make_server():
|
|
5
|
26 from server import JsonRpcApp
|
|
4
|
27 class myhandler(simple_server.WSGIRequestHandler):
|
|
|
28 def log_request(self, *a, **b):
|
|
|
29 pass # do not output log messages
|
|
|
30 application = JsonRpcApp(api(), auth=myauth)
|
|
|
31 return simple_server.make_server('localhost', 1337, application, handler_class=myhandler)
|
|
|
32 ##### server ^^^ #####
|
|
|
33
|
|
|
34 ##### client vvv #####
|
|
7
|
35 class NotAuthTest(unittest.TestCase):
|
|
|
36 def runTest(self):
|
|
|
37 from client.ServerProxy import ServerProxy, UnauthorisedException
|
|
|
38 self.client = ServerProxy('http://localhost:1337/',
|
|
|
39 username='testuser',
|
|
|
40 password='s3cr3t',
|
|
|
41 user_agent='InternerExploiter')
|
|
|
42 with self.assertRaises(UnauthorisedException):
|
|
|
43 self.assertEqual(self.client.api.mymethod(),self.client.mymethod())
|
|
|
44
|
|
4
|
45 class WibbleTests(unittest.TestCase):
|
|
|
46 def setUp(self):
|
|
5
|
47 from client.ServerProxy import ServerProxy
|
|
4
|
48 self.client = ServerProxy('http://localhost:1337/',
|
|
|
49 username='testuser',
|
|
|
50 password='s3cr3t',
|
|
|
51 user_agent='wibble_unittest')
|
|
|
52
|
|
5
|
53 class IgnoreClassNameTest(WibbleTests):
|
|
4
|
54 def runTest(self):
|
|
|
55 self.assertEqual(self.client.api.mymethod(),self.client.mymethod())
|
|
|
56
|
|
7
|
57 class ExceptionTest(WibbleTests):
|
|
|
58 def runTest(self):
|
|
|
59 with self.assertRaises(Exception):
|
|
|
60 self.client.raiseexception()
|
|
4
|
61
|
|
7
|
62 class BadRequestTest(WibbleTests):
|
|
|
63 def runTest(self):
|
|
|
64 from client.ServerProxy import BadRequestException
|
|
|
65 with self.assertRaises(BadRequestException):
|
|
|
66 self.client.FunctionDoesNotExist()
|
|
|
67
|
|
|
68 class EchoTest(WibbleTests):
|
|
|
69 def runTest(self):
|
|
|
70 POUND = '\xc2\xa3'
|
|
|
71 self.assertEqual(self.client.echo(POUND), 'ECHO: ' + POUND)
|
|
|
72 self.assertEqual(self.client.echo('hello mum!'), 'ECHO: hello mum!')
|
|
4
|
73 ##### client ^^^ #####
|
|
|
74
|
|
5
|
75 finished = False
|
|
4
|
76 def suite():
|
|
|
77 if platform.python_version().startswith('3'):
|
|
|
78 # no tests for python 3 because server not ported yet
|
|
|
79 return unittest.TestSuite()
|
|
7
|
80
|
|
|
81 # create server
|
|
4
|
82 def test_wrapper():
|
|
|
83 server = make_server()
|
|
5
|
84 while not finished:
|
|
|
85 server.handle_request()
|
|
4
|
86 thread = Thread(target=test_wrapper)
|
|
|
87 thread.start()
|
|
|
88 time.sleep(0.1) # wait for server thread to start
|
|
7
|
89
|
|
|
90 # tests are as client
|
|
4
|
91 suite = unittest.TestSuite()
|
|
7
|
92 suite.addTest(NotAuthTest())
|
|
5
|
93 suite.addTest(IgnoreClassNameTest())
|
|
7
|
94 suite.addTest(ExceptionTest())
|
|
|
95 suite.addTest(BadRequestTest())
|
|
|
96 suite.addTest(EchoTest())
|
|
4
|
97 return suite
|
|
|
98
|
|
|
99 if __name__ == '__main__':
|
|
5
|
100 unittest.TextTestRunner(verbosity=2).run(suite())
|
|
|
101 finished = True
|
|
4
|
102
|
|
5
|
103 # make a dummy request to get server thread out of loop
|
|
|
104 try:
|
|
|
105 import urllib
|
|
|
106 urllib.urlopen('http://localhost:1337/')
|
|
|
107 except:
|
|
|
108 pass
|
|
|
109
|
