@@ -123,24 +123,27 @@ class JSONRPCController(WSGIController):
except ValueError, e:
# catch JSON errors Here
return jsonrpc_error(retid=self._req_id,
message="JSON parse error ERR:%s RAW:%r" \
% (e, urllib.unquote_plus(raw_body)))
# check AUTH based on API KEY
try:
self._req_api_key = json_body['api_key']
self._req_id = json_body['id']
self._req_method = json_body['method']
self._request_params = json_body['args']
if not isinstance(self._request_params, dict):
self._request_params = {}
log.debug(
'method: %s, params: %s' % (self._req_method,
self._request_params)
)
except KeyError, e:
message='Incorrect JSON query missing %s' % e)
# check if we can find this session using api_key
u = User.get_by_api_key(self._req_api_key)
if u is None:
@@ -203,24 +206,25 @@ class JSONRPCController(WSGIController):
# skip the required param check if it's default value is
# NotImplementedType (default_empty)
if (default == default_empty and arg not in self._request_params):
return jsonrpc_error(
retid=self._req_id,
message=(
'Missing non optional `%s` arg in JSON DATA' % arg
self._rpc_args = {USER_SESSION_ATTR: u}
self._rpc_args.update(self._request_params)
self._rpc_args['action'] = self._req_method
self._rpc_args['environ'] = environ
self._rpc_args['start_response'] = start_response
status = []
headers = []
exc_info = []
def change_content(new_status, new_headers, new_exc_info=None):
status.append(new_status)
@@ -146,24 +146,52 @@ class BaseTestApi(object):
response = api_call(self, params)
expected = 'Invalid API KEY'
self._compare_error(id_, expected, given=response.body)
def test_api_missing_non_optional_param(self):
id_, params = _build_data(self.apikey, 'get_repo')
expected = 'Missing non optional `repoid` arg in JSON DATA'
def test_api_missing_non_optional_param_args_null(self):
params = params.replace('"args": {}', '"args": null')
def test_api_missing_non_optional_param_args_bad(self):
params = params.replace('"args": {}', '"args": 1')
def test_api_args_is_null(self):
id_, params = _build_data(self.apikey, 'get_users',)
self.assertEqual(response.status, '200 OK')
def test_api_args_is_bad(self):
def test_api_get_users(self):
ret_all = []
for usr in UserModel().get_all():
ret = usr.get_api_data()
ret_all.append(jsonify(ret))
expected = ret_all
self._compare_ok(id_, expected, given=response.body)
def test_api_get_user(self):
id_, params = _build_data(self.apikey, 'get_user',
Status change: