#
# Use of the software files and documentation is subject to license terms.
-'''A CI test script for odl-jsonrpc.'''
+"""A CI test script for odl-jsonrpc."""
import logging
from datetime import datetime
import json
import zmq
+
class StreamFormatter(logging.Formatter):
- '''Provide a custom timestamp for logging.'''
+ """Provide a custom timestamp for logging."""
+
def formatTime(self, record, datefmt=None):
- '''Return record time as a UTC timestamp (RFC 3339 Section 5.6).'''
- return datetime.utcfromtimestamp(record.created).isoformat('T') + 'Z'
-
-PARSE_ERROR_RESPONSE = json.dumps({
- 'jsonrpc': '2.0',
- 'error': {'code': -32700, 'message': 'Parse error'},
- 'id': None,
-})
-INVALID_REQUEST_RESPONSE = json.dumps({
- 'jsonrpc': '2.0',
- 'error': {'code': -32600, 'message': 'Invalid Request'},
- 'id': None,
-})
+ """Return record time as a UTC timestamp (RFC 3339 Section 5.6)."""
+ return datetime.utcfromtimestamp(record.created).isoformat("T") + "Z"
+
+
+PARSE_ERROR_RESPONSE = json.dumps(
+ {"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}
+)
+INVALID_REQUEST_RESPONSE = json.dumps(
+ {
+ "jsonrpc": "2.0",
+ "error": {"code": -32600, "message": "Invalid Request"},
+ "id": None,
+ }
+)
+
class Service(object):
- '''A service accepting JSON RPC Requests for the 'read' method.'''
+ """A service accepting JSON RPC Requests for the 'read' method."""
+
def __init__(self, store, entity, path, value):
self._store = store
self._entity = entity
self._path = path
self._value = value
- self._methods = {'read': self.read}
+ self._methods = {"read": self.read}
+
@staticmethod
def _parse_request(request):
- '''If `request` passes the most basic checks for a JSON RPC Request,
+ """If `request` passes the most basic checks for a JSON RPC Request,
then return its 'id', 'method' and 'params'.
- '''
+ """
try:
- version = request['jsonrpc']
- id_ = request['id']
- method = request['method']
- params = request['params']
+ version = request["jsonrpc"]
+ id_ = request["id"]
+ method = request["method"]
+ params = request["params"]
except KeyError:
raise ValueError()
- if version != '2.0':
+ if version != "2.0":
raise ValueError()
return (version, id_, method, params)
+
def read(self, store, entity, path):
- '''The implementation of the `read` method for this test. If `store`,
+ """The implementation of the `read` method for this test. If `store`,
`entity` and `path` do not match expected values, then return the
configured fixed value: otherwise raise :class:`ValueError`.
- '''
+ """
if self._store != store or self._entity != entity or self._path != path:
- raise ValueError('unexpected param values')
+ raise ValueError("unexpected param values")
return self._value
+
def execute_json(self, string):
- '''Execute an encoded JSON RPC Request from `string`. Return a 2-tuple
+ """Execute an encoded JSON RPC Request from `string`. Return a 2-tuple
of (code, response), where code is process exit code and `response`
is an encoded JSON RPC Response. A zero exit code is returned if the
method implementation of this service was invoked successfully:
non-zero otherwise.
- '''
+ """
try:
request = json.loads(string)
except ValueError:
(version, id_, method, params) = self._parse_request(request)
except ValueError:
return (2, INVALID_REQUEST_RESPONSE)
- response = {'jsonrpc': version, 'id': id_}
+ response = {"jsonrpc": version, "id": id_}
try:
- ### assumes that params are supplied as a list for call by position
- response['result'] = self._methods[method](*params) # pylint: disable=star-args
+ # assumes that params are supplied as a list for call by position
+ response["result"] = self._methods[method](
+ *params
+ ) # pylint: disable=star-args
except KeyError:
- response['error'] = {'code': -32601, 'message': 'Method not found'}
+ response["error"] = {"code": -32601, "message": "Method not found"}
code = 3
except TypeError:
- response['error'] = {'code': -32602, 'message': 'Invalid params'}
+ response["error"] = {"code": -32602, "message": "Invalid params"}
code = 4
except ValueError as exc:
- response['error'] = {'code': -32603, 'message': 'Internal error'}
- response['error']['data'] = str(exc)
+ response["error"] = {"code": -32603, "message": "Internal error"}
+ response["error"]["data"] = str(exc)
code = 5
else:
code = 0
return (code, json.dumps(response))
+
def init_logging(level):
- '''Initialise the default logger at logging `level`.'''
+ """Initialise the default logger at logging `level`."""
logger = logging.getLogger()
logger.setLevel(level)
handler = logging.StreamHandler()
- formatter = StreamFormatter(
- fmt='%(asctime)s:%(levelname)s:%(name)s:%(message)s'
- )
+ formatter = StreamFormatter(fmt="%(asctime)s:%(levelname)s:%(name)s:%(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
+
def json_or_string(value):
- '''If `value` is a JSON-encoded string, then return the decoded value:
+ """If `value` is a JSON-encoded string, then return the decoded value:
otherwise return `value` as a string.
- '''
+ """
try:
return json.loads(value)
except ValueError:
return str(value)
+
def main():
- '''Run a ZMQ REP socket on `uri` for accepting a single JSON RPC Request.
+ """Run a ZMQ REP socket on `uri` for accepting a single JSON RPC Request.
To successfully invoke the 'read' method, the Request 'params' must be
[`store`, `entity`, `path`] with the exact same values as were specified
in the command line args. On success, a JSON RPC Response with 'result'
`value` will be returned: otherwise, a JSON RPC Response with 'error'
will be returned.
- '''
+ """
init_logging(logging.INFO)
aparser = ArgumentParser(description=main.__doc__)
- aparser.add_argument('uri')
- aparser.add_argument('store', type=json_or_string)
- aparser.add_argument('entity', type=json_or_string)
- aparser.add_argument('path', type=json_or_string)
- aparser.add_argument('value', type=json_or_string)
+ aparser.add_argument("uri")
+ aparser.add_argument("store", type=json_or_string)
+ aparser.add_argument("entity", type=json_or_string)
+ aparser.add_argument("path", type=json_or_string)
+ aparser.add_argument("value", type=json_or_string)
args = aparser.parse_args()
service = Service(args.store, args.entity, args.path, args.value)
- zock = zmq.Context().socket(zmq.REP) # pylint: disable=no-member
+ zock = zmq.Context().socket(zmq.REP) # pylint: disable=no-member
zock.bind(args.uri)
- logging.info('ZMQ REP listening on %s', args.uri)
+ logging.info("ZMQ REP listening on %s", args.uri)
request = zock.recv()
- logging.info('>%s', request)
+ logging.info(">%s", request)
(code, response) = service.execute_json(request)
- logging.info('<%s', response)
+ logging.info("<%s", response)
zock.send(response)
- logging.info('exiting with code %d', code)
+ logging.info("exiting with code %d", code)
zock.close()
sys.exit(code)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
Code Review / integration / test.git / commitdiff