Commit 66fbfbf8 authored by Valentin Benozillo's avatar Valentin Benozillo

erp5_web_service: Add RESTAPIClientConnector test

parent 2e20ffd4
# -*- coding: utf-8 -*-
# Copyright (c) 2002-2015 Nexedi SA and Contributors. All Rights Reserved.
from json import dumps
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from httplib import HTTPSConnection
from erp5.component.mixin.RESTAPIClientConnectorMixin import RESTAPIClientConnectorMixin
import mock
expected_output_body_dict = {
u'output': 'output',
}
class HTTPResponse_getresponse():
def __init__(self, status=200):
self.status = status
def getheaders(self):
return [
('content-type', 'application/json'),
]
def read(self):
return dumps(expected_output_body_dict)
class RESTAPIError(Exception):
__allow_access_to_unprotected_subobjects__ = {
'header_dict': 1,
'body': 1,
'status': 1,
}
def __init__(self, header_dict, body, status):
super(RESTAPIError, self).__init__()
self.header_dict = header_dict
self.body = body
self.status = status
class RESTAPIClientConnector(RESTAPIClientConnectorMixin):
meta_type = 'REST API Client Connector'
security = RESTAPIClientConnectorMixin.security
ClientConnectorError = RESTAPIError
def _getAccessToken(self):
return 'access_token'
def getBaseUrl(self):
return 'https://example.com/'
def getCaCertificatePem(self):
return 'ca_certificate_pem'
def getBindAddress(self):
return 'bind_address'
class TestRESTAPIClientConnector(ERP5TypeTestCase):
def afterSetUp(self):
self.rest_api_client_connection = RESTAPIClientConnector(id='rest_api_client_connection')
def test_api_call(self):
input_body_dict = {
'input': 'input',
}
timeout = 1
with mock.patch(
'ssl.create_default_context',
) as mock_ssl_create_default_context, mock.patch(
'httplib.HTTPSConnection.request',
) as mock_https_connection_request, mock.patch(
'httplib.HTTPSConnection.getresponse',
return_value=HTTPResponse_getresponse()
), mock.patch('httplib.HTTPSConnection', return_value=HTTPSConnection) as mock_https_connection:
header_dict, body_dict, status = self.rest_api_client_connection.call(
archive_resource=None,
method='POST',
path='/path',
body=input_body_dict,
timeout=timeout,
)
# Check request
ssl_create_default_context_argument_dict = mock_ssl_create_default_context.call_args.kwargs
self.assertEqual(
ssl_create_default_context_argument_dict['cadata'],
'ca_certificate_pem'
)
https_connection_argument_dict = mock_https_connection.call_args.kwargs
self.assertTrue(
https_connection_argument_dict['timeout'] <= timeout
)
self.assertEqual(
https_connection_argument_dict['host'],
'example.com'
)
self.assertEqual(
https_connection_argument_dict['source_address'],
('bind_address', 0)
)
https_connection_request_argument_dict = mock_https_connection_request.call_args.kwargs
self.assertEqual(
https_connection_request_argument_dict['body'],
dumps(input_body_dict)
)
self.assertEqual(
https_connection_request_argument_dict['url'],
'/path'
)
self.assertEqual(
https_connection_request_argument_dict['headers']['Authorization'],
'Bearer access_token'
)
self.assertEqual(
https_connection_request_argument_dict['headers']['content-type'],
'application/json'
)
self.assertEqual(
https_connection_request_argument_dict['method'],
'POST'
)
# Check response
self.assertEqual(
header_dict['content-type'],
'application/json'
)
self.assertEqual(
body_dict,
expected_output_body_dict
)
self.assertEqual(
status,
200
)
\ No newline at end of file
<?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>testRESTAPIClientConnectorMixin</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>test.erp5.testRESTAPIClientConnectorMixin</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Test Component</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>
test.erp5.testFTPConnection
test.erp5.testRESTAPIClientConnectorMixin
test.erp5.testWebServiceTool
\ 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