scott/alertify
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Compare commits

base: scott:main
Branches Tags
scott:main
scott:scottwallacesh/dev
scott:2.0
scott:1.6
scott:1.5
scott:1.4
scott:1.3
scott:1.2
scott:1.1
scott:1.0
..
compare: scott:1.4
Branches Tags
scott:scottwallacesh/dev
scott:main
scott:2.0
scott:1.6
scott:1.5
scott:1.4
scott:1.3
scott:1.2
scott:1.1
scott:1.0

No commits in common. "main" and "1.4" have entirely different histories.
main ... 1.4

19 changed files with 310 additions and 821 deletions
Show stats Expand all files Collapse all files

39
.github/workflows/python-package.yml vendored
View file

@ -1,39 +0,0 @@
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
name: Python package
on:
push:
branches: [ main, dev/* ]
pull_request:
branches: [ main, dev/* ]
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: [3.6, 3.7, 3.8]
steps:
- uses: actions/checkout@v2
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install flake8 pytest
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest
run: |
pytest

3
.gitignore vendored
View file

@ -1,3 +1,2 @@
__pycache__/
.pyenv/
.pytest_cache/
__pycache__/

8
Dockerfile
View file

@ -1,3 +1,4 @@
# For more information, please refer to https://aka.ms/vscode-docker-python
FROM python:3.8-slim-buster
# Keeps Python from generating .pyc files in the container
@ -6,18 +7,21 @@ ENV PYTHONDONTWRITEBYTECODE 1
# Turns off buffering for easier container logging
ENV PYTHONUNBUFFERED 1
# Install pip requirements
ADD requirements.txt .
RUN python -m pip install -r requirements.txt
WORKDIR /app
COPY alertify.py /app
COPY src /app/src
ADD src/alertify.py /app
ADD src/gotify.py /app
# Switching to a non-root user, please refer to https://aka.ms/vscode-docker-python-user-rights
RUN useradd appuser && chown -R appuser /app
USER appuser
EXPOSE 8080
# During debugging, this entry point will be overridden. For more information, please refer to https://aka.ms/vscode-docker-python-debug
CMD ["python", "alertify.py"]
HEALTHCHECK --interval=30s --timeout=3s --retries=1 \

30
README.md
View file

@ -13,26 +13,26 @@ optional arguments:
-H, --healthcheck simply exit with 0 for healthy or 1 when unhealthy
The following environment variables will override any config or default:
* DELETE_ONRESOLVE (default: False)
* DISABLE_RESOLVED (default: False)
* GOTIFY_KEY_APP (default: None)
* GOTIFY_KEY_CLIENT (default: None)
* GOTIFY_URL_PREFIX (default: http://localhost)
* LISTEN_PORT (default: 8080)
* VERBOSE (default: 0)
* DELETE_ONRESOLVE (default: False)
* DISABLE_RESOLVED (default: False)
* GOTIFY_CLIENT (default: None)
* GOTIFY_KEY (default: None)
* GOTIFY_PORT (default: 80)
* GOTIFY_SERVER (default: localhost)
* LISTEN_PORT (default: 8080)
* VERBOSE (default: False)
```
# Notes
* Requires Python v3.6 or greater (f-strings are used)
* Listens on port 8080 by default.
* Forwards `resolved` alerts, if not disabled.
* Resolved alerts delete the original alert, if enabled.
* Requires a Gotify app key to send alerts to Gotify
* Requires a Gotify client key to delete original alert on resolution
* Defaults, if not sent:
| Field | Default value |
|-------------|---------------|
| Description | `[nodata]` |
| Instance | `[unknown]` |
| Priority | `5` |
| Severity | `Warning` |
@ -47,7 +47,7 @@ docker build . -t 'alertify:latest'
e.g.
```bash
docker run --name alertify -p 8080:8080 -e TZ=Europe/London -e GOTIFY_KEY_APP=_APPKEY_ -e GOTIFY_URL_PREFIX=http://gotify alertify:latest
docker run --name alertify -p 8080:8080 -e TZ=Europe/London -e GOTIFY_KEY=_APPKEY_ -e GOTIFY_SERVER=gotify -e GOTIFY_PORT=80 alertify:latest
```
## Compose:
@ -72,9 +72,9 @@ services:
- "8080:8080"
environment:
- TZ=Europe/London
- DELETE_ONRESOLVE=true
- GOTIFY_KEY_APP=_APPKEY_
- GOTIFY_KEY_CLIENT=_CLIENTKEY_
- GOTIFY_URL_PREFIX=http://gotify
- GOTIFY_KEY=_APPKEY_
- GOTIFY_CLIENT=_CLIENTKEY_
- GOTIFY_SERVER=gotify
- GOTIFY_PORT=80
restart: unless-stopped
```

87
alertify.py
View file

@ -1,87 +0,0 @@
#!/usr/bin/python3
"""
Main entrypoint to run Alertify
"""
import argparse
import logging
import os
import sys
from src import Alertify
if __name__ == '__main__':
def parse_cli() -> argparse.ArgumentParser:
"""
Function to parse the CLI
"""
maxlen = max([len(key) for key in Alertify.Config.defaults()])
defaults = [
f' * {key.upper().ljust(maxlen)} (default: {val if val != "" else "None"})'
for key, val in Alertify.Config.defaults().items()
]
parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description='Bridge between Prometheus Alertmanager and Gotify\n',
epilog='The following environment variables will override any config or default:\n'
+ '\n'.join(defaults),
)
parser.add_argument(
'-c',
'--config',
default=f'{os.path.splitext(__file__)[0]}.yaml',
help=f'path to config YAML. (default: {os.path.splitext(__file__)[0]}.yaml)',
)
parser.add_argument(
'-H',
'--healthcheck',
action='store_true',
help='simply exit with 0 for healthy or 1 when unhealthy',
)
return parser.parse_args()
def main() -> int:
"""
Main program logic
"""
logging.basicConfig(
format='%(levelname)s: %(message)s',
level=logging.INFO,
)
args = parse_cli()
alertify = Alertify.Alertify()
alertify.configure(args.config)
# -----------------------------
# Calculate logging level
# -----------------------------
# Config :: Verbose: 0 = WARNING, 1 = INFO, 2 = DEBUG
# Logging :: Loglevel: 30 = WARNING, 20 = INFO, 10 = DEBUG
logger = logging.getLogger()
logger.setLevel(max(logging.WARNING - (alertify.config.verbose * 10), 10))
# -----------------------------
if args.healthcheck:
_, status = alertify.healthcheck()
# Invert the sense of 'healthy' for Unix CLI usage
return not status == 200
logging.info('Version: %s', Alertify.__version__)
if alertify.config.verbose:
logging.debug('Parsed config:')
for key, val in alertify.config.items():
logging.debug('%s: %s', key, val)
alertify.run()
return 0
sys.exit(main())

3
example.yaml
View file

@ -1,6 +1,7 @@
---
disable_resolved: false
gotify_key: sOmEsEcReTkEy1
gotify_url_prefix: http://gotifyserver.example.net
gotify_port: "80"
gotify_server: gotifyserver.example.net
listen_port: "8080"
verbose: true

4
requirements.txt
View file

@ -1,4 +1,2 @@
flask
flask-classful
# To ensure app dependencies are ported from your virtual environment/host machine into your container, run 'pip freeze > requirements.txt' in the terminal to overwrite this file
pyyaml
requests

100
src/Alertify/__init__.py
View file

@ -1,100 +0,0 @@
"""
Alertify module to act as a bridge between Prometheus Alertmanager and Gotify
"""
# pylint: disable=invalid-name
__author__ = 'Scott Wallace'
__email__ = 'scott@wallace.sh'
__maintainer__ = 'Scott Wallace'
__version__ = '2.0'
import json
import logging
from typing import Optional, Tuple
import werkzeug.exceptions
from flask import Flask, request, request_started
from flask_classful import FlaskView, route
from .config import Config
from .gotify import Gotify
from .health import Healthcheck
from .messaging import MessageHandler
webapp = Flask(__name__)
# FIXME: * Find a better way to deny FlaskView methods without using a `_`
# prefix or raising a NotFound exception
class Alertify(FlaskView):
"""
Main Alertify class
"""
route_base = '/'
trailing_slash = False
def __init__(self):
# Instantiate with defaults
self.configure()
def configure(self, configfile: Optional[str] = None):
"""
Configure from a configfile
"""
# Deny via HTTP
if request:
raise werkzeug.exceptions.NotFound
self.config = Config(configfile)
self.gotify = Gotify(
self.config.gotify_url_prefix,
self.config.gotify_key_app,
self.config.gotify_key_client,
)
self.msg_hndlr = MessageHandler(
self.gotify,
self.config.disable_resolved,
self.config.delete_onresolve,
)
def run(self):
"""
Listen on port and run webserver
"""
# Deny via HTTP
if request:
raise werkzeug.exceptions.NotFound
webapp.run(host='0.0.0.0', port=self.config.listen_port)
@route('/alert', methods=['POST'])
def alert(self) -> Tuple[str, int]:
"""
Handle the alerts from Alertmanager
"""
message = request.get_json()
logging.debug(
'Received from Alertmanager:\n%s',
json.dumps(message, indent=2),
)
for alertmsg in message['alerts']:
response = self.msg_hndlr.process(alertmsg)
try:
return response['reason'], response['status']
except UnboundLocalError:
return '', 204
def healthcheck(self) -> Tuple[str, int]:
"""
Perform a healthcheck and return the results
"""
response = Healthcheck(self.gotify).gotify_alive()
return response['reason'], response['status']
Alertify.register(webapp)

81
src/Alertify/config.py
View file

@ -1,81 +0,0 @@
"""
Module to handle Alertify's configuration
"""
import inspect
import logging
import os
from distutils.util import strtobool
from typing import Optional
import yaml
class Config:
"""
Class to handle the config
"""
delete_onresolve = bool(False)
disable_resolved = bool(False)
gotify_key_app = str()
gotify_key_client = str()
gotify_url_prefix = str('http://localhost')
listen_port = int(8080)
verbose = int(0)
def __init__(self, configfile: Optional[str] = None):
"""
Method to parse a configuration file
"""
logging.debug('Parsing config')
parsed = {}
try:
with open(configfile, 'r') as file:
parsed = yaml.safe_load(file.read())
except FileNotFoundError as error:
logging.warning('No config file found (%s)', error.filename)
except TypeError:
logging.warning('No config file provided.')
# Iterate over the config defaults and check for environment variable
# overrides, then check for any items in the config, otherwise
# use the default values.
for key, default_val in self.defaults().items():
userval = os.environ.get(key.upper(), parsed.get(key, default_val))
# Ensure the types are adhered to.
if isinstance(default_val, bool):
setattr(self, key, strtobool(str(userval)))
else:
setattr(self, key, type(default_val)(userval))
def items(self) -> list:
"""
Method to return an iterator for the configured items
"""
return {key: getattr(self, key) for key in self.__dict__}.items()
@classmethod
def keys(cls) -> list:
"""
Method to return the defaults as a list of dict_keys
"""
return [
attr[0]
for attr in inspect.getmembers(cls)
if not attr[0].startswith('_')
and not any(
[
inspect.ismethod(attr[1]),
callable(attr[1]),
]
)
]
@classmethod
def defaults(cls) -> dict:
"""
Classmethod to return the defaults as a dictionary
"""
return {key: getattr(cls, key) for key in cls.keys()}

21
src/Alertify/health.py
View file

@ -1,21 +0,0 @@
"""
Module for handling any healthcheck related activity
"""
from typing import Tuple
from .gotify import Gotify
class Healthcheck:
"""
Class to handle the healthchecks
"""
def __init__(self, gotify_client: Gotify):
self.gotify = gotify_client
def gotify_alive(self) -> Tuple[str, int]:
"""
Simple method to return the Gotify healthcheck response
"""
return self.gotify.healthcheck()

76
src/Alertify/messaging.py
View file

@ -1,76 +0,0 @@
"""
Module for handling the messaging
"""
import logging
from typing import Optional
from .gotify import Gotify
class MessageHandler:
"""
Class to handle alert messaging
"""
def __init__(
self,
gotify_client: Gotify,
disable_resolved: Optional[bool] = False,
delete_onresolve: Optional[bool] = False,
):
self.gotify = gotify_client
self.disable_resolved = disable_resolved
self.delete_onresolve = delete_onresolve
def process(self, alert: dict) -> dict:
"""
Method to process the alert message
"""
try:
if alert['status'] == 'resolved':
if self.disable_resolved:
logging.info('Ignoring resolved messages')
return {
'status': 200,
'reason': 'Ignored. "resolved" messages are disabled',
}
if self.delete_onresolve:
for alert_id in self.gotify.find_byfingerprint(alert):
if not self.gotify.delete(alert_id):
logging.error(
'There was a problem removing message ID %d',
alert_id,
)
return {'status': 200, 'reason': 'Message deletion complete'}
prefix = 'resolved'
else:
prefix = alert['labels'].get('severity', 'warning')
instance = alert['labels'].get('instance', None)
gotify_msg = {
'title': '[{}] {}'.format(
prefix.upper(),
alert['annotations'].get('summary'),
),
'message': '{}{}'.format(
f'{instance}: ' if instance else '',
alert['annotations'].get('description', ''),
),
'priority': int(alert['labels'].get('priority', 5)),
'extras': {
'alertify': {
'fingerprint': alert.get('fingerprint', None),
}
},
}
except KeyError as error:
logging.error('KeyError: %s', error)
return {
'status': 400,
'reason': f'Missing field: {error}',
}
return self.gotify.send_alert(gotify_msg)

256
src/alertify.py Normal file
View file

@ -0,0 +1,256 @@
#!/usr/bin/env python3
"""
Module to act as a bridge between Prometheus Alertmanager and Gotify
"""
import argparse
import functools
import json
import logging
import os
import sys
from distutils.util import strtobool
from http.server import HTTPServer, SimpleHTTPRequestHandler
import yaml
import gotify
DEFAULTS = {
'delete_onresolve': bool(False),
'disable_resolved': bool(False),
'gotify_client': str(),
'gotify_key': str(),
'gotify_port': int(80),
'gotify_server': str('localhost'),
'listen_port': int(8080),
'verbose': bool(False),
}
class HTTPHandler(SimpleHTTPRequestHandler):
"""
Class to handle the HTTP requests from a client
"""
config = None
def _alerts(self):
"""
Method to handle the request for alerts
"""
if not healthy(self.config):
logging.error('Check requirements')
self._respond(500, 'Server not configured correctly')
return
content_length = int(self.headers['Content-Length'])
rawdata = self.rfile.read(content_length)
try:
am_msg = json.loads(rawdata.decode())
except json.decoder.JSONDecodeError as error:
logging.error('Bad JSON: %s', error)
self._respond(400, f'Bad JSON: {error}')
return
logging.debug('Received from Alertmanager:\n%s',
json.dumps(am_msg, indent=2))
gotify_client = gotify.Gotify(
self.config.get('gotify_server'),
self.config.get('gotify_port'),
self.config.get('gotify_key'),
self.config.get('gotify_client')
)
for alert in am_msg['alerts']:
try:
if alert['status'] == 'resolved':
if self.config.get('disable_resolved'):
logging.info('Ignoring resolved messages')
self._respond(
200, 'Ignored. "resolved" messages are disabled')
continue
if self.config.get('delete_onresolve'):
alert_id = gotify_client.find_byfingerprint(alert)
if alert_id:
response = gotify_client.delete(alert_id)
continue
logging.debug(
'Could not find a matching message to delete.')
prefix = 'Resolved'
else:
prefix = alert['labels'].get(
'severity', 'warning').capitalize()
gotify_msg = {
'title': '{}: {}'.format(
prefix,
alert['annotations'].get('summary'),
),
'message': '{}: {}'.format(
alert['labels'].get('instance', '[unknown]'),
alert['annotations'].get('description', '[nodata]'),
),
'priority': int(alert['labels'].get('priority', 5)),
'extras': {
'alertify': {
'fingerprint': alert.get('fingerprint', None)
}
}
}
except KeyError as error:
logging.error('KeyError: %s', error)
self._respond(400, f'Missing field: {error}')
return
response = gotify_client.send_alert(gotify_msg)
try:
self._respond(response['status'], response['reason'])
except UnboundLocalError:
self._respond('204', '')
def _respond(self, status, message):
"""
Method to output a simple HTTP status and string to the client
"""
self.send_response(int(status) or 500)
self.end_headers()
self.wfile.write(bytes(str(message).encode()))
# Override built-in method
def do_GET(self): # pylint: disable=invalid-name
"""
Method to handle GET requests
"""
if self.path == '/healthcheck':
if not healthy(self.config):
logging.error('Check requirements')
self._respond(500, 'ERR')
self._respond(200, 'OK')
# Override built-in method
def do_POST(self): # pylint: disable=invalid-name
"""
Method to handle POST requests from AlertManager
"""
if self.path == '/alert':
self._alerts()
# FIXME: This isn't right. A normal method doesn't work, however.
@classmethod
def set_config(cls, config):
"""
Classmethod to add config to the class
"""
cls.config = config
def healthy(config):
"""
Simple function to return if all the requirements are met
"""
return all([
len(config.get('gotify_key', ''))
])
@functools.lru_cache
def parse_config(configfile):
"""
Function to parse a configuration file
"""
config = {}
try:
with open(configfile, 'r') as file:
parsed = yaml.safe_load(file.read())
except FileNotFoundError as error:
logging.warning('No config file found (%s)', error.filename)
parsed = {}
# Iterate over the DEFAULTS dictionary and check for environment variables
# of the same name, then check for any items in the YAML config, otherwise
# use the default values.
# Ensure the types are adhered to.
for key, val in DEFAULTS.items():
config[key] = os.environ.get(key.upper(), parsed.get(key, val))
if isinstance(val, bool):
config[key] = strtobool(str(config[key]))
else:
config[key] = type(val)(config[key])
return config
if __name__ == '__main__':
def parse_cli():
"""
Function to parse the CLI
"""
maxlen = max([len(key) for key in DEFAULTS])
defaults = [
f' * {key.upper().ljust(maxlen)} (default: {val if val != "" else "None"})'
for key, val in DEFAULTS.items()
]
parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description='Bridge between Prometheus Alertmanager and Gotify\n',
epilog='The following environment variables will override any config or default:\n' +
'\n'.join(defaults)
)
parser.add_argument(
'-c', '--config',
default=f'{os.path.splitext(__file__)[0]}.yaml',
help=f'path to config YAML. (default: {os.path.splitext(__file__)[0]}.yaml)',
)
parser.add_argument(
'-H', '--healthcheck',
action='store_true',
help='simply exit with 0 for healthy or 1 when unhealthy',
)
return parser.parse_args()
def main():
"""
main()
"""
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
args = parse_cli()
config = parse_config(args.config)
if config.get('verbose'):
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
if args.healthcheck:
# Invert the sense of 'healthy' for Unix CLI usage
return not healthy(config)
listen_port = config.get('listen_port')
logging.debug(
'Config:\n%s',
yaml.dump(config, explicit_start=True, default_flow_style=False)
)
logging.info('Starting web server on port %d', listen_port)
try:
with HTTPServer(('', listen_port), HTTPHandler) as webserver:
HTTPHandler.set_config(config)
webserver.serve_forever()
except KeyboardInterrupt:
logging.info('Exiting')
return 0
sys.exit(main())

0
src/conftest.py
View file

74
src/Alertify/gotify.py → src/gotify.py
View file

@ -1,12 +1,9 @@
"""
Module to handle communication with the Gotify server
"""
import http.client
import json
import logging
import socket
from typing import Optional
from requests import request
class Gotify:
@ -14,8 +11,8 @@ class Gotify:
Class to handle Gotify communications
"""
def __init__(self, url_prefix: str, app_key: str, client_key: Optional[str] = None):
self.url_prefix = url_prefix
def __init__(self, server, port, app_key, client_key=None):
self.api = http.client.HTTPConnection(server, port)
self.app_key = app_key
self.client_key = client_key
self.base_headers = {
@ -23,7 +20,7 @@ class Gotify:
'Accept': 'application/json',
}
def _call(self, method: str, url: str, data: Optional[object] = None) -> dict:
def _call(self, method, url, body=None):
"""
Method to call Gotify with an app or client key as appropriate
"""
@ -33,46 +30,43 @@ class Gotify:
else:
headers['X-Gotify-Key'] = self.app_key
logging.debug('Sending to Gotify:\n%s', data)
logging.debug('Sending to Gotify:\n%s', body)
try:
response = request(
method,
f'{self.url_prefix}{url}',
json=data,
headers=headers,
)
except (ConnectionRefusedError, socket.gaierror) as error:
logging.error('Connection error: %s', error)
self.api.request(method, url, body=body, headers=headers)
response = self.api.getresponse()
except ConnectionRefusedError as error:
logging.error(error)
return {
'status': error.errno,
'reason': error.strerror,
'reason': error.strerror
}
resp_obj = {
'status': response.status_code,
'status': response.status,
'reason': response.reason,
'json': None,
'json': None
}
if len(response.content) > 0:
rawbody = response.read()
if len(rawbody) > 0:
try:
resp_obj['json'] = json.loads(response.content.decode())
resp_obj['json'] = json.loads(rawbody.decode())
except json.decoder.JSONDecodeError as error:
logging.error('Could not parse JSON: %s', error)
logging.error(error)
logging.debug('Returned from Gotify:\n%s', json.dumps(resp_obj, indent=2))
logging.debug('Status: %s, Reason: %s', resp_obj['status'], resp_obj['reason'])
return resp_obj
def delete(self, msg_id: str) -> dict:
def delete(self, msg_id):
"""
Method to delete a message from the Gotify server
"""
logging.debug('Deleting message ID: %s', msg_id)
return self._call('DELETE', f'/message/{msg_id}')
def find_byfingerprint(self, message: str) -> list:
def find_byfingerprint(self, message):
"""
Method to return the ID of a matching message
"""
@ -80,43 +74,33 @@ class Gotify:
new_fingerprint = message['fingerprint']
except KeyError:
logging.debug('No fingerprint found in new message')
return list()
return None
msg_list = []
for old_message in self.messages():
try:
old_fingerprint = old_message['extras']['alertify']['fingerprint']
if old_fingerprint == new_fingerprint:
msg_list.append(old_message['id'])
return old_message['id']
except KeyError:
logging.warning(
'No fingerprint found in message ID: %s',
old_message['id'],
)
logging.debug('No fingerprint found in message ID: %s', old_message['id'])
continue
return msg_list
logging.debug('No fingerprint matched.')
return None
def messages(self) -> dict:
def messages(self):
"""
Method to return a list of messages from the Gotify server
"""
if not self.client_key:
logging.warning(
'No client key is configured. No messages could be retrieved.'
)
return dict()
logging.debug('No client key is configured. No messages could be retrieved.')
return []
logging.debug('Fetching existing messages from Gotify')
return self._call('GET', '/message')['json'].get('messages', [])
def send_alert(self, payload: dict) -> dict:
def send_alert(self, payload):
"""
Method to send a message payload to a Gotify server
"""
logging.debug('Sending message to Gotify')
return self._call('POST', '/message', data=payload)
def healthcheck(self) -> dict:
"""
Method to perform a healthcheck against Gotify
"""
return self._call('GET', '/health')
return self._call('POST', '/message', body=json.dumps(payload, indent=2))

82
src/tests/Alertify/test___init__.py
View file

@ -1,82 +0,0 @@
"""Test"""
import unittest
from unittest.mock import patch
import flask
import Alertify # pylint: disable=import-error
class AlertifyTest(unittest.TestCase):
"""
Tests for methods in the Alertify class.
"""
@classmethod
def setUpClass(cls):
cls.alertify = Alertify.Alertify()
@classmethod
def tearDownClass(cls):
pass
def setUp(self):
pass
def tearDown(self):
pass
def test_configure(self):
"""Test"""
self.alertify.configure(None)
self.assertDictEqual(
self.alertify.config.defaults(),
Alertify.Config.defaults(),
)
@patch('Alertify.messaging.MessageHandler.process')
def test_alert(self, mock_process):
"""Test"""
mock_process.return_value = {
'status': 200,
'reason': 'OK',
'json': None,
}
with flask.Flask(__name__).test_request_context(
'/alert',
data='{"alerts": []}',
headers={'Content-type': 'application/json'},
):
self.assertTupleEqual(
self.alertify.alert(),
('', 204),
)
@patch('Alertify.health.Healthcheck.gotify_alive')
def test_healthcheck(self, mock_gotify_alive):
"""Test"""
mock_gotify_alive.return_value = {
'status': 200,
'reason': 'OK',
'json': None,
}
self.assertEqual(
self.alertify.healthcheck(),
('OK', 200),
)
@patch('Alertify.health.Healthcheck.gotify_alive')
def test_bad_healthcheck(self, mock_gotify_alive):
"""Test"""
mock_gotify_alive.return_value = {
'status': 408,
'reason': 'Request Timeout',
'json': None,
}
self.assertNotEqual(
self.alertify.healthcheck(),
('OK', 200),
)

53
src/tests/Alertify/test_config.py
View file

@ -1,53 +0,0 @@
"""Test"""
import unittest
from Alertify import config # pylint: disable=import-error
class ConfigTest(unittest.TestCase):
"""
Tests for methods in the Config class.
"""
@classmethod
def setUpClass(cls):
cls.defaults = {
'delete_onresolve': bool(False),
'disable_resolved': bool(False),
'gotify_key_app': str(),
'gotify_key_client': str(),
'gotify_url_prefix': str('http://localhost'),
'listen_port': int(8080),
'verbose': int(0),
}
@classmethod
def tearDownClass(cls):
pass
def setUp(self):
pass
def tearDown(self):
pass
def test_items(self):
"""Test"""
self.assertEqual(
config.Config().items(),
self.defaults.items(),
)
def test_keys(self):
"""Test"""
self.assertListEqual(
config.Config.keys(),
list(self.defaults.keys()),
)
def test_defaults(self):
"""Test"""
self.assertDictEqual(
config.Config.defaults(),
self.defaults,
)

120
src/tests/Alertify/test_gotify.py
View file

@ -1,120 +0,0 @@
"""
Module to handle unit tests for the Alertify.gotify module
"""
import unittest
from unittest.mock import patch
from Alertify import gotify # pylint: disable=import-error
class GotifyTest(unittest.TestCase):
"""
Tests for methods in the Gotify class.
"""
@classmethod
def setUpClass(cls):
cls.gotify_client = gotify.Gotify('http://localhost', '')
@classmethod
def tearDownClass(cls):
pass
def setUp(self):
pass
def tearDown(self):
pass
@patch('requests.Session.request')
def test_delete(self, mock_request):
"""Test"""
mock_request.return_value.status_code = 200
mock_request.return_value.reason = 'OK'
mock_request.return_value.content = ''
self.assertDictEqual(
self.gotify_client.delete('123'),
{
'status': 200,
'reason': 'OK',
'json': None,
},
)
@patch('Alertify.gotify.Gotify.messages')
def test_find_byfingerprint(self, mock_messages):
"""Test"""
mock_messages.return_value = [
{
'id': 42,
'extras': {'alertify': {'fingerprint': 'deadbeefcafebabe'}},
}
]
self.assertListEqual(
self.gotify_client.find_byfingerprint({'fingerprint': 'deadbeefcafebabe'}),
[42],
)
def test_messages(self):
"""Test"""
self.assertDictEqual(
self.gotify_client.messages(),
dict(),
)
@patch('requests.Session.request')
def test_send_alert_empty(self, mock_request):
"""Test"""
mock_request.return_value.status_code = 200
mock_request.return_value.reason = 'OK'
mock_request.return_value.content = ''
self.assertDictEqual(
self.gotify_client.send_alert(dict()),
{
'status': 200,
'reason': 'OK',
'json': None,
},
)
@patch('requests.Session.request')
def test_send_alert_dummy(self, mock_request):
"""Test"""
mock_request.return_value.status_code = 200
mock_request.return_value.reason = 'OK'
mock_request.return_value.content = ''
self.assertDictEqual(
self.gotify_client.send_alert(
{
'title': 'TITLE',
'message': 'MESSAGE',
'priority': 0,
'extras': dict(),
}
),
{
'status': 200,
'reason': 'OK',
'json': None,
},
)
@patch('requests.Session.request')
def test_healthcheck(self, mock_request):
"""Test"""
mock_request.return_value.status_code = 200
mock_request.return_value.reason = 'OK'
mock_request.return_value.content = ''
self.assertDictEqual(
self.gotify_client.healthcheck(),
{
'status': 200,
'reason': 'OK',
'json': None,
},
)

43
src/tests/Alertify/test_health.py
View file

@ -1,43 +0,0 @@
"""Test"""
import unittest
from unittest.mock import patch
from Alertify import gotify, health # pylint: disable=import-error
class HealthcheckTest(unittest.TestCase):
"""
Tests for methods in the Healthcheck class.
"""
@classmethod
def setUpClass(cls):
cls.healthcheck = health.Healthcheck(gotify.Gotify('http://localhost', '', ''))
@classmethod
def tearDownClass(cls):
pass
def setUp(self):
pass
def tearDown(self):
pass
@patch('Alertify.health.Healthcheck.gotify_alive')
def test_gotify_alive(self, mock_gotify_alive):
"""Test"""
mock_gotify_alive.return_value = {
'status': 200,
'reason': 'OK',
'json': None,
}
self.assertDictEqual(
self.healthcheck.gotify_alive(),
{
'status': 200,
'reason': 'OK',
'json': None,
},
)

51
src/tests/Alertify/test_messaging.py
View file

@ -1,51 +0,0 @@
"""Test"""
import unittest
from unittest.mock import patch
from Alertify import gotify, messaging # pylint: disable=import-error
class MessageHandlerTest(unittest.TestCase):
"""
Tests for methods in the MessageHandler class.
"""
@classmethod
def setUpClass(cls):
cls.messaging = messaging.MessageHandler(
gotify.Gotify('http://localhost', '', '')
)
@classmethod
def tearDownClass(cls):
pass
def setUp(self):
pass
def tearDown(self):
pass
@patch('Alertify.gotify.Gotify.send_alert')
def test_process(self, mock_send_alert):
"""Test"""
mock_send_alert.return_value = {
'status': 200,
'reason': 'OK',
'json': None,
}
self.assertDictEqual(
self.messaging.process(
{
'status': 'firing',
'labels': {},
'annotations': {},
}
),
{
'status': 200,
'reason': 'OK',
'json': None,
},
)