wutta/wuttjamaican
3
0
Fork 0
Code Issues Activity

feat: add problem checks + handler feature

Browse source 
the basic idea is to run nightly checks and send email if problems are
found.  it should also support variations on that theme,
e.g. configuring a check to only run on certain weekdays.
This commit is contained in:
Lance Edgar 2025-08-10 11:07:30 -05:00
parent eb6ad9884c
commit 7550a7a860
12 changed files with 900 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

53
tests/cli/test_problems.py Normal file
View file

@ -0,0 +1,53 @@
# -*- coding: utf-8; -*-
from unittest.mock import Mock, patch
from wuttjamaican.testing import ConfigTestCase
from wuttjamaican.cli import problems as mod
from wuttjamaican.problems import ProblemHandler, ProblemCheck
class FakeCheck(ProblemCheck):
system_key = 'wuttatest'
problem_key = 'fake_check'
title = "Fake problem check"
class TestProblems(ConfigTestCase):
def test_basic(self):
ctx = Mock()
ctx.parent.wutta_config = self.config
# nb. avoid printing to console
with patch.object(mod.rich, 'print') as rich_print:
# nb. use fake check
with patch.object(ProblemHandler, 'get_all_problem_checks', return_value=[FakeCheck]):
with patch.object(ProblemHandler, 'run_problem_checks') as run_problem_checks:
# list problem checks
orig_organize = ProblemHandler.organize_problem_checks
def mock_organize(checks):
return orig_organize(None, checks)
with patch.object(ProblemHandler, 'organize_problem_checks', side_effect=mock_organize) as organize_problem_checks:
mod.problems(ctx, list_checks=True)
organize_problem_checks.assert_called_once_with([FakeCheck])
run_problem_checks.assert_not_called()
# warning if unknown system key requested
rich_print.reset_mock()
# nb. just --list for convenience
# note that since we also specify invalid --system, no checks will
# match and hence nothing significant will be printed to stdout
mod.problems(ctx, list_checks=True, systems=['craziness'])
rich_print.assert_called_once()
self.assertEqual(len(rich_print.call_args.args), 1)
self.assertIn("No problem reports exist for system", rich_print.call_args.args[0])
self.assertEqual(len(rich_print.call_args.kwargs), 0)
run_problem_checks.assert_not_called()
# run problem checks
mod.problems(ctx)
run_problem_checks.assert_called_once_with([FakeCheck])

6
tests/test_app.py
View file

@ -634,6 +634,12 @@ app_title = WuttaTest
people = self.app.get_people_handler()
self.assertIsInstance(people, PeopleHandler)
def test_get_problem_handler(self):
from wuttjamaican.problems import ProblemHandler
handler = self.app.get_problem_handler()
self.assertIsInstance(handler, ProblemHandler)
def test_get_report_handler(self):
from wuttjamaican.reports import ReportHandler

270
tests/test_problems.py Normal file
View file

