matthias/mayan-edms
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add fields to the workflow HTTP POST action

Browse Source 
Add username, password, and headers fields.
Update the timeout field to support templates.

Signed-off-by: Roberto Rosario <roberto.rosario@mayan-edms.com>
This commit is contained in:
Roberto Rosario
2019-11-08 00:18:36 -04:00
parent e91232f722
commit 6e1bcae0fe
3 changed files with 255 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
mayan/apps/document_states/tests/literals.py
View File

@@ -4,6 +4,23 @@ from ..literals import FIELD_TYPE_CHOICE_CHAR
TEST_INDEX_LABEL = 'test workflow index'
TEST_HEADERS_KEY = 'test key'
TEST_HEADERS_VALUE = 'test value'
TEST_HEADERS_JSON = '{{"{}": "{}"}}'.format(
TEST_HEADERS_KEY, TEST_HEADERS_VALUE
)
TEST_HEADERS_JSON_TEMPLATE_KEY = 'test key'
TEST_HEADERS_JSON_TEMPLATE_VALUE = '{{ document.label }}'
TEST_HEADERS_JSON_TEMPLATE = '{{"{}": "{}"}}'.format(
TEST_HEADERS_JSON_TEMPLATE_KEY, TEST_HEADERS_JSON_TEMPLATE_VALUE
)
TEST_HEADERS_AUTHENTICATION_KEY = 'Authorization'
TEST_HEADERS_AUTHENTICATION_VALUE = 'Basic dGVzdHVzZXJuYW1lOnRlc3RwYXNzd29yZA=='
TEST_PAYLOAD_JSON = '{"label": "label"}'
TEST_PAYLOAD_TEMPLATE_DOCUMENT_LABEL = '{"label": "{{ document.label }}"}'
TEST_SERVER_USERNAME = 'testusername'
TEST_SERVER_PASSWORD = 'testpassword'
TEST_WORKFLOW_LABEL = 'test workflow label'
TEST_WORKFLOW_INTERNAL_NAME = 'test_workflow_label'
TEST_WORKFLOW_LABEL_EDITED = 'test workflow label edited'

134
mayan/apps/document_states/tests/test_workflow_actions.py Normal file
View File

@@ -0,0 +1,134 @@
from __future__ import unicode_literals
import json
import mock
from mayan.apps.common.tests.base import GenericViewTestCase
from mayan.apps.common.tests.mixins import TestServerTestCaseMixin
from mayan.apps.common.tests.mocks import request_method_factory
from mayan.apps.document_states.tests.mixins import WorkflowTestMixin
from mayan.apps.document_states.tests.base import ActionTestCase
from ..workflow_actions import HTTPPostAction
from .literals import (
TEST_HEADERS_AUTHENTICATION_KEY, TEST_HEADERS_AUTHENTICATION_VALUE,
TEST_HEADERS_KEY, TEST_HEADERS_JSON, TEST_HEADERS_JSON_TEMPLATE,
TEST_HEADERS_JSON_TEMPLATE_KEY, TEST_HEADERS_VALUE, TEST_PAYLOAD_JSON,
TEST_PAYLOAD_TEMPLATE_DOCUMENT_LABEL, TEST_SERVER_USERNAME,
TEST_SERVER_PASSWORD
)
class HTTPPostWorkflowActionTestCase(
TestServerTestCaseMixin, GenericViewTestCase, WorkflowTestMixin,
ActionTestCase
):
auto_add_test_view = True
@mock.patch('requests.api.request')
def test_http_post_action_simple(self, mock_object):
mock_object.side_effect = request_method_factory(test_case=self)
action = HTTPPostAction(
form_data={
'url': self.testserver_url,
}
)
action.execute(context={'document': self.test_document})
self.assertFalse(self.test_view_request is None)
@mock.patch('requests.api.request')
def test_http_post_action_payload_simple(self, mock_object):
mock_object.side_effect = request_method_factory(test_case=self)
action = HTTPPostAction(
form_data={
'url': self.testserver_url,
'payload': TEST_PAYLOAD_JSON,
}
)
action.execute(context={'document': self.test_document})
self.assertEqual(
json.loads(self.test_view_request.body),
{'label': 'label'}
)
@mock.patch('requests.api.request')
def test_http_post_action_payload_template(self, mock_object):
mock_object.side_effect = request_method_factory(test_case=self)
action = HTTPPostAction(
form_data={
'url': self.testserver_url,
'payload': TEST_PAYLOAD_TEMPLATE_DOCUMENT_LABEL,
}
)
action.execute(context={'document': self.test_document})
self.assertEqual(
json.loads(self.test_view_request.body),
{'label': self.test_document.label}
)
@mock.patch('requests.api.request')
def test_http_post_action_headers_simple(self, mock_object):
mock_object.side_effect = request_method_factory(test_case=self)
action = HTTPPostAction(
form_data={
'url': self.testserver_url,
'headers': TEST_HEADERS_JSON,
}
)
action.execute(context={'document': self.test_document})
self.assertTrue(
TEST_HEADERS_KEY in self.test_view_request.META,
)
self.assertEqual(
self.test_view_request.META[TEST_HEADERS_KEY], TEST_HEADERS_VALUE
)
@mock.patch('requests.api.request')
def test_http_post_action_headers_template(self, mock_object):
mock_object.side_effect = request_method_factory(test_case=self)
action = HTTPPostAction(
form_data={
'url': self.testserver_url,
'headers': TEST_HEADERS_JSON_TEMPLATE,
}
)
action.execute(context={'document': self.test_document})
self.assertTrue(
TEST_HEADERS_JSON_TEMPLATE_KEY in self.test_view_request.META,
)
self.assertEqual(
self.test_view_request.META[TEST_HEADERS_JSON_TEMPLATE_KEY],
self.test_document.label
)
@mock.patch('requests.api.request')
def test_http_post_action_authentication(self, mock_object):
mock_object.side_effect = request_method_factory(test_case=self)
action = HTTPPostAction(
form_data={
'url': self.testserver_url,
'username': TEST_SERVER_USERNAME,
'password': TEST_SERVER_PASSWORD
}
)
action.execute(context={'document': self.test_document})
self.assertTrue(
TEST_HEADERS_AUTHENTICATION_KEY in self.test_view_request.META,
)
self.assertEqual(
self.test_view_request.META[TEST_HEADERS_AUTHENTICATION_KEY],
TEST_HEADERS_AUTHENTICATION_VALUE
)

