Browse Source

test: add ensure_ascii setting to AuthServiceProxy

Add a setting ensure_ascii to AuthServiceProxy. This setting,
defaulting to True (backwards compatible),
is passed through to json.dumps. If set to False, non-ASCII characters
>0x80 are not escaped. This is useful for testing server
input processing, as well as slightly more bandwidth friendly in case of
heavy unicode usage.
0.13
Wladimir J. van der Laan 9 years ago
parent
commit
a406fcb6ca
  1. 16
      qa/rpc-tests/test_framework/authproxy.py

16
qa/rpc-tests/test_framework/authproxy.py

@ -67,9 +67,11 @@ def EncodeDecimal(o):
class AuthServiceProxy(object): class AuthServiceProxy(object):
__id_count = 0 __id_count = 0
def __init__(self, service_url, service_name=None, timeout=HTTP_TIMEOUT, connection=None): # ensure_ascii: escape unicode as \uXXXX, passed to json.dumps
def __init__(self, service_url, service_name=None, timeout=HTTP_TIMEOUT, connection=None, ensure_ascii=True):
self.__service_url = service_url self.__service_url = service_url
self._service_name = service_name self._service_name = service_name
self.ensure_ascii = ensure_ascii # can be toggled on the fly by tests
self.__url = urlparse.urlparse(service_url) self.__url = urlparse.urlparse(service_url)
if self.__url.port is None: if self.__url.port is None:
port = 80 port = 80
@ -134,12 +136,12 @@ class AuthServiceProxy(object):
AuthServiceProxy.__id_count += 1 AuthServiceProxy.__id_count += 1
log.debug("-%s-> %s %s"%(AuthServiceProxy.__id_count, self._service_name, log.debug("-%s-> %s %s"%(AuthServiceProxy.__id_count, self._service_name,
json.dumps(args, default=EncodeDecimal))) json.dumps(args, default=EncodeDecimal, ensure_ascii=self.ensure_ascii)))
postdata = json.dumps({'version': '1.1', postdata = json.dumps({'version': '1.1',
'method': self._service_name, 'method': self._service_name,
'params': args, 'params': args,
'id': AuthServiceProxy.__id_count}, default=EncodeDecimal) 'id': AuthServiceProxy.__id_count}, default=EncodeDecimal, ensure_ascii=self.ensure_ascii)
response = self._request('POST', self.__url.path, postdata) response = self._request('POST', self.__url.path, postdata.encode('utf-8'))
if response['error'] is not None: if response['error'] is not None:
raise JSONRPCException(response['error']) raise JSONRPCException(response['error'])
elif 'result' not in response: elif 'result' not in response:
@ -149,9 +151,9 @@ class AuthServiceProxy(object):
return response['result'] return response['result']
def _batch(self, rpc_call_list): def _batch(self, rpc_call_list):
postdata = json.dumps(list(rpc_call_list), default=EncodeDecimal) postdata = json.dumps(list(rpc_call_list), default=EncodeDecimal, ensure_ascii=self.ensure_ascii)
log.debug("--> "+postdata) log.debug("--> "+postdata)
return self._request('POST', self.__url.path, postdata) return self._request('POST', self.__url.path, postdata.encode('utf-8'))
def _get_response(self): def _get_response(self):
http_response = self.__conn.getresponse() http_response = self.__conn.getresponse()
@ -167,7 +169,7 @@ class AuthServiceProxy(object):
responsedata = http_response.read().decode('utf8') responsedata = http_response.read().decode('utf8')
response = json.loads(responsedata, parse_float=decimal.Decimal) response = json.loads(responsedata, parse_float=decimal.Decimal)
if "error" in response and response["error"] is None: if "error" in response and response["error"] is None:
log.debug("<-%s- %s"%(response["id"], json.dumps(response["result"], default=EncodeDecimal))) log.debug("<-%s- %s"%(response["id"], json.dumps(response["result"], default=EncodeDecimal, ensure_ascii=self.ensure_ascii)))
else: else:
log.debug("<-- "+responsedata) log.debug("<-- "+responsedata)
return response return response

Loading…
Cancel
Save