Commit 7e56ed68 authored by Matthieu Lamalle's avatar Matthieu Lamalle

Migration to Gitlab

parent 314ad033
......@@ -5,10 +5,10 @@ from json import loads, dumps
from aiohttp.web import HTTPForbidden, StreamResponse, HTTPInternalServerError, Response
from autobahn.wamp.exception import ApplicationError
from zephir.controller import ZephirCommonController, run
from zephir.http import register
from zephir.message import list_messages, get_message, is_message_defined
from apibridge import get_messages
from zephir.i18n import _
class ApiBridgeCtrl(ZephirCommonController):
......@@ -45,6 +45,10 @@ class ApiBridgeCtrl(ZephirCommonController):
async def forbidden(self, request, uri):
raise HTTPForbidden(reason=_('{} is a private message').format(uri))
@register('/v1', http_type='get')
async def http_api(self, request, _session_user):
return dumps(get_messages("v1"))
if __name__ == "__main__":
run(ApiBridgeCtrl)
#!/bin/bash
set -eo pipefail
if [ -z $ZEPHIR_TEST_TYPE ] && [ ! -z $1 ]; then
ZEPHIR_TEST_TYPE=$1
fi
case $ZEPHIR_TEST_TYPE in
"unit")
cd /srv/test/unit
PYTHONPATH='.' /usr/bin/python3-coverage run /usr/bin/py.test-3 -s -x
if [ ! -z "$ZEPHIR_TEST_COVERAGE_PATTERN" ]; then
/usr/bin/python3-coverage html -d /srv/report --include=$ZEPHIR_TEST_COVERAGE_PATTERN
/usr/bin/python3-coverage report --include=$ZEPHIR_TEST_COVERAGE_PATTERN
fi
;;
"integration")
py.test /srv/test/unit/10_apidescription/test_apidescription.py -x
exit $?
;;
"e2e")
echo "No e2e tests yet."
;;
*)
1>&2 echo "Unknown test type: '$ZEPHIR_TEST_TYPE'"
exit 1
esac
from .zephirmessage import get_messages
__all__ = ['get_messages']
from zephir.message import list_messages, parse_definition, get_message
from collections import OrderedDict
from os.path import join, basename, dirname
from glob import glob
from tiramisu import StrOption, IntOption, BoolOption, ChoiceOption, OptionDescription, Config
from tiramisu_web import TiramisuWeb
ALLOW_PRIVATE = False
ROOTPATH = join('..')
def _get_description(description,
name):
# hack because some description are, in fact some help
if not '\n' in description and len(description) <= 150:
doc = description
else:
doc = name
return doc
def _get_option(name,
arg,
file_path):
"""generate option
"""
props = []
if not hasattr(arg, 'default'):
props.append('mandatory')
description = arg.description.strip().rstrip()
kwargs = {'name': name,
'doc': _get_description(description, name),
'properties': frozenset(props)}
if hasattr(arg, 'default'):
kwargs['default'] = arg.default
type_ = arg.type
if type_ == 'Dict' or 'String' in type_:
return StrOption(**kwargs)
elif 'Number' in type_ or type_ == 'ID' or type_ == 'Integer':
return IntOption(**kwargs)
elif type_ == 'Boolean':
return BoolOption(**kwargs)
raise Exception('unsupported type {} in {}'.format(type_, file_path))
def _parse_args(message_def,
options,
file_path,
needs):
"""build option with args/kwargs
"""
new_options = OrderedDict()
for name, arg in message_def.parameters.items():
new_options[name] = arg
if arg.ref:
needs.setdefault(message_def.uri, {}).setdefault(arg.ref, []).append(name)
for name, arg in new_options.items():
options.append(_get_option(name, arg, file_path))
def _parse_responses(message_def,
file_path):
"""build option with returns
"""
responses = OrderedDict()
if message_def.response:
keys = {'': {'description': '',
'columns': {}}}
provides = {}
to_list = True
param_type = message_def.response.type
if param_type.startswith('[]'):
to_list = False
param_type = param_type[2:]
if param_type in ['Dict', 'File']:
pass
if message_def.response.parameters is not None:
for name, obj in message_def.response.parameters.items():
if name in responses:
raise Exception('multi response with name {} in {}'.format(name, file_path))
descr = obj.description.strip().rstrip()
keys['']['columns'][name] = {'description': obj.description,
'type': obj.type}
ref = obj.ref
if ref:
provides[name] = ref
else:
keys['']['columns'][name] = {'description': descr,
'type': obj.type}
ref = obj.ref
if ref:
provides[name] = ref
responses['keys'] = keys
responses['to_list'] = to_list
responses['to_dict'] = False
responses['provides'] = provides
else:
name = 'response'
keys['']['columns'][name] = {'description': message_def.response.description,
'type': message_def.response.type}
ref = message_def.response.ref
if ref:
provides[name] = ref
responses['keys'] = keys
responses['to_list'] = to_list
responses['to_dict'] = True
responses['provides'] = provides
return responses
def _getoptions_from_yml(message_def,
version,
optiondescriptions,
file_path,
needs):
if message_def.pattern == 'event' and message_def.response:
raise Exception('event with response?: {}'.format(file_path))
if message_def.pattern == 'rpc' and not message_def.response:
print('rpc without response?: {}'.format(file_path))
options = [StrOption('version',
'version',
version,
properties=frozenset(['hidden']))]
_parse_args(message_def, options, file_path, needs)
name = message_def.uri
description = message_def.description.strip().rstrip()
optiondescriptions[name] = (description, options)
def _get_root_option(optiondescriptions):
"""get root option
"""
domains = OrderedDict()
for name in optiondescriptions.keys():
domains.setdefault(name.split('.')[0], []).append(name)
domains_list = list(domains.keys())
domains_list.sort()
domain_option = ChoiceOption('domain',
'Sélectionner le domaine',
tuple(domains_list),
properties=frozenset(['mandatory']))
options_obj = [domain_option]
messages = OrderedDict()
for domain in domains_list:
requires = [{'option': domain_option,
'expected': domain,
'action': 'disabled',
'inverse': True}]
option = ChoiceOption('message_' + domain,
'Sélectionner le message du domaine {}'.format(domain),
tuple(domains[domain]),
requires=requires,
properties=frozenset(['mandatory']))
options_obj.append(option)
messages[domain] = option
for name, options_descr in optiondescriptions.items():
description, options = options_descr
message = messages[name.split('.')[0]]
if len(options) != 1:
requires = [{'option': message,
'expected': name,
'action': 'disabled',
'inverse': True}]
properties = None
else:
requires = None
properties = frozenset(['disabled'])
options_obj.append(OptionDescription(name.replace('.', '_'),
description,
options,
properties=properties,
requires=requires))
return OptionDescription('root', 'root', options_obj)
def get_messages(version="v1", list_msg=None):
"""generate description from yml files
version : version of the API to return
list_msg : alternate list of MessageDefinitions for testing purpose
"""
optiondescriptions = OrderedDict()
responses = OrderedDict()
needs = OrderedDict()
messages = []
if list_msg is None:
list_msg = list(list_messages())
list_msg.sort()
for message_name in list_msg:
if message_name.startswith('{}.'.format(version)):
messages.append((message_name, get_message(message_name)))
else:
for msg in list_msg:
messages.append(('{}.{}'.format(version, msg.uri), msg))
for message_name, message_def in messages:
if message_name.startswith('{}.'.format(version)):
if message_def.pattern not in ['rpc', 'event'] or \
(not message_def.public and not ALLOW_PRIVATE):
continue
if message_def.uri in responses:
raise Exception('uri {} allready loader'.format(message_def.uri))
_getoptions_from_yml(message_def,
version,
optiondescriptions,
message_name,
needs)
responses[message_def.uri] = _parse_responses(message_def,
message_name)
root = _get_root_option(optiondescriptions)
try:
api = Config(root)
except Exception as err:
raise Exception('error when generating root optiondescription: {}'.format(err))
tiramisu = TiramisuWeb(api, remotable='none')
form = [
{'title': 'Exécuter',
'cmd': 'ajax',
'type': 'submit'}]
values = {'options': tiramisu.get_jsonform(form),
'responses': responses,
'needs': needs}
return values
from collections import OrderedDict
from apibridge import get_messages
from zephir.message import MessageDefinition
def test_empty_api():
"""Check if api is empty"""
result = {'needs': OrderedDict(),
'options': {'form': [{'clearable': True, 'key': 'domain', 'type': 'choice'},
{'cmd': 'ajax', 'title': 'Exécuter', 'type': 'submit'}],
'model': [{'key': 'domain', 'required': True}],
'schema': OrderedDict([('domain',
{'enum': (),
'name': 'domain',
'title': 'Sélectionner le domaine',
'type': 'choice'})])},
'responses': OrderedDict()}
assert get_messages("v1", []) == result
def test_full_api():
"""Check if api is not empty"""
yaml = [MessageDefinition({'uri': 'api.testMessage', 'description': 'Message de test pour _get_api.',
'sampleuse': 'zephir-client config.configuration.server.get \'{"id": 15}\'',
'pattern': 'rpc', 'public': True, 'domain': 'config-domain',
'parameters': {'server_id': {'type': 'Number', 'ref': 'Server.ServerId',
'description': 'Id de la configuration.'}},
'response': {'type': 'String', 'description': 'Contenu de la configuration.'}}),
MessageDefinition({'uri': 'api.testMessagePrivate', 'description': 'Message de test privé pour _get_api.',
'sampleuse': 'zephir-client config.configuration.server.get \'{"id": 15}\'',
'pattern': 'rpc', 'public': False, 'domain': 'config-domain',
'parameters': {'server_id': {'type': 'Number', 'ref': 'Server.ServerId',
'description': 'Id de la configuration.'}},
'response': {'type': 'String', 'description': 'Contenu de la configuration.'}}),
MessageDefinition({'uri': 'api.testMessageEvent', 'description': 'Message de test event pour _get_api.',
'sampleuse': 'zephir-client config.configuration.server.get \'{"id": 15}\'',
'pattern': 'event', 'public': True, 'domain': 'config-domain',
'parameters': {'server_id': {'type': 'Number', 'ref': 'Server.ServerId',
'description': 'Id de la configuration.'}}})]
result = {'needs': OrderedDict([('api.testMessage', {'Server.ServerId': ['server_id']}),
('api.testMessageEvent',
{'Server.ServerId': ['server_id']})]),
'options': {'form': [{'clearable': True,
'dependencies': {'default': {'hide': ['message_api']},
'expected': {'api': {'show': ['message_api']}}},
'key': 'domain',
'type': 'choice'},
{'clearable': True,
'dependencies': {'default': {'hide': ['api_testMessage',
'api_testMessageEvent']},
'expected': {'api.testMessage': {'hide': ['api_testMessageEvent'],
'show': ['api_testMessage']},
'api.testMessageEvent': {'hide': ['api_testMessage'],
'show': ['api_testMessageEvent']}}},
'key': 'message_api',
'type': 'choice'},
{'clearable': True,
'key': 'api_testMessage.version',
'type': 'input'},
{'allowedpattern': '[0-9]',
'clearable': True,
'key': 'api_testMessage.server_id',
'type': 'input'},
{'clearable': True,
'key': 'api_testMessageEvent.version',
'type': 'input'},
{'allowedpattern': '[0-9]',
'clearable': True,
'key': 'api_testMessageEvent.server_id',
'type': 'input'},
{'cmd': 'ajax', 'title': 'Exécuter', 'type': 'submit'}],
'model': [{'key': 'domain', 'required': True},
{'hidden': True, 'key': 'message_api', 'required': True},
{'hidden': True,
'key': 'api_testMessage',
'properties': ['disabled']},
{'display': False,
'hidden': True,
'key': 'api_testMessage.version',
'owner': 'default',
'value': 'v1'},
{'key': 'api_testMessage.server_id', 'required': True},
{'hidden': True,
'key': 'api_testMessageEvent',
'properties': ['disabled']},
{'display': False,
'hidden': True,
'key': 'api_testMessageEvent.version',
'owner': 'default',
'value': 'v1'},
{'key': 'api_testMessageEvent.server_id',
'required': True}],
'schema': OrderedDict([('domain',
{'enum': ('api',),
'name': 'domain',
'title': 'Sélectionner le domaine',
'type': 'choice'}),
('message_api',
{'enum': ('api.testMessage',
'api.testMessageEvent'),
'name': 'message_api',
'title': 'Sélectionner le message du '
'domaine api',
'type': 'choice'}),
('api_testMessage',
{'name': 'api_testMessage',
'properties': OrderedDict([('api_testMessage.version',
{'name': 'api_testMessage.version',
'title': 'version',
'type': 'string',
'value': 'v1'}),
('api_testMessage.server_id',
{'name': 'api_testMessage.server_id',
'title': 'Id '
'de '
'la '
'configuration.',
'type': 'number'})]),
'title': 'Message de test pour _get_api.',
'type': 'object'}),
('api_testMessageEvent',
{'name': 'api_testMessageEvent',
'properties': OrderedDict([('api_testMessageEvent.version',
{'name': 'api_testMessageEvent.version',
'title': 'version',
'type': 'string',
'value': 'v1'}),
('api_testMessageEvent.server_id',
{'name': 'api_testMessageEvent.server_id',
'title': 'Id '
'de '
'la '
'configuration.',
'type': 'number'})]),
'title': 'Message de test event pour '
'_get_api.',
'type': 'object'})])},
'responses': OrderedDict([('api.testMessage',
OrderedDict([('keys',
{'': {'columns': {'response': {'description': 'Contenu '
'de '
'la '
'configuration.',
'type': 'String'}},
'description': ''}}),
('to_list', True),
('to_dict', True),
('provides', {})])),
('api.testMessageEvent', OrderedDict())])}
assert get_messages("v1", yaml) == result
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment