github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity

Add GitHub Actions workflow_call notification support 

---

For more details, open the [Copilot Workspace session](https://copilot-workspace.githubnext.com/caronc/apprise?shareId=XXXX-XXXX-XXXX-XXXX).
pull/1232/head
Colton English 2024-11-04 10:57:02 -07:00
parent 01c1082ad8
commit 87c8e76818
2 changed files with 381 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

188
apprise/plugins/github_workflow.py Normal file
View File

@ -0,0 +1,188 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import requests
from .base import NotifyBase
from ..common import NotifyType
from ..utils import validate_regex
from ..utils import parse_bool
from ..utils import parse_list
from ..utils import parse_url
from ..utils import unquote
from ..utils import quote
from ..utils import is_exclusive_match
from ..logger import logger
from ..locale import gettext_lazy as _
class NotifyGitHubWorkflow(NotifyBase):
"""
A wrapper for GitHub Actions workflow_call notifications
"""
# The default descriptive name associated with the Notification
service_name = 'GitHub Workflow'
# The services URL
service_url = 'https://github.com/features/actions'
# The default secure protocol
secure_protocol = 'github+workflow'
# The default notify format
notify_format = 'text'
# The maximum allowable characters allowed in the body per message
body_maxlen = 10000
# Define object templates
templates = (
'{schema}://{token}@{repository}/{workflow}',
)
# Define our template tokens
template_tokens = {
'schema': {
'name': _('Schema'),
'type': 'string',
'required': True,
},
'token': {
'name': _('Token'),
'type': 'string',
'private': True,
'required': True,
},
'repository': {
'name': _('Repository'),
'type': 'string',
'required': True,
},
'workflow': {
'name': _('Workflow'),
'type': 'string',
'required': True,
},
}
def __init__(self, token, repository, workflow, **kwargs):
"""
Initialize GitHub Workflow Object
"""
super().__init__(**kwargs)
self.token = token
self.repository = repository
self.workflow = workflow
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform GitHub Workflow Notification
"""
headers = {
'Authorization': f'token {self.token}',
'Accept': 'application/vnd.github.v3+json',
}
payload = {
'ref': 'main',
'inputs': {
'title': title,
'body': body,
}
}
notify_url = f'https://api.github.com/repos/{self.repository}/actions/workflows/{self.workflow}/dispatches'
self.logger.debug('GitHub Workflow POST URL: %s' % notify_url)
self.logger.debug('GitHub Workflow Payload: %s' % str(payload))
# Always call throttle before any remote server i/o is made
self.throttle()
try:
r = requests.post(
notify_url,
json=payload,
headers=headers,
verify=self.verify_certificate,
timeout=self.request_timeout,
)
if r.status_code != requests.codes.no_content:
# We had a problem
status_str = \
NotifyGitHubWorkflow.http_response_code_lookup(r.status_code)
self.logger.warning(
'Failed to send GitHub Workflow notification: '
'{}{}error={}.'.format(
status_str,
', ' if status_str else '',
r.status_code))
self.logger.debug(
'Response Details:\r\n{}'.format(r.content))
# We failed
return False
else:
self.logger.info('Sent GitHub Workflow notification.')
except requests.RequestException as e:
self.logger.warning(
'A Connection error occurred sending GitHub Workflow notification.')
self.logger.debug('Socket Exception: %s' % str(e))
# We failed
return False
return True
@staticmethod
def parse_url(url):
"""
Parses the URL and returns enough arguments that can allow
us to re-instantiate this object.
"""
results = NotifyBase.parse_url(url)
if not results:
# We're done early as we couldn't load the results
return results
# Token
results['token'] = unquote(results['user'])
# Repository
results['repository'] = unquote(results['host'])
# Workflow
results['workflow'] = unquote(results['fullpath'][1:])
return results

193
test/test_plugin_github_workflow.py Normal file
View File

@ -0,0 +1,193 @@
# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
import requests
import pytest
from apprise import Apprise
from apprise import NotifyType
from apprise.plugins.github_workflow import NotifyGitHubWorkflow
from helpers import AppriseURLTester
# Disable logging for a cleaner testing output
import logging
logging.disable(logging.CRITICAL)
# Our Testing URLs
apprise_url_tests = (
##################################
# NotifyGitHubWorkflow
##################################
('github+workflow://', {
# invalid host details (parsing fails very early)
'instance': None,
}),
('github+workflow://:@/', {
# invalid host details (parsing fails very early)
'instance': None,
}),
('github+workflow://token@repository/workflow', {
# All tokens provided - we're good
'instance': NotifyGitHubWorkflow,
}),
('github+workflow://token@repository/workflow', {
'instance': NotifyGitHubWorkflow,
# force a failure
'response': False,
'requests_response_code': requests.codes.internal_server_error,
}),
('github+workflow://token@repository/workflow', {
'instance': NotifyGitHubWorkflow,
# throw a bizzare code forcing us to fail to look it up
'response': False,
'requests_response_code': 999,
}),
('github+workflow://token@repository/workflow', {
'instance': NotifyGitHubWorkflow,
# Throws a series of connection and transfer exceptions when this flag
# is set and tests that we gracfully handle them
'test_requests_exceptions': True,
}),
)
def test_plugin_github_workflow_urls():
"""
NotifyGitHubWorkflow() Apprise URLs
"""
# Run our general tests
AppriseURLTester(tests=apprise_url_tests).run_all()
@pytest.fixture
def github_workflow_url():
return 'github+workflow://token@repository/workflow'
@pytest.fixture
def request_mock(mocker):
"""
Prepare requests mock.
"""
mock_post = mocker.patch("requests.post")
mock_post.return_value = requests.Request()
mock_post.return_value.status_code = requests.codes.no_content
return mock_post
def test_plugin_github_workflow_send_success(request_mock, github_workflow_url):
"""
NotifyGitHubWorkflow() Send - success.
Test cases where URL and JSON is valid.
"""
# Instantiate our URL
obj = Apprise.instantiate(github_workflow_url)
assert isinstance(obj, NotifyGitHubWorkflow)
assert obj.notify(
body="body", title='title',
notify_type=NotifyType.INFO) is True
assert request_mock.called is True
assert request_mock.call_args_list[0][0][0].startswith(
'https://api.github.com/repos/repository/actions/workflows/workflow/dispatches')
# Our Posted JSON Object
posted_json = request_mock.call_args_list[0][1]['json']
assert 'ref' in posted_json
assert posted_json['ref'] == 'main'
assert 'inputs' in posted_json
assert posted_json['inputs']['title'] == 'title'
assert posted_json['inputs']['body'] == 'body'
def test_plugin_github_workflow_send_failure(request_mock, github_workflow_url):
"""
NotifyGitHubWorkflow() Send - failure.
Test cases where URL and JSON is invalid.
"""
# Instantiate our URL
obj = Apprise.instantiate(github_workflow_url)
assert isinstance(obj, NotifyGitHubWorkflow)
# Simulate a failure response
request_mock.return_value.status_code = requests.codes.bad_request
assert obj.notify(
body="body", title='title',
notify_type=NotifyType.INFO) is False
assert request_mock.called is True
assert request_mock.call_args_list[0][0][0].startswith(
'https://api.github.com/repos/repository/actions/workflows/workflow/dispatches')
# Our Posted JSON Object
posted_json = request_mock.call_args_list[0][1]['json']
assert 'ref' in posted_json
assert posted_json['ref'] == 'main'
assert 'inputs' in posted_json
assert posted_json['inputs']['title'] == 'title'
assert posted_json['inputs']['body'] == 'body'
def test_plugin_github_workflow_edge_cases():
"""
NotifyGitHubWorkflow() Edge Cases
"""
# Initializes the plugin with an invalid token
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token='@', repository='repo', workflow='workflow')
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token='', repository='repo', workflow='workflow')
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token=None, repository='repo', workflow='workflow')
# Whitespace also acts as an invalid token value
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token=' ', repository='repo', workflow='workflow')
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token='token', repository=None, workflow='workflow')
# Whitespace also acts as an invalid token value
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token='token', repository=' ', workflow='workflow')
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token='token', repository='repo', workflow=None)
# Whitespace also acts as an invalid token value
with pytest.raises(TypeError):
NotifyGitHubWorkflow(token='token', repository='repo', workflow=' ')
# test case where no tokens are specified
obj = NotifyGitHubWorkflow(token='token', repository='repo', workflow='workflow')
assert isinstance(obj, NotifyGitHubWorkflow)