External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-11 10:00:01 -03:30
Code Issues Packages Projects Releases Wiki Activity

Refactor notification backends to use CustomNotificationBase

Browse Source
This commit is contained in:
Jim Ladd 2019-10-15 17:30:56 -07:00 committed by Ryan Petrello
parent 6cd6a42e20
commit 4e9ec271c5
No known key found for this signature in database
GPG Key ID: F2AA5F2122351777
11 changed files with 49 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
awx/main/notifications/custom_notification_base.py Normal file
View File

@ -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}}

16
awx/main/notifications/email_backend.py
View File

@ -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)

9
awx/main/notifications/grafana_backend.py
View File

@ -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)

9
awx/main/notifications/hipchat_backend.py
View File

@ -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

9
awx/main/notifications/irc_backend.py
View File

@ -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

9
awx/main/notifications/mattermost_backend.py
View File

@ -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)

6
awx/main/notifications/pagerduty_backend.py
View File

@ -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},

8
awx/main/notifications/rocketchat_backend.py
View File

@ -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)

9
awx/main/notifications/slack_backend.py
View File

@ -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

9
awx/main/notifications/twilio_backend.py
View File

@ -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

4
awx/main/notifications/webhook_backend.py
View File

@ -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"},