Merge in wsme-extdirect
This commit is contained in:
Christophe de Vienne
parent
2f27d2b7fd
commit
f807817e5f
@ -34,6 +34,7 @@ wsme.protocols =
|
||||
restjson = wsme.rest.protocol:RestProtocol
|
||||
restxml = wsme.rest.protocol:RestProtocol
|
||||
soap = wsmeext.soap:SoapProtocol
|
||||
extdirect = wsmeext.extdirect:ExtDirectProtocol
|
||||
|
||||
[files]
|
||||
packages =
|
||||
|
||||
1
wsmeext/extdirect/__init__.py
Normal file
1
wsmeext/extdirect/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from wsmeext.extdirect.protocol import ExtDirectProtocol
|
||||
120
wsmeext/extdirect/datastore.py
Normal file
120
wsmeext/extdirect/datastore.py
Normal file
@ -0,0 +1,120 @@
|
||||
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,), {})
|
||||
444
wsmeext/extdirect/protocol.py
Normal file
444
wsmeext/extdirect/protocol.py
Normal file
@ -0,0 +1,444 @@
|
||||
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 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):
|
||||
return self.read_std_arguments(context)
|
||||
elif isinstance(context, FormExtCallContext):
|
||||
return self.read_form_arguments(context)
|
||||
|
||||
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)
|
||||
20
wsmeext/extdirect/sadatastore.py
Normal file
20
wsmeext/extdirect/sadatastore.py
Normal file
@ -0,0 +1,20 @@
|
||||
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
|
||||
)
|
||||
241
wsmeext/tests/test_extdirect.py
Normal file
241
wsmeext/tests/test_extdirect.py
Normal file
@ -0,0 +1,241 @@
|
||||
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:
|
||||
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
|
||||
}
|
||||
}]
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user