@ -0,0 +1,270 @@
# -*- coding: utf-8; -*-
import datetime
from unittest.mock import patch
from wuttjamaican import problems as mod
from wuttjamaican.testing import ConfigTestCase
class TestProblemCheck(ConfigTestCase):
def make_check(self):
return mod.ProblemCheck(self.config)
def test_system_key(self):
check = self.make_check()
self.assertRaises(AttributeError, getattr, check, 'system_key')
def test_problem_key(self):
check = self.make_check()
self.assertRaises(AttributeError, getattr, check, 'problem_key')
def test_title(self):
check = self.make_check()
self.assertRaises(AttributeError, getattr, check, 'title')
def test_find_problems(self):
check = self.make_check()
problems = check.find_problems()
self.assertEqual(problems, [])
def test_get_email_context(self):
check = self.make_check()
problems = check.find_problems()
context = check.get_email_context(problems)
self.assertEqual(context, {})
def test_make_email_attachments(self):
check = self.make_check()
problems = check.find_problems()
context = check.get_email_context(problems)
attachments = check.make_email_attachments(context)
self.assertIsNone(attachments)
class FakeProblemCheck(mod.ProblemCheck):
system_key = 'wuttatest'
problem_key = 'fake_check'
title = "Fake problem check"
# def find_problems(self):
# return [{'foo': 'bar'}]
class TestProblemHandler(ConfigTestCase):
def setUp(self):
super().setUp()
self.handler = self.make_handler()
def make_handler(self):
return mod.ProblemHandler(self.config)
def test_get_all_problem_checks(self):
# no checks by default
checks = self.handler.get_all_problem_checks()
self.assertIsInstance(checks, list)
self.assertEqual(len(checks), 0)
# but let's configure our fake check
self.config.setdefault('wutta.problems.modules', 'tests.test_problems')
checks = self.handler.get_all_problem_checks()
self.assertIsInstance(checks, list)
self.assertEqual(len(checks), 1)
def test_filtered_problem_checks(self):
# no checks by default
checks = self.handler.filter_problem_checks()
self.assertIsInstance(checks, list)
self.assertEqual(len(checks), 0)
# but let's configure our fake check
self.config.setdefault('wutta.problems.modules', 'tests.test_problems')
checks = self.handler.filter_problem_checks()
self.assertIsInstance(checks, list)
self.assertEqual(len(checks), 1)
# filter by system_key
checks = self.handler.filter_problem_checks(systems=['wuttatest'])
self.assertEqual(len(checks), 1)
checks = self.handler.filter_problem_checks(systems=['something_else'])
self.assertEqual(len(checks), 0)
# filter by problem_key
checks = self.handler.filter_problem_checks(problems=['fake_check'])
self.assertEqual(len(checks), 1)
checks = self.handler.filter_problem_checks(problems=['something_else'])
self.assertEqual(len(checks), 0)
# filter by both
checks = self.handler.filter_problem_checks(systems=['wuttatest'], problems=['fake_check'])
self.assertEqual(len(checks), 1)
checks = self.handler.filter_problem_checks(systems=['wuttatest'], problems=['bad_check'])
self.assertEqual(len(checks), 0)
def test_get_supported_systems(self):
# no checks by default
systems = self.handler.get_supported_systems()
self.assertIsInstance(systems, list)
self.assertEqual(len(systems), 0)
# but let's configure our fake check
self.config.setdefault('wutta.problems.modules', 'tests.test_problems')
systems = self.handler.get_supported_systems()
self.assertIsInstance(systems, list)
self.assertEqual(systems, ['wuttatest'])
def test_get_system_title(self):
title = self.handler.get_system_title('wutta')
self.assertEqual(title, 'wutta')
def test_is_enabled(self):
check = FakeProblemCheck(self.config)
# enabled by default
self.assertTrue(self.handler.is_enabled(check))
# config can disable
self.config.setdefault('wutta.problems.wuttatest.fake_check.enabled', 'false')
self.assertFalse(self.handler.is_enabled(check))
def test_should_run_for_weekday(self):
check = FakeProblemCheck(self.config)
# should run by default
for weekday in range(7):
self.assertTrue(self.handler.should_run_for_weekday(check, weekday))
# config can disable, e.g. for weekends
self.config.setdefault('wutta.problems.wuttatest.fake_check.day5', 'false')
self.config.setdefault('wutta.problems.wuttatest.fake_check.day6', 'false')
for weekday in range(5):
self.assertTrue(self.handler.should_run_for_weekday(check, weekday))
for weekday in (5, 6):
self.assertFalse(self.handler.should_run_for_weekday(check, weekday))
def test_organize_problem_checks(self):
checks = [FakeProblemCheck]
organized = self.handler.organize_problem_checks(checks)
self.assertIsInstance(organized, dict)
self.assertEqual(list(organized), ['wuttatest'])
self.assertIsInstance(organized['wuttatest'], dict)
self.assertEqual(list(organized['wuttatest']), ['fake_check'])
self.assertIs(organized['wuttatest']['fake_check'], FakeProblemCheck)
def test_find_problems(self):
check = FakeProblemCheck(self.config)
problems = self.handler.find_problems(check)
self.assertEqual(problems, [])
def test_get_email_key(self):
check = FakeProblemCheck(self.config)
key = self.handler.get_email_key(check)
self.assertEqual(key, 'wuttatest_problems_fake_check')
def test_get_global_email_context(self):
context = self.handler.get_global_email_context()
self.assertEqual(context, {})
def test_get_check_email_context(self):
check = FakeProblemCheck(self.config)
problems = []
context = self.handler.get_check_email_context(check, problems)
self.assertEqual(context, {'system_title': 'wuttatest'})
def test_send_problem_report(self):
check = FakeProblemCheck(self.config)
problems = []
with patch.object(self.app, 'send_email') as send_email:
self.handler.send_problem_report(check, problems)
send_email.assert_called_once_with('wuttatest_problems_fake_check', {
'system_title': 'wuttatest',
'config': self.config,
'app': self.app,
'check': check,
'problems': problems,
}, default_subject="Fake problem check", attachments=None)
def test_run_problem_check(self):
with patch.object(FakeProblemCheck, 'find_problems') as find_problems:
with patch.object(self.handler, 'send_problem_report') as send_problem_report:
# check runs by default
find_problems.return_value = [{'foo': 'bar'}]
problems = self.handler.run_problem_check(FakeProblemCheck)
self.assertEqual(problems, [{'foo': 'bar'}])
find_problems.assert_called_once_with()
send_problem_report.assert_called_once()
# does not run if generally disabled
find_problems.reset_mock()
send_problem_report.reset_mock()
with patch.object(self.handler, 'is_enabled', return_value=False):
problems = self.handler.run_problem_check(FakeProblemCheck)
self.assertIsNone(problems)
find_problems.assert_not_called()
send_problem_report.assert_not_called()
# unless caller gives force flag
problems = self.handler.run_problem_check(FakeProblemCheck, force=True)
self.assertEqual(problems, [{'foo': 'bar'}])
find_problems.assert_called_once_with()
send_problem_report.assert_called_once()
# does not run if disabled for weekday
find_problems.reset_mock()
send_problem_report.reset_mock()
weekday = datetime.date.today().weekday()
self.config.setdefault(f'wutta.problems.wuttatest.fake_check.day{weekday}', 'false')
problems = self.handler.run_problem_check(FakeProblemCheck)
self.assertIsNone(problems)
find_problems.assert_not_called()
send_problem_report.assert_not_called()
# unless caller gives force flag
problems = self.handler.run_problem_check(FakeProblemCheck, force=True)
self.assertEqual(problems, [{'foo': 'bar'}])
find_problems.assert_called_once_with()
send_problem_report.assert_called_once()
def test_run_problem_checks(self):
with patch.object(FakeProblemCheck, 'find_problems') as find_problems:
with patch.object(self.handler, 'send_problem_report') as send_problem_report:
# check runs by default
find_problems.return_value = [{'foo': 'bar'}]
self.handler.run_problem_checks([FakeProblemCheck])
find_problems.assert_called_once_with()
send_problem_report.assert_called_once()
# does not run if generally disabled
find_problems.reset_mock()
send_problem_report.reset_mock()
with patch.object(self.handler, 'is_enabled', return_value=False):
self.handler.run_problem_checks([FakeProblemCheck])
find_problems.assert_not_called()
send_problem_report.assert_not_called()
# unless caller gives force flag
self.handler.run_problem_checks([FakeProblemCheck], force=True)
find_problems.assert_called_once_with()
send_problem_report.assert_called_once()
# does not run if disabled for weekday
find_problems.reset_mock()
send_problem_report.reset_mock()
weekday = datetime.date.today().weekday()
self.config.setdefault(f'wutta.problems.wuttatest.fake_check.day{weekday}', 'false')
self.handler.run_problem_checks([FakeProblemCheck])
find_problems.assert_not_called()
send_problem_report.assert_not_called()
# unless caller gives force flag
self.handler.run_problem_checks([FakeProblemCheck], force=True)
find_problems.assert_called_once_with()
send_problem_report.assert_called_once()