141
mayan/apps/document_states/workflow_actions.py
View File

@@ -101,7 +101,10 @@ class HTTPPostAction(WorkflowAction):
}, 'timeout': {
'label': _('Timeout'),
'class': 'django.forms.IntegerField', 'default': DEFAULT_TIMEOUT,
'help_text': _('Time in seconds to wait for a response.'),
'help_text': _(
'Time in seconds to wait for a response. Can be a static '
'value or a template. '
),
'required': True
}, 'payload': {
@@ -116,53 +119,117 @@ class HTTPPostAction(WorkflowAction):
'"workflow_instance", "datetime", "transition", "user", '
'and "comment" attributes.'
), 'required': False
}
},
},
}, 'username': {
'label': _('Username'),
'class': 'django.forms.CharField', 'kwargs': {
'help_text': _(
'Username to use for making the request with HTTP Basic '
'Auth. Can be a static value or a template. Templates '
'receive the workflow log entry instance as part of '
'their context via the variable "entry_log". '
'The "entry_log" in turn provides the '
'"workflow_instance", "datetime", "transition", "user", '
'and "comment" attributes.'
), 'max_length': 192, 'required': False
},
}, 'password': {
'label': _('Password'),
'class': 'django.forms.CharField', 'kwargs': {
'help_text': _(
'Password to use for making the request with HTTP Basic '
'Auth. Can be a static value or a template. Templates '
'receive the workflow log entry instance as part of '
'their context via the variable "entry_log". '
'The "entry_log" in turn provides the '
'"workflow_instance", "datetime", "transition", "user", '
'and "comment" attributes.'
), 'max_length': 192, 'required': False
},
}, 'headers': {
'label': _('Headers'),
'class': 'django.forms.CharField', 'kwargs': {
'help_text': _(
'Headers to send with the HTTP request. Must be in JSON '
'format. Can be a static value or a template. Templates '
'receive the workflow log entry instance as part of '
'their context via the variable "entry_log". '
'The "entry_log" in turn provides the '
'"workflow_instance", "datetime", "transition", "user", '
'and "comment" attributes.'
), 'required': False
},
}
}
field_order = ('url', 'timeout', 'payload')
field_order = (
'url', 'username', 'password', 'headers', 'timeout', 'payload'
)
label = _('Perform a POST request')
widgets = {
'payload': {
'class': 'django.forms.widgets.Textarea', 'kwargs': {
'attrs': {'rows': '10'},
}
},
'headers': {
'class': 'django.forms.widgets.Textarea', 'kwargs': {
'attrs': {'rows': '10'},
}
}
}
def render_load(self, field_name, context):
"""
Method to perform a template render and subsequent JSON load.
"""
render_result = self.render(
field_name=field_name, context=context
) or '{}'
try:
load_result = json.loads(render_result, strict=False)
except Exception as exception:
raise WorkflowStateActionError(
_('%(field_name)s JSON error: %(exception)s') % {
'field_name': field_name, 'exception': exception
}
)
logger.debug('load result: %s', load_result)
return load_result
def render(self, field_name, context):
try:
result = Template(self.form_data.get(field_name, '')).render(
context=Context(context)
)
except Exception as exception:
raise WorkflowStateActionError(
_('%(field_name)s template error: %(exception)s') % {
'field_name': field_name, 'exception': exception
}
)
logger.debug('%s template result: %s', field_name, result)
return result
def execute(self, context):
self.url = self.form_data.get('url')
self.payload = self.form_data.get('payload')
url = self.render(field_name='url', context=context)
username = self.render(field_name='username', context=context)
password = self.render(field_name='password', context=context)
timeout = self.render(field_name='timeout', context=context)
headers = self.render_load(field_name='headers', context=context)
payload = self.render_load(field_name='payload', context=context)
try:
url = Template(self.url).render(
context=Context(context)
)
except Exception as exception:
raise WorkflowStateActionError(
_('URL template error: %s') % exception
authentication = None
if username or password:
authentication = requests.auth.HTTPBasicAuth(
username=username, password=password
)
logger.debug('URL template result: %s', url)
try:
result = Template(self.payload or '{}').render(
context=Context(context)
)
except Exception as exception:
raise WorkflowStateActionError(
_('Payload template error: %s') % exception
)
logger.debug('payload template result: %s', result)
try:
payload = json.loads(result, strict=False)
except Exception as exception:
raise WorkflowStateActionError(
_('Payload JSON error: %s') % exception
)
logger.debug('payload json result: %s', payload)
requests.post(url=url, json=payload, timeout=self.form_data['timeout'])
requests.post(
url=url, json=payload, timeout=timeout,
auth=authentication, headers=headers
)