changeset 8:685479d1f0a7

Tests available for both py2 and py3
author Ben Croston <ben@croston.org>
date Sun, 04 Sep 2011 23:37:41 +0100
parents 3a6f3193cc7d
children 23743b0f67ab
files setup.py wibble/client/ServerProxy.py wibble/tests.py
diffstat 3 files changed, 51 insertions(+), 44 deletions(-) [+]
line wrap: on
line diff
--- a/setup.py	Thu Sep 01 19:19:03 2011 +0100
+++ b/setup.py	Sun Sep 04 23:37:41 2011 +0100
@@ -21,7 +21,6 @@
 if platform.python_version().startswith('2'):
     # we can build server with python 2
     install_requires.append('webob>=1.0.0')
-    extra['test_suite'] = 'wibble.tests.suite'
 
 if platform.python_version().startswith('3'):
     # we can't build server with python 3
@@ -41,5 +40,6 @@
       url              = 'http://www.wyre-it.co.uk/wibble/',
       classifiers      = classifiers,
       platforms        = ['Any'],
+      test_suite       = 'wibble.tests.suite',
       **extra)
 
--- a/wibble/client/ServerProxy.py	Thu Sep 01 19:19:03 2011 +0100
+++ b/wibble/client/ServerProxy.py	Sun Sep 04 23:37:41 2011 +0100
@@ -27,6 +27,12 @@
 import copy
 import socket
 import hashlib
+import platform
+
+if platform.python_version().startswith('3'):
+    IS_PY3 = True
+else:
+    IS_PY3 = False
 
 class _Method(object):
     def __init__(self, call, name, username=None, password=None):
@@ -55,7 +61,10 @@
         if self._username is not None:
             request['username'] = self._username
         if self._password is not None:
-            request['password'] = hashlib.md5(self._password).hexdigest()
+            if IS_PY3:
+                request['password'] = hashlib.md5(self._password.encode()).hexdigest()
+            else:
+                request['password'] = hashlib.md5(self._password).hexdigest()
 
         resp = self.call(json.dumps(request))
         if resp is not None and resp['error'] is None and resp['id'] == request['id']:
@@ -144,7 +153,10 @@
         except socket.error:
             raise NetworkSocketException
         if response.status == 200:
-          return json.loads(response.read())
+            if IS_PY3:
+                return json.loads(response.read().decode())
+            else:
+                return json.loads(response.read())
         elif response.status == 400:
             raise BadRequestException
         elif response.status == 401:
@@ -154,7 +166,10 @@
         elif response.status == 404:
             raise NotFoundException
         elif response.status == 500:
-            msg = json.loads(response.read())
+            if IS_PY3:
+                msg = json.loads(response.read().decode())
+            else:
+                msg = json.loads(response.read())
             raise Exception('JSONRPCError\n%s'%msg['error']['error'])
         elif response.status == 502:
             raise BadGatewayException
@@ -173,20 +188,3 @@
         # magic method dispatcher
         return _Method(self.__request, name, self._username, self._password)
 
-if __name__ == '__main__':
-    ##### btc fixme
-    jsonrpc_client = ServerProxy('http://localhost:1337/', username='testuser', password='', user_agent='Py2NotInternetExploiter')
-    #jsonrpc_client = ServerProxy('https://www.croston.org/test/index.py',
-    #                             username='testuser',
-    #                             password='',
-    #                             user_agent='Py2NotInternetExploiter')
-    assert jsonrpc_client.api.mymethod() == jsonrpc_client.mymethod()
-    try:
-        print jsonrpc_client.wibble('this should fail')
-    except BadRequestException:
-        pass # test passed
-    else:
-       raise Exception('Test failed (calling unknown method)')
-
-    print 'All tests passed'
-
--- a/wibble/tests.py	Thu Sep 01 19:19:03 2011 +0100
+++ b/wibble/tests.py	Sun Sep 04 23:37:41 2011 +0100
@@ -8,19 +8,19 @@
 
 ##### server vvv #####
 class api(object):
-  def mymethod(self):
-      return 'wibbler woz ere'
+    def mymethod(self):
+        return 'wibbler woz ere'
 
-  def echo(self, mystring):
-      return 'ECHO: ' + mystring
+    def echo(self, mystring):
+        return 'ECHO: ' + mystring
 
-  def raiseexception(self):
-      dividebyzeroerror = 1/0
+    def raiseexception(self):
+        dividebyzeroerror = 1/0
 
 def myauth(username, password, useragent=None):
-  return username == 'testuser' and \
-         hashlib.md5('s3cr3t').hexdigest() == password and \
-         useragent == 'wibble_unittest'
+    return username == 'testuser' and \
+           hashlib.md5('s3cr3t').hexdigest() == password and \
+           useragent == 'wibble_unittest'
 
 def make_server():
     from server import JsonRpcApp
@@ -38,7 +38,7 @@
         self.client = ServerProxy('http://localhost:1337/',
                                   username='testuser',
                                   password='s3cr3t',
-                                  user_agent='InternerExploiter')
+                                  user_agent='InternetExploiter')
         with self.assertRaises(UnauthorisedException):
             self.assertEqual(self.client.api.mymethod(),self.client.mymethod())
 
@@ -67,25 +67,25 @@
 
 class EchoTest(WibbleTests):
     def runTest(self):
-        POUND = '\xc2\xa3'
+        if platform.python_version().startswith('3'):
+            POUND = '\u00A3'
+        else:
+            POUND = unicode('\u00A3')
         self.assertEqual(self.client.echo(POUND), 'ECHO: ' + POUND)
         self.assertEqual(self.client.echo('hello mum!'), 'ECHO: hello mum!')
 ##### client ^^^ #####
 
 finished = False
 def suite():
-    if platform.python_version().startswith('3'):
-        # no tests for python 3 because server not ported yet
-        return unittest.TestSuite()
-
-    # create server
-    def test_wrapper():
-        server = make_server()
-        while not finished:
-            server.handle_request()
-    thread = Thread(target=test_wrapper)
-    thread.start()
-    time.sleep(0.1) # wait for server thread to start
+    if platform.python_version().startswith('2'):
+        # create server
+        def test_wrapper():
+            server = make_server()
+            while not finished:
+                server.handle_request()
+        thread = Thread(target=test_wrapper)
+        thread.start()
+        time.sleep(0.1) # wait for server thread to start
 
     # tests are as client
     suite = unittest.TestSuite()
@@ -97,6 +97,15 @@
     return suite
 
 if __name__ == '__main__':
+    import sys
+    if platform.python_version().startswith('2 ') and 'serve' in sys.argv:
+        print 'Listening on port 1337 (Ctrl-C qo quit)...'
+        server = make_server()
+        try:
+            server.serve_forever()
+        except KeyboardInterrupt:
+            sys.exit()
+
     unittest.TextTestRunner(verbosity=2).run(suite())
     finished = True