shaba/awx
1
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2024-10-30 22:21:13 +03:00
Code Releases Activity

save/validate messages

Browse Source
This commit is contained in:
Jim Ladd 2019-08-14 11:06:06 -07:00
parent 3bb0aa4eec
commit 13b9679496
3 changed files with 116 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
awx/api/serializers.py
View File

@ -4160,6 +4160,43 @@ class NotificationTemplateSerializer(BaseSerializer):
d['recent_notifications'] = self._recent_notifications(obj)
return d
def validate_messages(self, messages):
if messages is None:
return None
error_list = []
collected_messages = []
# Validate structure / content types
if not isinstance(messages, dict):
error_list.append(_("Expected dict for 'messages' field, found {}".format(type(messages))))
else:
for event in messages:
if event not in ['started', 'success', 'error']:
error_list.append(_("Event '{}' invalid, must be one of 'started', 'success', or 'error'").format(event))
continue
event_messages = messages[event]
if event_messages is None:
continue
if not isinstance(event_messages, dict):
error_list.append(_("Expected dict for event '{}', found {}").format(event, type(event_messages)))
continue
for message_type in event_messages:
if message_type not in ['message', 'body']:
error_list.append(_("Message type '{}' invalid, must be either 'message' or 'body'").format(message_type))
continue
message = event_messages[message_type]
if message is None:
continue
if not isinstance(message, str):
error_list.append(_("Expected string for '{}', found {}, ").format(message_type, type(message)))
continue
if message_type == 'message':
if '\n' in message:
error_list.append(_("Messages cannot contain newlines (found newline in {} event)".format(event)))
continue
collected_messages.append(message)
def validate(self, attrs):
from awx.api.views import NotificationTemplateDetail

20
awx/main/models/notifications.py
View File

@ -84,6 +84,26 @@ class NotificationTemplate(CommonModelNameNotUnique):
def save(self, *args, **kwargs):
new_instance = not bool(self.pk)
update_fields = kwargs.get('update_fields', [])
# preserve existing notification messages if not overwritten by new messages
if not new_instance:
old_nt = NotificationTemplate.objects.get(pk=self.id)
old_messages = old_nt.messages
new_messages = self.messages
if old_messages is not None:
for event in ['started', 'success', 'error']:
if not new_messages.get(event, {}) and old_messages.get(event, {}):
new_messages[event] = old_messages[event]
continue
if new_messages.get(event, {}) and old_messages.get(event, {}):
old_event_msgs = old_messages[event]
new_event_msgs = new_messages[event]
for msg_type in ['message', 'body']:
if msg_type not in new_event_msgs and old_event_msgs.get(msg_type, None):
new_event_msgs[msg_type] = old_event_msgs[msg_type]
new_messages.setdefault(event, None)
for field in filter(lambda x: self.notification_class.init_parameters[x]['type'] == "password",
self.notification_class.init_parameters):
if self.notification_configuration[field].startswith("$encrypted$"):

59
awx/main/tests/unit/api/serializers/test_notification_template_serializers.py Normal file
View File

@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
import pytest
from rest_framework.serializers import ValidationError
# AWX
from awx.api.serializers import NotificationTemplateSerializer
class StubNotificationTemplate():
notification_type = 'email'
class TestNotificationTemplateSerializer():
@pytest.mark.parametrize('valid_messages',
[None,
{'started': None},
{'started': {'message': None}},
{'started': {'message': 'valid'}},
{'started': {'body': 'valid'}},
{'started': {'message': 'valid', 'body': 'valid'}},
{'started': None, 'success': None, 'error': None},
{'started': {'message': None, 'body': None},
'success': {'message': None, 'body': None},
'error': {'message': None, 'body': None}},
{'started': {'message': '{{ job.id }}', 'body': '{{ job.status }}'},
'success': {'message': None, 'body': '{{ job_friendly_name }}'},
'error': {'message': '{{ url }}', 'body': None}},
{'started': {'body': '{{ job_summary_dict }}'}},
{'started': {'body': '{{ job.summary_fields.inventory.total_hosts }}'}},
{'started': {'body': u'Iñtërnâtiônàlizætiøn'}}
])
def test_valid_messages(self, valid_messages):
serializer = NotificationTemplateSerializer()
serializer.instance = StubNotificationTemplate()
serializer.validate_messages(valid_messages)
@pytest.mark.parametrize('invalid_messages',
[1,
[],
'',
{'invalid_event': ''},
{'started': 'should_be_dict'},
{'started': {'bad_message_type': ''}},
{'started': {'message': 1}},
{'started': {'message': []}},
{'started': {'message': {}}},
{'started': {'message': '{{ unclosed_braces'}},
{'started': {'message': '{{ undefined }}'}},
{'started': {'message': '{{ job.undefined }}'}},
{'started': {'message': '{{ job.id | bad_filter }}'}},
{'started': {'message': '{{ job.__class__ }}'}},
{'started': {'message': 'Newlines \n not allowed\n'}},
])
def test_invalid__messages(self, invalid_messages):
serializer = NotificationTemplateSerializer()
serializer.instance = StubNotificationTemplate()
with pytest.raises(ValidationError):
serializer.validate_messages(invalid_messages)