Merge pull request samuraisam#26 from alendit/testing-proxy

Added TestingServiceProxy, which works with Django unittests and refactor

jsonrpc/proxy.py

@@ -1,41 +1,68 @@
 import urllib, uuid
+from django.test.client import FakePayload
+
 from jsonrpc._json import loads, dumps
 from jsonrpc.types import *
 
 class ServiceProxy(object):
   def __init__(self, service_url, service_name=None, version='1.0'):
-    self.__version = str(version)
-    self.__service_url = service_url
-    self.__service_name = service_name
+    self.version = str(version)
+    self.service_url = service_url
+    self.service_name = service_name
 
   def __getattr__(self, name):
-    if self.__service_name != None:
-      name = "%s.%s" % (self.__service_name, name)
-    return ServiceProxy(self.__service_url, name, self.__version)
+    if self.service_name != None:
+      name = "%s.%s" % (self.service_name, name)
+    params = dict(self.__dict__, service_name=name)
+    return self.__class__(**params)
 
   def __repr__(self):
-    return {"jsonrpc": self.__version,
-            "method": self.__service_name}
-
+    return {"jsonrpc": self.version,
+            "method": self.service_name}
+
+  def send_payload(self, params):
+      """Performs the actual sending action and returns the result"""
+      return urllib.urlopen(self.service_url,
+                    dumps({
+                      "jsonrpc": self.version,
+                      "method": self.service_name,
+                      'params': params,
+                      'id': str(uuid.uuid1())})).read()
+
   def __call__(self, *args, **kwargs):
     params = kwargs if len(kwargs) else args
-    if Any.kind(params) == Object and self.__version != '2.0':
+    if Any.kind(params) == Object and self.version != '2.0':
       raise Exception('Unsupport arg type for JSON-RPC 1.0 '
                      '(the default version for this client, '
                      'pass version="2.0" to use keyword arguments)')
-    # req = urllib2.Request(self.__service_url, )
-    r = urllib.urlopen(self.__service_url,
-                        dumps({
-                          "jsonrpc": self.__version,
-                          "method": self.__service_name,
-                          'params': params,
-                          'id': str(uuid.uuid1())})).read()
+
+    r = self.send_payload(params)    
     y = loads(r)
     if u'error' in y:
       try:
         from django.conf import settings
         if settings.DEBUG:
-            print '%s error %r' % (self.__service_name, y)
+            print '%s error %r' % (self.service_name, y)
       except:
         pass
     return y
+
+class TestingServiceProxy(ServiceProxy):
+    """Service proxy which works inside Django unittests"""
+
+    def __init__(self, client, *args, **kwargs):
+        super(TestingServiceProxy, self).__init__(*args, **kwargs)
+        self.client = client
+
+    def send_payload(self, params):
+        dump = dumps({"jsonrpc" : self.version,
+                       "method" : self.service_name,
+                       "params" : params,
+                       "id" : str(uuid.uuid1())
+                       })
+        dump_payload = FakePayload(dump)
+        response = self.client.post(self.service_url,
+                          **{"wsgi.input" : dump_payload,
+                          'CONTENT_LENGTH' : len(dump)})
+        return response.content
+

test/test.py

@@ -172,7 +172,7 @@ def test_types(self):
 
 proc = None
 
-class ServiceProxyText(unittest.TestCase):      
+class ServiceProxyTest(unittest.TestCase):      
   def setUp(self):
     global proc
     if proc is None: