Commit 13e1b2b1 authored by Romain Courteaud's avatar Romain Courteaud 🐙

WIP: erp5_json_rpc_api: add JSON RPC Service portal type

This new Web Service can be configured to define a list of entry points associated to JSON Form.

Those entry points are accessed by a client by doing an HTTP Post with a JSON input body.
The input JSON is validated by the JSON Form.

An output JSON body is expected as response.
parent c852f636
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ActionInformation" module="Products.CMFCore.ActionInformation"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>action</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>categories</string> </key>
<value>
<tuple>
<string>action_type/object_view</string>
</tuple>
</value>
</item>
<item>
<key> <string>category</string> </key>
<value> <string>object_view</string> </value>
</item>
<item>
<key> <string>condition</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>icon</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>view</string> </value>
</item>
<item>
<key> <string>permissions</string> </key>
<value>
<tuple>
<string>View</string>
</tuple>
</value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Action Information</string> </value>
</item>
<item>
<key> <string>priority</string> </key>
<value> <float>1.0</float> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>View</string> </value>
</item>
<item>
<key> <string>visible</string> </key>
<value> <int>1</int> </value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="Expression" module="Products.CMFCore.Expression"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>text</string> </key>
<value> <string>string:${object_url}/JsonRpcService_view</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
##############################################################################
# coding:utf-8
# Copyright (c) 2023 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
# import base64
# import binascii
import json
import typing
# import six
from six.moves.urllib.parse import unquote
if typing.TYPE_CHECKING:
from typing import Any, Callable, Optional
from erp5.component.document.OpenAPITypeInformation import OpenAPIOperation, OpenAPIParameter
from ZPublisher.HTTPRequest import HTTPRequest
_ = (
OpenAPIOperation, OpenAPIParameter, HTTPRequest, Any, Callable, Optional)
# import jsonschema
from AccessControl import ClassSecurityInfo, ModuleSecurityInfo
from zExceptions import NotFound
from zope.publisher.interfaces import IPublishTraverse
import zope.component
import zope.interface
from Products.ERP5Type import Permissions, PropertySheet
from erp5.component.document.OpenAPITypeInformation import (
byteify,
OpenAPIError
)
from erp5.component.document.OpenAPIService import OpenAPIService
import jsonschema
class JsonRpcAPIError(OpenAPIError):
pass
class JsonRpcAPINotParsableJsonContent(JsonRpcAPIError):
type = "not-parsable-json-content"
status = 400
class JsonRpcAPINotJsonDictContent(JsonRpcAPIError):
type = "not-json-object-content"
status = 400
class JsonRpcAPIInvalidJsonDictContent(JsonRpcAPIError):
type = "invalid-json-object-content"
status = 400
class JsonRpcAPINotAllowedHttpMethod(JsonRpcAPIError):
type = "not-allowed-http-method"
status = 405
class JsonRpcAPIBadContentType(JsonRpcAPIError):
type = "unexpected-media-type"
status = 415
ModuleSecurityInfo(__name__).declarePublic(
JsonRpcAPIError.__name__,
JsonRpcAPINotParsableJsonContent.__name__,
JsonRpcAPINotJsonDictContent.__name__,
JsonRpcAPIInvalidJsonDictContent.__name__,
JsonRpcAPINotAllowedHttpMethod.__name__,
JsonRpcAPIBadContentType.__name__,
)
@zope.interface.implementer(IPublishTraverse)
class JsonRpcAPIService(OpenAPIService):
add_permission = Permissions.AddPortalContent
# Declarative security
security = ClassSecurityInfo()
security.declareObjectProtected(Permissions.AccessContentsInformation)
# Declarative properties
property_sheets = (
PropertySheet.Base,
PropertySheet.XMLObject,
PropertySheet.CategoryCore,
PropertySheet.DublinCore,
PropertySheet.Reference,
)
security.declareProtected(
Permissions.AccessContentsInformation, 'viewOpenAPIAsJson')
def viewOpenAPIAsJson(self):
"""Return the Open API as JSON, with the current endpoint added as first servers
"""
raise NotImplementedError()
def getMatchingOperation(self, request):
# type: (HTTPRequest) -> Optional[OpenAPIOperation]
# Compute the relative URL of the request path, by removing the
# relative URL of the Open API Service. This is tricky, because
# it may be in the acquisition context of a web section and the request
# might be using a virtual root with virtual paths.
# First, strip the left part of the URL corresponding to the "root"
web_section = self.getWebSectionValue()
root = web_section if web_section is not None else self.getPortalObject()
request_path_parts = [
unquote(part) for part in request['URL']
[1 + len(request.physicalPathToURL(root.getPhysicalPath())):].split('/')
]
# then strip everything corresponding to the "self" open api service.
# Here, unlike getPhysicalPath(), we don't use the inner acquistion,
# but keep the acquisition chain from this request traversal.
i = 0
for aq_parent in reversed(self.aq_chain[:self.aq_chain.index(root)]):
if aq_parent.id == request_path_parts[i]:
i += 1
else:
break
request_path_parts = request_path_parts[i:]
request_method = request.method.lower()
matched_operation = None
request.other['traverse_subpath'] = request_path_parts
if request_path_parts:
# Compare the request path with the web service configuration string
# Do not expect any string convention here (like not / or whatever).
# The convention is defined by the web service configuration only
request_path = '/'.join(request_path_parts)
matched_operation = None
for line in self.getJsonFormList():
if not line:
continue
line_split_list = line.split(' | ', 1)
if len(line_split_list) != 2:
raise ValueError('Unparsable configuration: %s' % line)
action_reference, callable_id = line_split_list
if request_path == action_reference:
matched_operation = callable_id
if matched_operation is None:
raise NotFound(request_path)
content_type = request.getHeader('Content-Type', '')
if 'application/json' not in content_type:
raise JsonRpcAPIBadContentType(
'Request Content-Type must be "application/json", not "%s"' % content_type
)
if (request_method != 'post'):
raise JsonRpcAPINotAllowedHttpMethod('Only HTTP POST accepted')
return matched_operation
def handleException(self, exception, request):
if isinstance(exception, JsonRpcAPIError):
# Prevent catching all exceptions in the script
# to prevent returning wrong content in case of bugs...
script_id = self.getErrorHandlerScriptId()
if script_id:
try:
script_result = getattr(self, script_id)(exception)
except Exception as e:
exception = e
else:
# If the script returns something, consider the exception
# has explicitely handled
if script_result:
response = request.response
response.setBody(json.dumps(script_result).encode())#, lock=True)
response.setHeader("Content-Type", "application/json")
response.setStatus(exception.status)#, lock=True)
return
# ... but if really needed, developer is still able to use the type based method
return super(JsonRpcAPIService, self).handleException(exception, request)
def executeMethod(self, request):
# type: (HTTPRequest) -> Any
operation = self.getMatchingOperation(request)
if operation is None:
raise NotFound()
json_form = getattr(self, operation)#self.getMethodForOperation(operation)
if json_form.getPortalType() != 'JSON Form':
raise ValueError('%s is not a JSON Form' % operation)
# parameters = self.extractParametersFromRequest(operation, request)
try:
json_data = byteify(json.loads(request.get('BODY')))
except BaseException as e:
raise JsonRpcAPINotParsableJsonContent(str(e))
if not isinstance(json_data, dict):
raise JsonRpcAPINotJsonDictContent("Did not received a JSON Object")
try:
result = json_form(json_data=json_data, list_error=False)#**parameters)
except jsonschema.exceptions.ValidationError as e:
raise JsonRpcAPIInvalidJsonDictContent(str(e))
response = request.RESPONSE
output_schema = json_form.getOutputSchema()
# XXX Hardcoded JSONForm behaviour
if (result == "Nothing to do") or (not result):
# If there is no output, ensure no output schema is defined
if output_schema:
raise ValueError('%s has an output schema but response is empty' % operation)
result = {
'status': 200,
'type': 'success-type',
'title': 'query completed'
}
else:
if not output_schema:
raise ValueError('%s does not have an output schema but response is not empty' % operation)
# Ensure the response matches the output schema
try:
jsonschema.validate(
result,
json.loads(output_schema),
format_checker=jsonschema.FormatChecker()
)
except jsonschema.exceptions.ValidationError as e:
raise ValueError(e.message)
response.setHeader("Content-Type", "application/json")
return json.dumps(result).encode()
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Document Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>default_reference</string> </key>
<value> <string>JsonRpcAPIService</string> </value>
</item>
<item>
<key> <string>default_source_reference</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>document.erp5.JsonRpcAPIService</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>text_content_error_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.Workflow"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_log</string> </key>
<value>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<allowed_content_type_list>
<portal_type id="Web Service Tool">
<item>JSON RPC Service</item>
</portal_type>
</allowed_content_type_list>
\ No newline at end of file
<property_sheet_list>
<portal_type id="JSON RPC Service">
<item>JsonRpcService</item>
</portal_type>
</property_sheet_list>
\ No newline at end of file
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Base Type" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>content_icon</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>JSON RPC Service</string> </value>
</item>
<item>
<key> <string>init_script</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>permission</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>searchable_text_property_id</string> </key>
<value>
<tuple>
<string>title</string>
<string>description</string>
<string>reference</string>
<string>short_title</string>
<string>json_form_list</string>
</tuple>
</value>
</item>
<item>
<key> <string>short_title</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>type_class</string> </key>
<value> <string>JsonRpcAPIService</string> </value>
</item>
<item>
<key> <string>type_interface</string> </key>
<value>
<tuple/>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<workflow_chain>
<chain>
<type>JSON RPC Service</type>
<workflow>edit_workflow, validation_workflow</workflow>
</chain>
</workflow_chain>
\ No newline at end of file
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Property Sheet" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_count</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>_mt_index</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
<item>
<key> <string>_tree</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>JsonRpcService</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Property Sheet</string> </value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="Length" module="BTrees.Length"/>
</pickle>
<pickle> <int>0</int> </pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="OOBTree" module="BTrees.OOBTree"/>
</pickle>
<pickle>
<none/>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="OOBTree" module="BTrees.OOBTree"/>
</pickle>
<pickle>
<none/>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Standard Property" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_local_properties</string> </key>
<value>
<tuple>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>mode</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
</tuple>
</value>
</item>
<item>
<key> <string>categories</string> </key>
<value>
<tuple>
<string>elementary_type/string</string>
</tuple>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>error_handler_script_id_property</string> </value>
</item>
<item>
<key> <string>mode</string> </key>
<value> <string>w</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Standard Property</string> </value>
</item>
<item>
<key> <string>range</string> </key>
<value> <int>1</int> </value>
</item>
<item>
<key> <string>storage_id</string> </key>
<value>
<none/>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Standard Property" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_local_properties</string> </key>
<value>
<tuple>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>mode</string> </value>
</item>
<item>
<key> <string>type</string> </key>
<value> <string>string</string> </value>
</item>
</dictionary>
</tuple>
</value>
</item>
<item>
<key> <string>categories</string> </key>
<value>
<tuple>
<string>elementary_type/lines</string>
</tuple>
</value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>A list of entry point for JSON Forms</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>json_form_property</string> </value>
</item>
<item>
<key> <string>mode</string> </key>
<value> <string>w</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Standard Property</string> </value>
</item>
<item>
<key> <string>property_default</string> </key>
<value> <string>python: ()</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Folder" module="OFS.Folder"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_objects</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>erp5_json_rpc_api</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ERP5 Form" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_objects</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>action</string> </key>
<value> <string>Base_edit</string> </value>
</item>
<item>
<key> <string>action_title</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>edit_order</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>encoding</string> </key>
<value> <string>UTF-8</string> </value>
</item>
<item>
<key> <string>enctype</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>group_list</string> </key>
<value>
<list>
<string>left</string>
<string>right</string>
<string>center</string>
<string>bottom</string>
<string>hidden</string>
</list>
</value>
</item>
<item>
<key> <string>groups</string> </key>
<value>
<dictionary>
<item>
<key> <string>bottom</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>center</string> </key>
<value>
<list>
<string>my_json_form_list</string>
<string>my_error_handler_script_id</string>
</list>
</value>
</item>
<item>
<key> <string>hidden</string> </key>
<value>
<list/>
</value>
</item>
<item>
<key> <string>left</string> </key>
<value>
<list>
<string>my_id</string>
<string>my_title</string>
<string>my_description</string>
</list>
</value>
</item>
<item>
<key> <string>right</string> </key>
<value>
<list>
<string>my_translated_validation_state_title</string>
</list>
</value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>JsonRpcService_view</string> </value>
</item>
<item>
<key> <string>method</string> </key>
<value> <string>POST</string> </value>
</item>
<item>
<key> <string>name</string> </key>
<value> <string>JsonRpcService_view</string> </value>
</item>
<item>
<key> <string>pt</string> </key>
<value> <string>form_view</string> </value>
</item>
<item>
<key> <string>row_length</string> </key>
<value> <int>4</int> </value>
</item>
<item>
<key> <string>stored_encoding</string> </key>
<value> <string>UTF-8</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>JSON RPC Service</string> </value>
</item>
<item>
<key> <string>unicode_mode</string> </key>
<value> <int>0</int> </value>
</item>
<item>
<key> <string>update_action</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>update_action_title</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>my_description</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_view_mode_description</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>delegated_list</string> </key>
<value>
<list>
<string>title</string>
</list>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>my_error_handler_script_id</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_string_field</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>Error Handler Script ID</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>my_id</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_view_mode_id</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>delegated_list</string> </key>
<value>
<list>
<string>description</string>
<string>title</string>
</list>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>my_json_form_list</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>description</string> </key>
<value> <string>api.entry.point | JSONForm ID</string> </value>
</item>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_lines_field</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>Json Forms</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>my_title</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_view_mode_title</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="ProxyField" module="Products.ERP5Form.ProxyField"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>id</string> </key>
<value> <string>my_translated_validation_state_title</string> </value>
</item>
<item>
<key> <string>message_values</string> </key>
<value>
<dictionary>
<item>
<key> <string>external_validator_failed</string> </key>
<value> <string>The input failed the external validator.</string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>overrides</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>tales</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string></string> </value>
</item>
</dictionary>
</value>
</item>
<item>
<key> <string>values</string> </key>
<value>
<dictionary>
<item>
<key> <string>field_id</string> </key>
<value> <string>my_view_mode_translated_workflow_state_title</string> </value>
</item>
<item>
<key> <string>form_id</string> </key>
<value> <string>Base_viewFieldLibrary</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
##############################################################################
# coding: utf-8
# Copyright (c) 2002-2023 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import io
import json
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from erp5.component.document.OpenAPITypeInformation import byteify
class JsonRpcAPITestCase(ERP5TypeTestCase):
_type_id = 'JSON RPC Service'
_action_list_text = ''
def addJSONForm(self, script_id, input_json_schema=None,
after_method_id=None, output_json_schema=None):
self.portal.portal_callables.newContent(
portal_type='JSON Form',
id=script_id,
text_content=input_json_schema,
after_method_id=after_method_id,
output_schema=output_json_schema
)
self.tic()
self._json_form_id_to_cleanup.append(script_id)
def addPythonScript(self, script_id, params, body):
skin_folder = self.portal.portal_skins['custom']
skin_folder.manage_addProduct['ERP5'].addPythonScriptThroughZMI(
id=script_id)
self.script = skin_folder.get(script_id)
self.script.setParameterSignature(params)
self.script.setBody(body)
self.tic()
self._python_script_id_to_cleanup.append(script_id)
self.portal.changeSkin(None)
def afterSetUp(self):
self.connector = self.portal.portal_web_services.newContent(
portal_type=self._type_id,
title=self.id(),
json_form_list=self._action_list_text.split('\n')
)
self.tic()
self._python_script_id_to_cleanup = []
self._json_form_id_to_cleanup = []
def beforeTearDown(self):
self.abort()
self.tic()
self.portal.portal_web_services.manage_delObjects([self.connector.getId()])
self.tic()
if self._json_form_id_to_cleanup:
self.portal.portal_callables.manage_delObjects(self._json_form_id_to_cleanup)
skin_folder = self.portal.portal_skins['custom']
if self._python_script_id_to_cleanup:
skin_folder.manage_delObjects(self._python_script_id_to_cleanup)
self.tic()
"""
_type_id = NotImplemented # type: str
_open_api_schema = NotImplemented # type: str
_open_api_schema_content_type = 'application/json'
_public_api = True
def afterSetUp(self):
if self._public_api:
open_api_type.setTypeWorkflowList(['publication_workflow'])
self.tic()
self.connector = self.portal.portal_web_services.newContent(
portal_type=self._type_id,
title=self.id(),
)
if self._public_api:
self.connector.publish()
else:
self.connector.validate()
self.tic()
self._python_script_id_to_cleanup = []
def beforeTearDown(self):
skin_folder = self.portal.portal_skins['custom']
if self._python_script_id_to_cleanup:
skin_folder.manage_delObjects(self._python_script_id_to_cleanup)
self.tic()
def addPythonScript(self, script_id, params, body):
skin_folder = self.portal.portal_skins['custom']
skin_folder.manage_addProduct['ERP5'].addPythonScriptThroughZMI(
id=script_id)
self._python_script_id_to_cleanup.append(script_id)
self.script = skin_folder.get(script_id)
self.script.setParameterSignature(params)
self.script.setBody(body)
self.tic()
self.portal.changeSkin(None)
"""
class TestJsonRpcAPIConnectorView(JsonRpcAPITestCase):
def test_view(self):
ret = self.publish(self.connector.getPath(), user='ERP5TypeTestCase')
self.assertEqual(ret.getStatus(), 200)
self.assertEqual(ret.getHeader('content-type'), 'text/html; charset=utf-8')
self.assertIn(b'<html', ret.getBody())
def test_viewOpenAPIAsJson(self):
ret = self.publish(
self.connector.getPath() + '/viewOpenAPIAsJson', user='ERP5TypeTestCase')
self.assertEqual(ret.getStatus(), 200)
body = json.load(io.BytesIO(ret.getBody()))
server_url = body['servers'][0]['url']
self.assertIn(self.connector.getPath(), server_url)
def test_erp5_document_methods(self):
self.connector.setTitle('Pet Store')
self.assertEqual(self.connector.getTitle(), 'Pet Store')
self.tic()
ret = self.publish(
self.connector.getPath() + '/getTitle', user='ERP5TypeTestCase')
self.assertEqual(ret.getStatus(), 200)
self.assertEqual(ret.getBody(), b'Pet Store')
def test_portal_skins_acquisition(self):
self.assertEqual(self.connector.JsonRpcService_view.getId(), 'JsonRpcService_view')
ret = self.publish(
self.connector.getPath() + '/JsonRpcService_view', user='ERP5TypeTestCase')
self.assertEqual(ret.getStatus(), 200)
self.assertNotIn(b'Site Error', ret.getBody())
def test_non_existing_attribute(self):
with self.assertRaises(AttributeError):
_ = self.connector.non_existing_attribute
ret = self.publish(
self.connector.getPath() + '/non_existing_attribute',
user='ERP5TypeTestCase')
self.assertEqual(ret.getStatus(), 404)
self.assertEqual(ret.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(ret.getBody()), {
"type": "not-found",
"title": 'non_existing_attribute'
})
class TestJsonRpcAPIErrorHandling(JsonRpcAPITestCase):
_action_list_text = '''error.handling.missing.callable | JsonRpcService_doesNotExist
error.handling.callable | JsonRpcService_testExample'''
def test_requestErrorHandling_wrongContentType(self):
response = self.publish(
self.connector.getPath() + '/error.handling.missing.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
json.dumps({}).encode()))
self.assertEqual(response.getStatus(), 415)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"status": 415,
"type": "unexpected-media-type",
"title": 'Request Content-Type must be "application/json", not ""'
})
def test_requestErrorHandling_wrongHTTPMethod(self):
response = self.publish(
self.connector.getPath() + '/error.handling.missing.callable',
user='ERP5TypeTestCase',
request_method='GET',
stdin=io.BytesIO(
json.dumps({}).encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 405)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"status": 405,
"type": "not-allowed-http-method",
"title": "Only HTTP POST accepted"
})
def test_requestErrorHandling_unknownActionReference(self):
response = self.publish(
self.connector.getPath() + '/error.handling.unknown.reference',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
json.dumps({}).encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 404)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "not-found",
"title": "error.handling.unknown.reference"
})
def test_requestErrorHandling_unknownCallable(self):
response = self.publish(
self.connector.getPath() + '/error.handling.missing.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
json.dumps({}).encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "AttributeError: 'RequestContainer' object has no attribute 'JsonRpcService_doesNotExist'"
})
def test_requestErrorHandling_notJsonBody(self):
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'1+2:"'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 400)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"status": 400,
"type": "not-parsable-json-content",
"title": "Extra data: line 1 column 2 - line 1 column 6 (char 1 - 5)"
})
def test_requestErrorHandling_notJsonDict(self):
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'[]'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 400)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"status": 400,
"type": "not-json-object-content",
"title": "Did not received a JSON Object"
})
def test_requestErrorHandling_badWebServiceConfiguration(self):
self.connector.edit(json_form_list=('foobarmissingseparator',))
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: Unparsable configuration: foobarmissingseparator"
})
def test_requestErrorHandling_notJsonForm(self):
self.connector.edit(json_form_list=('not.a.json.form | JsonRpcService_view',))
response = self.publish(
self.connector.getPath() + '/not.a.json.form',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: JsonRpcService_view is not a JSON Form"
})
def test_requestErrorHandling_invalidInputJsonSchema(self):
self.addJSONForm(
'JsonRpcService_testExample',
'1/2',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
'title': 'ValueError: Extra data: line 1 column 2 - line 1 column 4 (char 1 - 3)',
'type': 'unknown-error'
})
def test_requestErrorHandling_unknownAfterMethod(self):
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='THISSCRIPTDOESNOTEXIST',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "AttributeError: 'RequestContainer' object has no attribute 'THISSCRIPTDOESNOTEXIST'"
})
def test_requestErrorHandling_failingAfterMethod(self):
self.addPythonScript(
'JsonRpcService_fail',
'data_dict, json_form',
'1//0',
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_fail',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ZeroDivisionError: integer division or modulo by zero"
})
def test_requestErrorHandling_abortTransactionOnError(self):
self.addPythonScript(
'JsonRpcService_fail',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'1/0',
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_fail',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ZeroDivisionError: integer division or modulo by zero"
})
self.assertNotEqual(self.portal.getTitle(), "ooops")
def test_requestErrorHandling_noOutputWithOutputSchema(self):
self.addPythonScript(
'JsonRpcService_returnNothing',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
output_json_schema='''{"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {"foo": {"type": "string"}}}''',
after_method_id='JsonRpcService_returnNothing',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: JsonRpcService_testExample has an output schema but response is empty"
})
self.assertNotEqual(self.portal.getTitle(), "ooops")
def test_requestErrorHandling_emptyOutputWithOutputSchema(self):
self.addPythonScript(
'JsonRpcService_returnNothing',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'return {}'
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
output_json_schema='''{"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {"foo": {"type": "string"}}}''',
after_method_id='JsonRpcService_returnNothing',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: JsonRpcService_testExample has an output schema but response is empty"
})
self.assertNotEqual(self.portal.getTitle(), "ooops")
def test_requestErrorHandling_invalidOutputWithOutputSchema(self):
self.addPythonScript(
'JsonRpcService_returnNothing',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'return {"foo": 2}'
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
output_json_schema='''{"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {"foo": {"type": "string"}}}''',
after_method_id='JsonRpcService_returnNothing',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: 2 is not of type u'string'"
})
self.assertNotEqual(self.portal.getTitle(), "ooops")
def test_requestErrorHandling_outputWithoutOutputSchema(self):
self.addPythonScript(
'JsonRpcService_returnNothing',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'return {"foo": 2}'
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_returnNothing',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: JsonRpcService_testExample does not have an output schema but response is not empty"
})
self.assertNotEqual(self.portal.getTitle(), "ooops")
class TestJsonRpcAPICustomErrorHandling(JsonRpcAPITestCase):
_action_list_text = '''error.handling.missing.callable | JsonRpcService_doesNotExist
error.handling.callable | JsonRpcService_testExample'''
def afterSetUp(self):
super(TestJsonRpcAPICustomErrorHandling, self).afterSetUp()
script_id = 'JsonRpcService_testErrorHandler'
self.connector.setErrorHandlerScriptId(script_id)
self.addPythonScript(
script_id,
'exception',
'''
from erp5.component.document.JsonRpcAPIService import JsonRpcAPIBadContentType, JsonRpcAPINotAllowedHttpMethod, JsonRpcAPINotJsonDictContent, JsonRpcAPINotParsableJsonContent, JsonRpcAPIInvalidJsonDictContent
if isinstance(exception, JsonRpcAPIBadContentType):
return {"JsonRpcAPIBadContentType": "ScriptHandling %s" % str(exception)}
if isinstance(exception, JsonRpcAPINotAllowedHttpMethod):
return {"JsonRpcAPINotAllowedHttpMethod": "ScriptHandling %s" % str(exception)}
if isinstance(exception, JsonRpcAPINotJsonDictContent):
return {"JsonRpcAPINotJsonDictContent": "ScriptHandling %s" % str(exception)}
if isinstance(exception, JsonRpcAPINotParsableJsonContent):
return {"JsonRpcAPINotParsableJsonContent": "ScriptHandling %s" % str(exception)}
if isinstance(exception, JsonRpcAPIInvalidJsonDictContent):
return {"JsonRpcAPIInvalidJsonDictContent": "ScriptHandling %s" % str(exception)}
return {"unhandled exception type": str(exception)}
'''
)
self.tic()
def test_requestCustomErrorHandling_wrongContentType(self):
response = self.publish(
self.connector.getPath() + '/error.handling.missing.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
json.dumps({}).encode()))
self.assertEqual(response.getStatus(), 415)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"JsonRpcAPIBadContentType": 'ScriptHandling Request Content-Type must be "application/json", not ""'
})
def test_requestCustomErrorHandling_wrongHTTPMethod(self):
response = self.publish(
self.connector.getPath() + '/error.handling.missing.callable',
user='ERP5TypeTestCase',
request_method='GET',
stdin=io.BytesIO(
json.dumps({}).encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 405)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"JsonRpcAPINotAllowedHttpMethod": 'ScriptHandling Only HTTP POST accepted'
})
def test_requestCustomErrorHandling_unknownActionReference(self):
response = self.publish(
self.connector.getPath() + '/error.handling.unknown.reference',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
json.dumps({}).encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 404)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "not-found",
"title": "error.handling.unknown.reference"
})
def test_requestCustomErrorHandling_unknownCallable(self):
response = self.publish(
self.connector.getPath() + '/error.handling.missing.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
json.dumps({}).encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "AttributeError: 'RequestContainer' object has no attribute 'JsonRpcService_doesNotExist'"
})
def test_requestCustomErrorHandling_notJsonBody(self):
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'1+2:"'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 400)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"JsonRpcAPINotParsableJsonContent": 'ScriptHandling Extra data: line 1 column 2 - line 1 column 6 (char 1 - 5)'
})
def test_requestCustomErrorHandling_notJsonDict(self):
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'[]'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 400)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"JsonRpcAPINotJsonDictContent": "ScriptHandling Did not received a JSON Object"
})
def test_requestCustomErrorHandling_badWebServiceConfiguration(self):
self.connector.edit(json_form_list=('foobarmissingseparator',))
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: Unparsable configuration: foobarmissingseparator"
})
def test_requestCustomErrorHandling_notJsonForm(self):
self.connector.edit(json_form_list=('not.a.json.form | JsonRpcService_view',))
response = self.publish(
self.connector.getPath() + '/not.a.json.form',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ValueError: JsonRpcService_view is not a JSON Form"
})
def test_requestCustomErrorHandling_invalidInputJsonSchema(self):
self.addJSONForm(
'JsonRpcService_testExample',
'1/2',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
'title': 'ValueError: Extra data: line 1 column 2 - line 1 column 4 (char 1 - 3)',
'type': 'unknown-error'
})
def test_requestCustomErrorHandling_unknownAfterMethod(self):
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='THISSCRIPTDOESNOTEXIST',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "AttributeError: 'RequestContainer' object has no attribute 'THISSCRIPTDOESNOTEXIST'"
})
def test_requestCustomErrorHandling_failingAfterMethod(self):
self.addPythonScript(
'JsonRpcService_fail',
'data_dict, json_form',
'1//0',
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_fail',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ZeroDivisionError: integer division or modulo by zero"
})
def test_requestCustomErrorHandling_abortTransactionOnError(self):
self.addPythonScript(
'JsonRpcService_fail',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'1/0',
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_fail',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 500)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
json.loads(response.getBody()), {
"type": "unknown-error",
"title": "ZeroDivisionError: integer division or modulo by zero"
})
self.assertNotEqual(self.portal.getTitle(), "ooops")
def test_requestCustomErrorHandling_failingErrorHandling(self):
script_id = 'JsonRpcService_testErrorHandler2'
self.connector.setErrorHandlerScriptId(script_id)
self.addPythonScript(
script_id,
'exception',
'1//0'
)
self.addPythonScript(
'JsonRpcService_fail',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'from erp5.component.document.JsonRpcAPIService import JsonRpcAPIError\n'
'class CustomError(JsonRpcAPIError):\n'
' type = "custom-error-type"\n'
' status = 417\n'
'raise CustomError("custom error title")',
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_fail',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'}
)
self.assertNotEqual(self.portal.getTitle(), "ooops")
self.assertEqual(
response.getBody(),
json.dumps(
{
'type': 'unknown-error',
'title': "ZeroDivisionError: integer division or modulo by zero"
}).encode())
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(response.getStatus(), 500)
def test_requestCustomErrorHandling_emptyResult(self):
script_id = 'JsonRpcService_testErrorHandler2'
self.connector.setErrorHandlerScriptId(script_id)
self.addPythonScript(
script_id,
'exception',
'return'
)
self.addPythonScript(
'JsonRpcService_fail',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'from erp5.component.document.JsonRpcAPIService import JsonRpcAPIError\n'
'class CustomError(JsonRpcAPIError):\n'
' type = "custom-error-type"\n'
' status = 417\n'
'raise CustomError("custom error title")',
)
self.addJSONForm(
'JsonRpcService_testExample',
'{}',
after_method_id='JsonRpcService_fail',
)
response = self.publish(
self.connector.getPath() + '/error.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'}
)
self.assertNotEqual(self.portal.getTitle(), "ooops")
self.assertEqual(
response.getBody(),
json.dumps(
{
'type': 'custom-error-type',
'title': "custom error title",
'status': 417
}).encode())
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(response.getStatus(), 417)
class TestJsonRpcAPIJsonFormHandling(JsonRpcAPITestCase):
_action_list_text = '''json.form.handling.callable | JsonRpcService_callFromTest'''
def test_jsonFormHandling_emptyUseCase(self):
# No input json schema
# No output json schema
# No BODY
self.addJSONForm(
'JsonRpcService_callFromTest',
'{}',
)
response = self.publish(
self.connector.getPath() + '/json.form.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 200)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
byteify(json.loads(response.getBody())),
{
'type': 'success-type',
'title': "query completed",
'status': 200
})
def test_jsonFormHandling_noInputSchemaAndBodyContent(self):
self.addJSONForm(
'JsonRpcService_callFromTest',
'{}',
)
response = self.publish(
self.connector.getPath() + '/json.form.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{"a": "b"}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getStatus(), 200)
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
byteify(json.loads(response.getBody())),
{
'type': 'success-type',
'title': "query completed",
'status': 200
})
def test_jsonFormHandling_invalidBodyContent(self):
self.addJSONForm(
'JsonRpcService_callFromTest',
'''{"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {"foo": {"type": "string"}}}'''
)
response = self.publish(
self.connector.getPath() + '/json.form.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{"foo": 1}'.encode()),
env={'CONTENT_TYPE': 'application/json'})
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(
response.getBody(),
json.dumps(
{
'title': "1 is not of type u'string'",
'type': "invalid-json-object-content",
'status': 400
}).encode())
self.assertEqual(response.getStatus(), 400)
def test_jsonFormHandling_customError(self):
self.addPythonScript(
'JsonRpcService_customError',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'from erp5.component.document.JsonRpcAPIService import JsonRpcAPIError\n'
'class CustomError(JsonRpcAPIError):\n'
' type = "custom-error-type"\n'
' status = 417\n'
'raise CustomError("custom error title")',
)
self.addJSONForm(
'JsonRpcService_callFromTest',
'{}',
after_method_id='JsonRpcService_customError',
)
response = self.publish(
self.connector.getPath() + '/json.form.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'}
)
self.assertNotEqual(self.portal.getTitle(), "ooops")
self.assertEqual(
response.getBody(),
json.dumps(
{
'type': 'custom-error-type',
'title': "custom error title",
'status': 417
}).encode())
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(response.getStatus(), 417)
def test_jsonFormHandling_customErrorCustomerHandler(self):
script_id = 'JsonRpcService_testErrorHandler'
self.connector.setErrorHandlerScriptId(script_id)
self.addPythonScript(
script_id,
'exception',
'''
return {
"status2": exception.status,
"detail2": exception.detail,
"type2": exception.type,
"title2": str(exception)
}
'''
)
self.addPythonScript(
'JsonRpcService_customError',
'data_dict, json_form',
'context.getPortalObject().setTitle("ooops")\n'
'from erp5.component.document.JsonRpcAPIService import JsonRpcAPIError\n'
'class CustomError(JsonRpcAPIError):\n'
' type = "custom-error-type"\n'
' status = 417\n'
'raise CustomError("custom error title")',
)
self.addJSONForm(
'JsonRpcService_callFromTest',
'{}',
after_method_id='JsonRpcService_customError',
)
response = self.publish(
self.connector.getPath() + '/json.form.handling.callable',
user='ERP5TypeTestCase',
request_method='POST',
stdin=io.BytesIO(
'{}'.encode()),
env={'CONTENT_TYPE': 'application/json'}
)
self.assertNotEqual(self.portal.getTitle(), "ooops")
self.assertEqual(
response.getBody(),
json.dumps(
{
'type2': 'custom-error-type',
'title2': "custom error title",
'status2': 417
}).encode())
self.assertEqual(response.getHeader('content-type'), 'application/json')
self.assertEqual(response.getStatus(), 417)
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Test Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testJsonRpcAPIService</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testJsonRpcAPIService</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>text_content_error_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.Workflow"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_log</string> </key>
<value>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
erp5_open_api
erp5_json_form
\ No newline at end of file
Framework for implementing web services with json schemas
\ No newline at end of file
JSON RPC Service | view
\ No newline at end of file
document.erp5.JsonRpcAPIService
\ No newline at end of file
Web Service Tool | JSON RPC Service
\ No newline at end of file
JSON RPC Service
\ No newline at end of file
JSON RPC Service | JsonRpcService
\ No newline at end of file
JSON RPC Service | edit_workflow
JSON RPC Service | validation_workflow
\ No newline at end of file
erp5_json_rpc_api
\ No newline at end of file
test.erp5.testJsonRpcAPIService
\ No newline at end of file
erp5_full_text_mroonga_catalog
\ No newline at end of file
erp5_json_rpc_api
\ No newline at end of file
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