Refactor notification backends to use CustomNotificationBase

awx/main/notifications/custom_notification_base.py

@@ -0,0 +1,14 @@
+# Copyright (c) 2019 Ansible, Inc.
+# All Rights Reserved.
+
+from django.utils.encoding import smart_text
+from django.utils.translation import ugettext_lazy as _
+
+
+class CustomNotificationBase(object):
+    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
+    DEFAULT_BODY = smart_text(_("{{ job_friendly_name }} #{{ job.id }} had status {{ job.status }}, view details at {{ url }}\n\n{{ job_summary_dict }}"))
+
+    default_messages = {"started": {"message": DEFAULT_MSG, "body": None},
+                        "success": {"message": DEFAULT_MSG, "body": None},
+                        "error": {"message": DEFAULT_MSG, "body": None}}

awx/main/notifications/email_backend.py

@@ -1,12 +1,13 @@
 # Copyright (c) 2016 Ansible, Inc.
 # All Rights Reserved.
 
-from django.utils.encoding import smart_text
 from django.core.mail.backends.smtp import EmailBackend
-from django.utils.translation import ugettext_lazy as _
+
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
+from CustomNotificationBase import DEFAULT_MSG, DEFAULT_BODY
 
 
-class CustomEmailBackend(EmailBackend):
+class CustomEmailBackend(EmailBackend, CustomNotificationBase):
 
     init_parameters = {"host": {"label": "Host", "type": "string"},
                        "port": {"label": "Port", "type": "int"},
@@ -17,15 +18,12 @@ class CustomEmailBackend(EmailBackend):
                        "sender": {"label": "Sender Email", "type": "string"},
                        "recipients": {"label": "Recipient List", "type": "list"},
                        "timeout": {"label": "Timeout", "type": "int", "default": 30}}
-
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    DEFAULT_BODY = smart_text(_("{{ job_friendly_name }} #{{ job.id }} had status {{ job.status }}, view details at {{ url }}\n\n{{ job_summary_dict }}"))
-    default_messages = {"started": {"message": DEFAULT_MSG, "body": DEFAULT_BODY},
-                        "success": {"message": DEFAULT_MSG, "body": DEFAULT_BODY},
-                        "error": {"message": DEFAULT_MSG, "body": DEFAULT_BODY}}
     recipient_parameter = "recipients"
     sender_parameter = "sender"
 
+    default_messages = {"started": {"message": DEFAULT_MSG, "body": DEFAULT_BODY},
+                        "success": {"message": DEFAULT_MSG, "body": DEFAULT_BODY},
+                        "error": {"message": DEFAULT_MSG, "body": DEFAULT_BODY}}
 
     def format_body(self, body):
         # leave body unchanged (expect a string)

awx/main/notifications/grafana_backend.py

@@ -8,24 +8,21 @@ import dateutil.parser as dp
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 
 logger = logging.getLogger('awx.main.notifications.grafana_backend')
 
 
-class GrafanaBackend(AWXBaseEmailBackend):
+class GrafanaBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"grafana_url": {"label": "Grafana URL", "type": "string"},
                        "grafana_key": {"label": "Grafana API Key", "type": "password"}}
     recipient_parameter = "grafana_url"
     sender_parameter = None
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
-
     def __init__(self, grafana_key,dashboardId=None, panelId=None, annotation_tags=None, grafana_no_verify_ssl=False, isRegion=True,
                  fail_silently=False, **kwargs):
         super(GrafanaBackend, self).__init__(fail_silently=fail_silently)

awx/main/notifications/hipchat_backend.py

@@ -7,12 +7,14 @@ import requests
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.hipchat_backend')
 
 
-class HipChatBackend(AWXBaseEmailBackend):
+class HipChatBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"token": {"label": "Token", "type": "password"},
                        "rooms": {"label": "Destination Rooms", "type": "list"},
@@ -23,11 +25,6 @@ class HipChatBackend(AWXBaseEmailBackend):
     recipient_parameter = "rooms"
     sender_parameter = "message_from"
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
-
     def __init__(self, token, color, api_url, notify, fail_silently=False, **kwargs):
         super(HipChatBackend, self).__init__(fail_silently=fail_silently)
         self.token = token

awx/main/notifications/irc_backend.py

@@ -9,12 +9,14 @@ import irc.client
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.irc_backend')
 
 
-class IrcBackend(AWXBaseEmailBackend):
+class IrcBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"server": {"label": "IRC Server Address", "type": "string"},
                        "port": {"label": "IRC Server Port", "type": "int"},
@@ -25,11 +27,6 @@ class IrcBackend(AWXBaseEmailBackend):
     recipient_parameter = "targets"
     sender_parameter = None
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
-
     def __init__(self, server, port, nickname, password, use_ssl, fail_silently=False, **kwargs):
         super(IrcBackend, self).__init__(fail_silently=fail_silently)
         self.server = server

