diff --git a/README.rst b/README.rst index abd630b..e9b2f8b 100644 --- a/README.rst +++ b/README.rst @@ -58,7 +58,7 @@ Main features - Very simple API. - Supports user-defined simple and complex types. -- Multi-protocol : REST+Json, REST+XML, SOAP, ExtDirect and more to come. +- Multi-protocol : REST+Json, REST+XML, SOAP and more to come. - Extensible : easy to add more protocols or more base types. - Framework independence : adapters are provided to easily integrate your API in any web framework, for example a wsgi container, diff --git a/doc/changes.rst b/doc/changes.rst index fd8cb90..432e4f7 100644 --- a/doc/changes.rst +++ b/doc/changes.rst @@ -6,6 +6,7 @@ Changes * Remove support for turbogears * Remove support for cornice +* Remove support for ExtDirect * Remove SQLAlchemy support. It has never actually worked to begin with. 0.9.2 (2017-02-14) diff --git a/doc/conf.py b/doc/conf.py index c4b5696..7540f34 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -243,8 +243,8 @@ man_pages = [ autodoc_member_order = 'bysource' wsme_protocols = [ - 'restjson', 'restxml', 'soap', 'extdirect' -] + 'restjson', 'restxml', 'soap', + intersphinx_mapping = { 'python': ('http://docs.python.org/', None), diff --git a/doc/document.rst b/doc/document.rst index 39a54f2..115ee71 100644 --- a/doc/document.rst +++ b/doc/document.rst @@ -17,7 +17,7 @@ Here we consider that you already quick-started a sphinx project. extensions = ['ext'] - wsme_protocols = ['restjson', 'restxml', 'extdirect'] + wsme_protocols = ['restjson', 'restxml'] #. Copy :download:`toggle.js <_static/toggle.js>` and :download:`toggle.css <_static/toggle.css>` @@ -34,8 +34,7 @@ Config values .. confval:: wsme_protocols A list of strings that are WSME protocol names. If provided by an - additional package (for example WSME-Soap or WSME-ExtDirect), that package must - be installed. + additional package (for example WSME-Soap), that package must be installed. The types and services generated documentation will include code samples for each of these protocols. diff --git a/doc/protocols.rst b/doc/protocols.rst index a56a49d..0f3318e 100644 --- a/doc/protocols.rst +++ b/doc/protocols.rst @@ -253,114 +253,3 @@ Options ~~~~~~~ :tns: Type namespace - -ExtDirect ---------- - -:name: ``extdirect`` - -Implements the `Ext Direct`_ protocol. - -The provider definition is made available at the ``/extdirect/api.js`` subpath. - -The router url is ``/extdirect/router[/subnamespace]``. - -Options -~~~~~~~ - -:namespace: Base namespace of the api. Used for the provider definition. -:params_notation: Default notation for function call parameters. Can be - overridden for individual functions by adding the - ``extdirect_params_notation`` extra option to @expose. - - The possible notations are : - - - ``'named'`` -- The function will take only one object parameter - in which each property will be one of the parameters. - - ``'positional'`` -- The function will take as many parameters as - the function has, and their position will determine which parameter - they are. - -expose extra options -~~~~~~~~~~~~~~~~~~~~ - -:extdirect_params_notation: Override the params_notation for a particular - function. - -.. _Ext Direct: http://www.sencha.com/products/extjs/extdirect - -.. _protocols-the-example: - -The example ------------ - -In this document the same webservice example will be used to -illustrate the different protocols: - -.. code-block:: python - - class Person(object): - id = int - lastname = unicode - firstname = unicode - age = int - - hobbies = [unicode] - - def __init__(self, id=None, lastname=None, firstname=None, age=None, - hobbies=None): - if id: - self.id = id - if lastname: - self.lastname = lastname - if firstname: - self.firstname = firstname - if age: - self.age = age - if hobbies: - self.hobbies = hobbies - - persons = { - 1: Person(1, "Geller", "Ross", 30, ["Dinosaurs", "Rachel"]), - 2: Person(2, "Geller", "Monica", 28, ["Food", "Cleaning"]) - } - - class PersonController(object): - @expose(Person) - @validate(int) - def get(self, id): - return persons[id] - - @expose([Person]) - def list(self): - return persons.values() - - @expose(Person) - @validate(Person) - def update(self, p): - if p.id is Unset: - raise ClientSideError("id is missing") - persons[p.id] = p - return p - - @expose(Person) - @validate(Person) - def create(self, p): - if p.id is not Unset: - raise ClientSideError("I don't want an id") - p.id = max(persons.keys()) + 1 - persons[p.id] = p - return p - - @expose() - @validate(int) - def destroy(self, id): - if id not in persons: - raise ClientSideError("Unknown ID") - - - class WS(WSRoot): - person = PersonController() - - root = WS(webpath='ws') - diff --git a/setup.cfg b/setup.cfg index e179ecb..5449148 100644 --- a/setup.cfg +++ b/setup.cfg @@ -27,7 +27,6 @@ wsme.protocols = restjson = wsme.rest.protocol:RestProtocol restxml = wsme.rest.protocol:RestProtocol soap = wsmeext.soap:SoapProtocol - extdirect = wsmeext.extdirect:ExtDirectProtocol [files] packages = diff --git a/wsmeext/extdirect/__init__.py b/wsmeext/extdirect/__init__.py deleted file mode 100644 index ff14bd5..0000000 --- a/wsmeext/extdirect/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from wsmeext.extdirect.protocol import ExtDirectProtocol # noqa diff --git a/wsmeext/extdirect/datastore.py b/wsmeext/extdirect/datastore.py deleted file mode 100644 index 287e3a7..0000000 --- a/wsmeext/extdirect/datastore.py +++ /dev/null @@ -1,121 +0,0 @@ -import wsme -import wsme.types - -try: - import simplejson as json -except ImportError: - import json - - -class ReadResultBase(wsme.types.Base): - total = int - success = bool - message = wsme.types.text - - -def make_readresult(datatype): - ReadResult = type( - datatype.__name__ + 'ReadResult', - (ReadResultBase,), { - 'data': [datatype] - } - ) - return ReadResult - - -class DataStoreControllerMeta(type): - def __init__(cls, name, bases, dct): - if cls.__datatype__ is None: - return - if getattr(cls, '__readresulttype__', None) is None: - cls.__readresulttype__ = make_readresult(cls.__datatype__) - - cls.create = wsme.expose( - cls.__readresulttype__, - extdirect_params_notation='positional')(cls.create) - cls.create = wsme.validate(cls.__datatype__)(cls.create) - - cls.read = wsme.expose( - cls.__readresulttype__, - extdirect_params_notation='named')(cls.read) - cls.read = wsme.validate(str, str, int, int, int)(cls.read) - - cls.update = wsme.expose( - cls.__readresulttype__, - extdirect_params_notation='positional')(cls.update) - cls.update = wsme.validate(cls.__datatype__)(cls.update) - - cls.destroy = wsme.expose( - cls.__readresulttype__, - extdirect_params_notation='positional')(cls.destroy) - cls.destroy = wsme.validate(cls.__idtype__)(cls.destroy) - - -class DataStoreControllerMixin(object): - __datatype__ = None - __idtype__ = int - - __readresulttype__ = None - - def create(self, obj): - pass - - def read(self, query=None, sort=None, page=None, start=None, limit=None): - pass - - def update(self, obj): - pass - - def destroy(self, obj_id): - pass - - def model(self): - tpl = """ -Ext.define('%(appns)s.model.%(classname)s', { - extend: 'Ext.data.Model', - fields: %(fields)s, - - proxy: { - type: 'direct', - api: { - create: %(appns)s.%(controllerns)s.create, - read: %(appns)s.%(controllerns)s.read, - update: %(appns)s.%(controllerns)s.update, - destroy: %(appns)s.%(controllerns)s.destroy - }, - reader: { - root: 'data' - } - } -}); - """ - fields = [ - attr.name for attr in self.__datatype__._wsme_attributes - ] - d = { - 'appns': 'Demo', - 'controllerns': 'stores.' + self.__datatype__.__name__.lower(), - 'classname': self.__datatype__.__name__, - 'fields': json.dumps(fields) - } - return tpl % d - - def store(self): - tpl = """ -Ext.define('%(appns)s.store.%(classname)s', { - extend: 'Ext.data.Store', - model: '%(appns)s.model.%(classname)s' -}); -""" - d = { - 'appns': 'Demo', - 'classname': self.__datatype__.__name__, - } - - return tpl % d - - -DataStoreController = DataStoreControllerMeta( - 'DataStoreController', - (DataStoreControllerMixin,), {} -) diff --git a/wsmeext/extdirect/protocol.py b/wsmeext/extdirect/protocol.py deleted file mode 100644 index 23793b4..0000000 --- a/wsmeext/extdirect/protocol.py +++ /dev/null @@ -1,450 +0,0 @@ -import datetime -import decimal - -from simplegeneric import generic - -from wsme.exc import ClientSideError -from wsme.protocol import CallContext, Protocol, expose -from wsme.utils import parse_isodate, parse_isodatetime, parse_isotime -from wsme.rest.args import from_params -from wsme.types import iscomplex, isusertype, list_attributes, Unset -import wsme.types - -try: - import simplejson as json -except ImportError: - import json # noqa - -from six import u - - -class APIDefinitionGenerator(object): - tpl = """\ -Ext.ns("%(rootns)s"); - -if (!%(rootns)s.wsroot) { - %(rootns)s.wsroot = "%(webpath)s. -} - -%(descriptors)s - -Ext.syncRequire(['Ext.direct.*'], function() { - %(providers)s -}); -""" - descriptor_tpl = """\ -Ext.ns("%(fullns)s"); - -%(fullns)s.Descriptor = { - "url": %(rootns)s.wsroot + "extdirect/router/%(ns)s", - "namespace": "%(fullns)s", - "type": "remoting", - "actions": %(actions)s - "enableBuffer": true -}; -""" - provider_tpl = """\ - Ext.direct.Manager.addProvider(%(fullns)s.Descriptor); -""" - - def __init__(self): - pass - - def render(self, rootns, webpath, namespaces, fullns): - descriptors = u('') - for ns in sorted(namespaces): - descriptors += self.descriptor_tpl % { - 'ns': ns, - 'rootns': rootns, - 'fullns': fullns(ns), - 'actions': '\n'.join(( - ' ' * 4 + line - for line - in json.dumps(namespaces[ns], indent=4).split('\n') - )) - } - - providers = u('') - for ns in sorted(namespaces): - providers += self.provider_tpl % { - 'fullns': fullns(ns) - } - - r = self.tpl % { - 'rootns': rootns, - 'webpath': webpath, - 'descriptors': descriptors, - 'providers': providers, - } - return r - - -@generic -def fromjson(datatype, value): - if value is None: - return None - if iscomplex(datatype): - newvalue = datatype() - for attrdef in list_attributes(datatype): - if attrdef.name in value: - setattr(newvalue, attrdef.key, - fromjson(attrdef.datatype, value[attrdef.name])) - value = newvalue - elif isusertype(datatype): - value = datatype.frombasetype(fromjson(datatype.basetype, value)) - return value - - -@generic -def tojson(datatype, value): - if value is None: - return value - if iscomplex(datatype): - d = {} - for attrdef in list_attributes(datatype): - attrvalue = getattr(value, attrdef.key) - if attrvalue is not Unset: - d[attrdef.name] = tojson(attrdef.datatype, attrvalue) - value = d - elif isusertype(datatype): - value = tojson(datatype.basetype, datatype.tobasetype(value)) - return value - - -@fromjson.when_type(wsme.types.ArrayType) -def array_fromjson(datatype, value): - return [fromjson(datatype.item_type, item) for item in value] - - -@tojson.when_type(wsme.types.ArrayType) -def array_tojson(datatype, value): - if value is None: - return value - return [tojson(datatype.item_type, item) for item in value] - - -@fromjson.when_type(wsme.types.DictType) -def dict_fromjson(datatype, value): - if value is None: - return value - return dict(( - (fromjson(datatype.key_type, key), - fromjson(datatype.value_type, value)) - for key, value in value.items() - )) - - -@tojson.when_type(wsme.types.DictType) -def dict_tojson(datatype, value): - if value is None: - return value - return dict(( - (tojson(datatype.key_type, key), - tojson(datatype.value_type, value)) - for key, value in value.items() - )) - - -@tojson.when_object(wsme.types.bytes) -def bytes_tojson(datatype, value): - if value is None: - return value - return value.decode('ascii') - - -# raw strings -@fromjson.when_object(wsme.types.bytes) -def bytes_fromjson(datatype, value): - if value is not None: - value = value.encode('ascii') - return value - - -# unicode strings - -@fromjson.when_object(wsme.types.text) -def text_fromjson(datatype, value): - if isinstance(value, wsme.types.bytes): - return value.decode('utf-8') - return value - - -# datetime.time - -@fromjson.when_object(datetime.time) -def time_fromjson(datatype, value): - if value is None or value == '': - return None - return parse_isotime(value) - - -@tojson.when_object(datetime.time) -def time_tojson(datatype, value): - if value is None: - return value - return value.isoformat() - - -# datetime.date - -@fromjson.when_object(datetime.date) -def date_fromjson(datatype, value): - if value is None or value == '': - return None - return parse_isodate(value) - - -@tojson.when_object(datetime.date) -def date_tojson(datatype, value): - if value is None: - return value - return value.isoformat() - - -# datetime.datetime - -@fromjson.when_object(datetime.datetime) -def datetime_fromjson(datatype, value): - if value is None or value == '': - return None - return parse_isodatetime(value) - - -@tojson.when_object(datetime.datetime) -def datetime_tojson(datatype, value): - if value is None: - return value - return value.isoformat() - - -# decimal.Decimal - -@fromjson.when_object(decimal.Decimal) -def decimal_fromjson(datatype, value): - if value is None: - return value - return decimal.Decimal(value) - - -@tojson.when_object(decimal.Decimal) -def decimal_tojson(datatype, value): - if value is None: - return value - return str(value) - - -class ExtCallContext(CallContext): - def __init__(self, request, namespace, calldata): - super(ExtCallContext, self).__init__(request) - self.namespace = namespace - - self.tid = calldata['tid'] - self.action = calldata['action'] - self.method = calldata['method'] - self.params = calldata['data'] - - -class FormExtCallContext(CallContext): - def __init__(self, request, namespace): - super(FormExtCallContext, self).__init__(request) - self.namespace = namespace - - self.tid = request.params['extTID'] - self.action = request.params['extAction'] - self.method = request.params['extMethod'] - self.params = [] - - -class ExtDirectProtocol(Protocol): - """ - ExtDirect protocol. - - For more detail on the protocol, see - http://www.sencha.com/products/extjs/extdirect. - - .. autoattribute:: name - .. autoattribute:: content_types - """ - name = 'extdirect' - displayname = 'ExtDirect' - content_types = ['application/json', 'text/javascript'] - - def __init__(self, namespace='', params_notation='named', nsfolder=None): - self.namespace = namespace - self.appns, self.apins = namespace.rsplit('.', 2) \ - if '.' in namespace else (namespace, '') - self.default_params_notation = params_notation - self.appnsfolder = nsfolder - - @property - def api_alias(self): - if self.appnsfolder: - alias = '/%s/%s.js' % ( - self.appnsfolder, - self.apins.replace('.', '/')) - return alias - - def accept(self, req): - path = req.path - assert path.startswith(self.root._webpath) - path = path[len(self.root._webpath):] - - return ( - path == self.api_alias or - path == "/extdirect/api" or - path.startswith("/extdirect/router") - ) - - def iter_calls(self, req): - path = req.path - - assert path.startswith(self.root._webpath) - path = path[len(self.root._webpath):].strip() - - assert path.startswith('/extdirect/router'), path - path = path[17:].strip('/') - - if path: - namespace = path.split('.') - else: - namespace = [] - - if 'extType' in req.params: - req.wsme_extdirect_batchcall = False - yield FormExtCallContext(req, namespace) - else: - data = json.loads(req.body.decode('utf8')) - req.wsme_extdirect_batchcall = isinstance(data, list) - if not req.wsme_extdirect_batchcall: - data = [data] - req.callcount = len(data) - - for call in data: - yield ExtCallContext(req, namespace, call) - - def extract_path(self, context): - path = list(context.namespace) - - if context.action: - path.append(context.action) - - path.append(context.method) - - return path - - def read_std_arguments(self, context): - funcdef = context.funcdef - notation = funcdef.extra_options.get('extdirect_params_notation', - self.default_params_notation) - args = context.params - if notation == 'positional': - kw = dict( - (argdef.name, fromjson(argdef.datatype, arg)) - for argdef, arg in zip(funcdef.arguments, args) - ) - elif notation == 'named': - if len(args) == 0: - args = [{}] - elif len(args) > 1: - raise ClientSideError( - "Named arguments: takes a single object argument") - args = args[0] - kw = dict( - (argdef.name, fromjson(argdef.datatype, args[argdef.name])) - for argdef in funcdef.arguments if argdef.name in args - ) - else: - raise ValueError("Invalid notation: %s" % notation) - return kw - - def read_form_arguments(self, context): - kw = {} - for argdef in context.funcdef.arguments: - value = from_params(argdef.datatype, context.request.params, - argdef.name, set()) - if value is not Unset: - kw[argdef.name] = value - return kw - - def read_arguments(self, context): - if isinstance(context, ExtCallContext): - kwargs = self.read_std_arguments(context) - elif isinstance(context, FormExtCallContext): - kwargs = self.read_form_arguments(context) - wsme.runtime.check_arguments(context.funcdef, (), kwargs) - return kwargs - - def encode_result(self, context, result): - return json.dumps({ - 'type': 'rpc', - 'tid': context.tid, - 'action': context.action, - 'method': context.method, - 'result': tojson(context.funcdef.return_type, result) - }) - - def encode_error(self, context, infos): - return json.dumps({ - 'type': 'exception', - 'tid': context.tid, - 'action': context.action, - 'method': context.method, - 'message': '%(faultcode)s: %(faultstring)s' % infos, - 'where': infos['debuginfo']}) - - def prepare_response_body(self, request, results): - r = ",\n".join(results) - if request.wsme_extdirect_batchcall: - return "[\n%s\n]" % r - else: - return r - - def get_response_status(self, request): - return 200 - - def get_response_contenttype(self, request): - return "text/javascript" - - def fullns(self, ns): - return ns and '%s.%s' % (self.namespace, ns) or self.namespace - - @expose('/extdirect/api', "text/javascript") - @expose('${api_alias}', "text/javascript") - def api(self): - namespaces = {} - for path, funcdef in self.root.getapi(): - if len(path) > 1: - namespace = '.'.join(path[:-2]) - action = path[-2] - else: - namespace = '' - action = '' - if namespace not in namespaces: - namespaces[namespace] = {} - if action not in namespaces[namespace]: - namespaces[namespace][action] = [] - notation = funcdef.extra_options.get('extdirect_params_notation', - self.default_params_notation) - method = { - 'name': funcdef.name} - - if funcdef.extra_options.get('extdirect_formhandler', False): - method['formHandler'] = True - method['len'] = 1 if notation == 'named' \ - else len(funcdef.arguments) - namespaces[namespace][action].append(method) - webpath = self.root._webpath - if webpath and not webpath.endswith('/'): - webpath += '/' - return APIDefinitionGenerator().render( - namespaces=namespaces, - webpath=webpath, - rootns=self.namespace, - fullns=self.fullns, - ) - - def encode_sample_value(self, datatype, value, format=False): - r = tojson(datatype, value) - content = json.dumps(r, ensure_ascii=False, indent=4 if format else 0, - sort_keys=format) - return ('javascript', content) diff --git a/wsmeext/extdirect/sadatastore.py b/wsmeext/extdirect/sadatastore.py deleted file mode 100644 index 44d79cb..0000000 --- a/wsmeext/extdirect/sadatastore.py +++ /dev/null @@ -1,19 +0,0 @@ -from wsmeext.extdirect import datastore - - -class SADataStoreController(datastore.DataStoreController): - __dbsession__ = None - __datatype__ = None - - def read(self, query=None, sort=None, page=None, start=None, limit=None): - q = self.__dbsession__.query(self.__datatype__.__saclass__) - total = q.count() - if start is not None and limit is not None: - q = q.slice(start, limit) - return self.__readresulttype__( - data=[ - self.__datatype__(o) for o in q - ], - success=True, - total=total - ) diff --git a/wsmeext/tests/test_extdirect.py b/wsmeext/tests/test_extdirect.py deleted file mode 100644 index 4c5bea8..0000000 --- a/wsmeext/tests/test_extdirect.py +++ /dev/null @@ -1,243 +0,0 @@ -import base64 -import datetime -import decimal - -try: - import simplejson as json -except ImportError: - import json # noqa - -import wsme.tests.protocol -from wsme.utils import parse_isodatetime, parse_isodate, parse_isotime -from wsme.types import isarray, isdict, isusertype - -import six - -if six.PY3: - from urllib.parse import urlencode -else: - from urllib import urlencode # noqa - - -def encode_arg(value): - if isinstance(value, tuple): - value, datatype = value - else: - datatype = type(value) - - if isinstance(datatype, list): - value = [encode_arg((item, datatype[0])) for item in value] - elif isinstance(datatype, dict): - key_type, value_type = list(datatype.items())[0] - value = dict(( - (encode_arg((key, key_type)), - encode_arg((value, value_type))) - for key, value in value.items() - )) - elif datatype in (datetime.date, datetime.time, datetime.datetime): - value = value.isoformat() - elif datatype == wsme.types.binary: - value = base64.encodestring(value).decode('ascii') - elif datatype == wsme.types.bytes: - value = value.decode('ascii') - elif datatype == decimal.Decimal: - value = str(value) - return value - - -def decode_result(value, datatype): - if value is None: - return None - if datatype == wsme.types.binary: - value = base64.decodestring(value.encode('ascii')) - return value - if isusertype(datatype): - datatype = datatype.basetype - if isinstance(datatype, list): - value = [decode_result(item, datatype[0]) for item in value] - elif isarray(datatype): - value = [decode_result(item, datatype.item_type) for item in value] - elif isinstance(datatype, dict): - key_type, value_type = list(datatype.items())[0] - value = dict(( - (decode_result(key, key_type), - decode_result(value, value_type)) - for key, value in value.items() - )) - elif isdict(datatype): - key_type, value_type = datatype.key_type, datatype.value_type - value = dict(( - (decode_result(key, key_type), - decode_result(value, value_type)) - for key, value in value.items() - )) - elif datatype == datetime.time: - value = parse_isotime(value) - elif datatype == datetime.date: - value = parse_isodate(value) - elif datatype == datetime.datetime: - value = parse_isodatetime(value) - elif hasattr(datatype, '_wsme_attributes'): - for attr in datatype._wsme_attributes: - if attr.key not in value: - continue - value[attr.key] = decode_result(value[attr.key], attr.datatype) - elif datatype == decimal.Decimal: - value = decimal.Decimal(value) - elif datatype == wsme.types.bytes: - value = value.encode('ascii') - elif datatype is not None and type(value) != datatype: - value = datatype(value) - return value - - -class TestExtDirectProtocol(wsme.tests.protocol.ProtocolTestCase): - protocol = 'extdirect' - protocol_options = { - 'namespace': 'MyNS.api', - 'nsfolder': 'app' - } - - def call(self, fname, _rt=None, _no_result_decode=False, _accept=None, - **kw): - path = fname.split('/') - try: - func, funcdef, args = self.root._lookup_function(path) - arguments = funcdef.arguments - except Exception: - arguments = [] - if len(path) == 1: - ns, action, fname = '', '', path[0] - elif len(path) == 2: - ns, action, fname = '', path[0], path[1] - else: - ns, action, fname = '.'.join(path[:-2]), path[-2], path[-1] - print(kw) - - args = [ - dict( - (arg.name, encode_arg(kw[arg.name])) - for arg in arguments if arg.name in kw - ) - ] - print("args =", args) - data = json.dumps({ - 'type': 'rpc', - 'tid': 0, - 'action': action, - 'method': fname, - 'data': args, - }) - print(data) - headers = {'Content-Type': 'application/json'} - if _accept: - headers['Accept'] = _accept - res = self.app.post('/extdirect/router/%s' % ns, data, headers=headers, - expect_errors=True) - - print(res.body) - - if _no_result_decode: - return res - - data = json.loads(res.text) - if data['type'] == 'rpc': - r = data['result'] - return decode_result(r, _rt) - elif data['type'] == 'exception': - faultcode, faultstring = data['message'].split(': ', 1) - debuginfo = data.get('where') - raise wsme.tests.protocol.CallException( - faultcode, faultstring, debuginfo) - - def test_api_alias(self): - assert self.root._get_protocol('extdirect').api_alias == '/app/api.js' - - def test_get_api(self): - res = self.app.get('/app/api.js') - print(res.body) - assert res.body - - def test_positional(self): - self.root._get_protocol('extdirect').default_params_notation = \ - 'positional' - - data = json.dumps({ - 'type': 'rpc', - 'tid': 0, - 'action': 'misc', - 'method': 'multiply', - 'data': [2, 5], - }) - headers = {'Content-Type': 'application/json'} - res = self.app.post('/extdirect/router', data, headers=headers) - - print(res.body) - - data = json.loads(res.text) - assert data['type'] == 'rpc' - r = data['result'] - assert r == 10 - - def test_batchcall(self): - data = json.dumps([{ - 'type': 'rpc', - 'tid': 1, - 'action': 'argtypes', - 'method': 'setdate', - 'data': [{'value': '2011-04-06'}], - }, { - 'type': 'rpc', - 'tid': 2, - 'action': 'returntypes', - 'method': 'getbytes', - 'data': [] - }]) - print(data) - headers = {'Content-Type': 'application/json'} - res = self.app.post('/extdirect/router', data, headers=headers) - - print(res.body) - - rdata = json.loads(res.text) - - assert len(rdata) == 2 - - assert rdata[0]['tid'] == 1 - assert rdata[0]['result'] == '2011-04-06' - assert rdata[1]['tid'] == 2 - assert rdata[1]['result'] == 'astring' - - def test_form_call(self): - params = { - 'value[0].inner.aint': 54, - 'value[1].inner.aint': 55, - 'extType': 'rpc', - 'extTID': 1, - 'extAction': 'argtypes', - 'extMethod': 'setnestedarray', - } - - body = urlencode(params) - r = self.app.post( - '/extdirect/router', - body, - headers={'Content-Type': 'application/x-www-form-urlencoded'} - ) - print(r) - - assert json.loads(r.text) == { - "tid": "1", - "action": "argtypes", - "type": "rpc", - "method": "setnestedarray", - "result": [{ - "inner": { - "aint": 54 - } - }, { - "inner": { - "aint": 55 - } - }] - }