Add a simple integration test

This commit is contained in:
IamTheFij 2018-07-15 18:30:13 -07:00
parent 0c0a0d9085
commit c7081a0d32
2 changed files with 50 additions and 26 deletions

View File

@ -299,7 +299,7 @@ class Minitor(object):
self._monitor_counter = None self._monitor_counter = None
self._monitor_status_gauge = None self._monitor_status_gauge = None
def _parse_args(self): def _parse_args(self, args=None):
"""Parses command line arguments and returns them""" """Parses command line arguments and returns them"""
parser = ArgumentParser(description='Minimal monitoring') parser = ArgumentParser(description='Minimal monitoring')
parser.add_argument( parser.add_argument(
@ -321,7 +321,7 @@ class Minitor(object):
default=DEFAULT_METRICS_PORT, default=DEFAULT_METRICS_PORT,
help='Port to use when serving metrics', help='Port to use when serving metrics',
) )
return parser.parse_args() return parser.parse_args(args)
def _setup(self, config_path): def _setup(self, config_path):
"""Load all setup from YAML file at provided path""" """Load all setup from YAML file at provided path"""
@ -381,29 +381,31 @@ class Minitor(object):
) )
def _loop(self): def _loop(self):
"""The main run loop"""
while True: while True:
for monitor in self.monitors: self._check()
try:
result = monitor.check()
if result is not None:
self._logger.info(
'%s: %s',
monitor.name,
'SUCCESS' if result else 'FAILURE'
)
except MinitorAlert as minitor_alert:
self._logger.warning(minitor_alert)
self._handle_minitor_alert(minitor_alert)
# Track the status of the Monitor
if self._monitor_status_gauge:
self._monitor_status_gauge.labels(
monitor=monitor.name,
).set(int(monitor.is_up()))
sleep(self.check_interval) sleep(self.check_interval)
def _check(self):
"""The main run loop"""
for monitor in self.monitors:
try:
result = monitor.check()
if result is not None:
self._logger.info(
'%s: %s',
monitor.name,
'SUCCESS' if result else 'FAILURE'
)
except MinitorAlert as minitor_alert:
self._logger.warning(minitor_alert)
self._handle_minitor_alert(minitor_alert)
# Track the status of the Monitor
if self._monitor_status_gauge:
self._monitor_status_gauge.labels(
monitor=monitor.name,
).set(int(monitor.is_up()))
def _handle_minitor_alert(self, minitor_alert): def _handle_minitor_alert(self, minitor_alert):
"""Issues all alerts for a provided monitor""" """Issues all alerts for a provided monitor"""
monitor = minitor_alert.monitor monitor = minitor_alert.monitor
@ -411,9 +413,9 @@ class Minitor(object):
for alert in alerts: for alert in alerts:
self.alerts[alert].alert(str(minitor_alert), monitor) self.alerts[alert].alert(str(minitor_alert), monitor)
def run(self): def run(self, args=None):
"""Runs Minitor in a loop""" """Runs Minitor in a loop"""
args = self._parse_args() args = self._parse_args(args)
self._setup(args.config_path) self._setup(args.config_path)
self._validate_monitors() self._validate_monitors()
@ -424,9 +426,9 @@ class Minitor(object):
self._loop() self._loop()
def main(): def main(args=None):
try: try:
Minitor().run() Minitor().run(args)
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass
return 0 return 0

View File

@ -1,4 +1,8 @@
import os
from unittest.mock import patch
from minitor.main import call_output from minitor.main import call_output
from minitor.main import Minitor
class TestMinitor(object): class TestMinitor(object):
@ -12,3 +16,21 @@ class TestMinitor(object):
output, ex = call_output(['ls', '--not-a-real-flag']) output, ex = call_output(['ls', '--not-a-real-flag'])
assert output.startswith(b'ls: ') assert output.startswith(b'ls: ')
assert ex is not None assert ex is not None
def test_run(self):
"""Doesn't really check much, but a simple integration sanity test"""
test_loop_count = 5
os.environ.update({
'MAILGUN_API_KEY': 'test-mg-key',
'AVAILABLE_NUMBER': '555-555-5050',
'MY_PHONE': '555-555-0505',
'ACCOUNT_SID': 'test-account-id',
'ACCOUNT_TOKEN': 'test-account-token',
})
args = '--config ./sample-config.yml'.split(' ')
minitor = Minitor()
with patch.object(minitor, '_loop'):
minitor.run(args)
# Skip the loop, but run a single check
for _ in range(test_loop_count):
minitor._check()