awx/main/notifications/mattermost_backend.py

@@ -7,23 +7,20 @@ import json
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.mattermost_backend')
 
 
-class MattermostBackend(AWXBaseEmailBackend):
+class MattermostBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"mattermost_url": {"label": "Target URL", "type": "string"},
                        "mattermost_no_verify_ssl": {"label": "Verify SSL", "type": "bool"}}
     recipient_parameter = "mattermost_url"
     sender_parameter = None
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
-
     def __init__(self, mattermost_no_verify_ssl=False, mattermost_channel=None, mattermost_username=None,
                  mattermost_icon_url=None, fail_silently=False, **kwargs):
         super(MattermostBackend, self).__init__(fail_silently=fail_silently)

awx/main/notifications/pagerduty_backend.py

@@ -7,12 +7,15 @@ import pygerduty
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
+from CustomNotificationBase import DEFAULT_MSG
 
 logger = logging.getLogger('awx.main.notifications.pagerduty_backend')
 
 
-class PagerDutyBackend(AWXBaseEmailBackend):
+class PagerDutyBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"subdomain": {"label": "Pagerduty subdomain", "type": "string"},
                        "token": {"label": "API Token", "type": "password"},
@@ -21,7 +24,6 @@ class PagerDutyBackend(AWXBaseEmailBackend):
     recipient_parameter = "service_key"
     sender_parameter = "client_name"
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
     DEFAULT_BODY = "{{ job_summary_dict }}"
     default_messages = {"started": { "message": DEFAULT_MSG, "body": DEFAULT_BODY},
                         "success": { "message": DEFAULT_MSG, "body": DEFAULT_BODY},

awx/main/notifications/rocketchat_backend.py

@@ -7,22 +7,20 @@ import json
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.rocketchat_backend')
 
 
-class RocketChatBackend(AWXBaseEmailBackend):
+class RocketChatBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"rocketchat_url": {"label": "Target URL", "type": "string"},
                        "rocketchat_no_verify_ssl": {"label": "Verify SSL", "type": "bool"}}
     recipient_parameter = "rocketchat_url"
     sender_parameter = None
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
 
     def __init__(self, rocketchat_no_verify_ssl=False, rocketchat_username=None, rocketchat_icon_url=None, fail_silently=False, **kwargs):
         super(RocketChatBackend, self).__init__(fail_silently=fail_silently)

awx/main/notifications/slack_backend.py

@@ -6,24 +6,21 @@ from slackclient import SlackClient
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.slack_backend')
 WEBSOCKET_TIMEOUT = 30
 
 
-class SlackBackend(AWXBaseEmailBackend):
+class SlackBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"token": {"label": "Token", "type": "password"},
                        "channels": {"label": "Destination Channels", "type": "list"}}
     recipient_parameter = "channels"
     sender_parameter = None
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
-
     def __init__(self, token, hex_color="", fail_silently=False, **kwargs):
         super(SlackBackend, self).__init__(fail_silently=fail_silently)
         self.token = token

awx/main/notifications/twilio_backend.py

@@ -7,12 +7,14 @@ from twilio.rest import Client
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.twilio_backend')
 
 
-class TwilioBackend(AWXBaseEmailBackend):
+class TwilioBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"account_sid": {"label": "Account SID", "type": "string"},
                        "account_token": {"label": "Account Token", "type": "password"},
@@ -21,11 +23,6 @@ class TwilioBackend(AWXBaseEmailBackend):
     recipient_parameter = "to_numbers"
     sender_parameter = "from_number"
 
-    DEFAULT_MSG = "{{ job_friendly_name }} #{{ job.id }} '{{ job.name }}' {{ job.status }}: {{ url }}"
-    default_messages = {"started": {"message": DEFAULT_MSG},
-                        "success": {"message": DEFAULT_MSG},
-                        "error": {"message": DEFAULT_MSG}}
-
     def __init__(self, account_sid, account_token, fail_silently=False, **kwargs):
         super(TwilioBackend, self).__init__(fail_silently=fail_silently)
         self.account_sid = account_sid

awx/main/notifications/webhook_backend.py

@@ -7,13 +7,15 @@ import requests
 
 from django.utils.encoding import smart_text
 from django.utils.translation import ugettext_lazy as _
+
 from awx.main.notifications.base import AWXBaseEmailBackend
 from awx.main.utils import get_awx_version
+from awx.main.notifications.custom_notification_base import CustomNotificationBase
 
 logger = logging.getLogger('awx.main.notifications.webhook_backend')
 
 
-class WebhookBackend(AWXBaseEmailBackend):
+class WebhookBackend(AWXBaseEmailBackend, CustomNotificationBase):
 
     init_parameters = {"url": {"label": "Target URL", "type": "string"},
                        "http_method": {"label": "HTTP Method", "type": "string", "default": "POST"},