+ A scheduled poll checks Zabbix toner levels. A warning email fires when + a toner crosses to the warning threshold or below, a critical email at + the critical threshold. Alerts fire once per crossing; a refill re-arms + them. +
+ +No users loaded
++ Selected users receive low-toner alerts at their account email. +
+No roles loaded
++ Every active member of a selected role receives low-toner alerts. +
++ Comma-separated. Empty uses the site-wide alert recipients + (Settings > System > Email). +
++ Low-toner alerts post to this team's webhook (set on Settings > + Support Teams). Empty uses the site-wide alert webhook. +
+{name} {color} toner is at ' + f'{remaining:.0f}% ({tier}).
{link_html}') + webhook_text = f'{name} {color} toner at {remaining:.0f}% ({tier})' + try: + send_webhook(subject, webhook_text, url=alert_team_webhook()) + recipients = alert_recipients() or site_alert_recipients() + if recipients: + send_email(recipients, subject, html) + except Exception: + logger.exception('Toner alert failed for printer %s', printer.printerid) + + +def check_supplies(service=None): + """Poll Zabbix for all active printers' toner and alert on crossings. + + Returns a summary dict. Intended to run on a schedule (a scheduled task or + cron calling `flask printers check-toner-alerts`).""" + from shopdb.api import Asset + from .zabbix_service import ZabbixService + + service = service or ZabbixService() + summary = {'printers': 0, 'polled': 0, 'alerts': 0, 'rearmed': 0} + if not service.isconfigured: + logger.warning('Zabbix not configured; skipping toner alert poll') + return summary + + warning_threshold = _threshold('printers_alert_warning_threshold', 5) + critical_threshold = _threshold('printers_alert_critical_threshold', 0) + + printers = db.session.query(Printer).join(Asset).filter( + Asset.isactive == True).all() # noqa: E712 + for printer in printers: + summary['printers'] += 1 + ipaddress = _printer_ipaddress(printer) + if not ipaddress or ipaddress == 'USB': + continue + supplies = service.getsuppliesbyip(ipaddress) + if not supplies: + continue + summary['polled'] += 1 + for supply in supplies: + supplyname = supply.get('name', '') + if derivesupplytype(supplyname) != 'toner': + continue + try: + remaining = float(supply.get('level', 0)) + except (TypeError, ValueError): + continue + color = supply.get('color') or derivecolor(supplyname) + supplykey = color if color and color != 'none' else supplyname + supplykey = supplykey[:64] + tier = alerttier(remaining, warning_threshold, critical_threshold) + + row = PrinterSupplyAlert.query.filter_by( + printerid=printer.printerid, supplykey=supplykey).first() + prior = row.lasttier if row else 'ok' + if tier == prior: + continue + if row is None: + row = PrinterSupplyAlert( + printerid=printer.printerid, supplykey=supplykey, + lasttier=tier) + db.session.add(row) + else: + row.lasttier = tier + + if TIER_RANK[tier] > TIER_RANK[prior]: + _fire_alert(printer, color, remaining, tier) + summary['alerts'] += 1 + else: + summary['rearmed'] += 1 + + db.session.commit() + return summary diff --git a/plugins/printers/services/supply_parts.py b/plugins/printers/services/supply_parts.py index 178b109..8ba7c27 100644 --- a/plugins/printers/services/supply_parts.py +++ b/plugins/printers/services/supply_parts.py @@ -10,10 +10,34 @@ and reading the matching part numbers out of the database. from typing import Dict, List, Optional -# alert thresholds (percent remaining) +# display thresholds (percent remaining) for the ok/low/critical report badge CRITICAL_THRESHOLD = 5 LOW_THRESHOLD = 10 +# toner email-alert tiers (percent remaining). Distinct from the display +# badge above: a warning email fires at or below TONER_WARNING_THRESHOLD, a +# critical email at or below TONER_CRITICAL_THRESHOLD (empty). +TONER_WARNING_THRESHOLD = 5 +TONER_CRITICAL_THRESHOLD = 0 + + +def alerttier(remaining: float, warning: float = TONER_WARNING_THRESHOLD, + critical: float = TONER_CRITICAL_THRESHOLD) -> str: + """Map a toner percent-remaining to an email-alert tier. + + critical at or below the critical threshold (default 0 = empty), warning at + or below the warning threshold (default 5), else ok. Both are configurable + (printers_alert_critical_threshold / printers_alert_warning_threshold).""" + if remaining <= critical: + return 'critical' + if remaining <= warning: + return 'warning' + return 'ok' + + +# tier severity rank; an alert fires only when the rank increases (worsens) +TIER_RANK = {'ok': 0, 'warning': 1, 'critical': 2} + def derivesupplytype(name: str) -> str: """Map a Zabbix item name to a supply type.""" diff --git a/tests/test_plugins/test_printers_supply_alerts.py b/tests/test_plugins/test_printers_supply_alerts.py new file mode 100644 index 0000000..0bc07aa --- /dev/null +++ b/tests/test_plugins/test_printers_supply_alerts.py @@ -0,0 +1,165 @@ +"""Low-toner alert tests. + +Invariants: the alert tier boundaries (5% warning, 0% critical), the +fire-once-per-crossing state machine (re-arm on refill), toner-only scope, +and recipient routing to a support team's webhook. +""" + +import pytest +from unittest.mock import patch + +from shopdb.api import db +from shopdb.core.models import AssetType, Setting, SupportTeam +from plugins.printers.models import PrinterSupplyAlert +from plugins.printers.services.supply_parts import alerttier +from plugins.printers.services.supply_alerts import check_supplies + + +class FakeZabbix: + """Stand-in Zabbix service: canned supplies keyed by IP.""" + + def __init__(self, byip): + self._byip = byip + self.isconfigured = True + + def getsuppliesbyip(self, ip): + return self._byip.get(ip, []) + + +@pytest.fixture +def printer_assettype(db): + at = AssetType(assettype='printer', pluginname='printers', + tablename='printers', description='Printers') + db.session.add(at) + db.session.commit() + return at + + +@pytest.fixture +def ip_comtype(db): + from shopdb.core.models import CommunicationType + row = CommunicationType.query.filter_by(comtype='IP').first() + if not row: + row = CommunicationType(comtype='IP', description='IP address') + db.session.add(row) + db.session.commit() + return row + + +@pytest.fixture +def network_printer(client, auth_headers, printer_assettype, ip_comtype): + """A network printer at 10.0.0.5 with a black toner, via the real API.""" + resp = client.post('/api/printers', json={ + 'assetnumber': 'CSF01-LaserJet', + 'hostname': 'wjprn05', + 'ipaddress': '10.0.0.5', + }, headers=auth_headers) + assert resp.status_code == 201, resp.get_json() + return '10.0.0.5' + + +def test_alerttier_boundaries(): + assert alerttier(50) == 'ok' + assert alerttier(6) == 'ok' + assert alerttier(5) == 'warning' + assert alerttier(1) == 'warning' + assert alerttier(0) == 'critical' + + +def _toner(level, color='black'): + return {'name': f'{color} toner', 'level': level, 'color': color} + + +def test_warning_crossing_fires_once_and_rearms(app, db, network_printer): + ip = network_printer + with app.app_context(): + # 20% -> ok, no alert + with patch('shopdb.api.send_webhook') as hook1, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(20)]})) + assert not hook1.called + + # 4% -> warning, alert fires + with patch('shopdb.api.send_webhook') as hook2, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(4)]})) + assert hook2.called + + # still 3% -> already warning, no repeat + with patch('shopdb.api.send_webhook') as hook3, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(3)]})) + assert not hook3.called + + # refill to 80% -> re-arm (state back to ok), no alert + with patch('shopdb.api.send_webhook') as hook4, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(80)]})) + assert not hook4.called + row = PrinterSupplyAlert.query.filter_by(supplykey='black').first() + assert row.lasttier == 'ok' + + # drop to 2% again -> warning fires again after the re-arm + with patch('shopdb.api.send_webhook') as hook5, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(2)]})) + assert hook5.called + + +def test_direct_to_critical_fires_critical(app, db, network_printer): + ip = network_printer + with app.app_context(): + with patch('shopdb.api.send_webhook') as hook, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(0)]})) + assert hook.called + subject = hook.call_args.args[0] + assert 'CRITICAL' in subject + row = PrinterSupplyAlert.query.filter_by(supplykey='black').first() + assert row.lasttier == 'critical' + + +def test_alerttier_respects_custom_thresholds(): + # warning at <=20, critical at <=8 + assert alerttier(25, warning=20, critical=8) == 'ok' + assert alerttier(20, warning=20, critical=8) == 'warning' + assert alerttier(8, warning=20, critical=8) == 'critical' + + +def test_custom_warning_threshold_setting_fires(app, db, network_printer): + ip = network_printer + with app.app_context(): + Setting.set('printers_alert_warning_threshold', '20') + # 15% would be 'ok' under the default 5% warning, but fires at 20% + with patch('shopdb.api.send_webhook') as hook, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(15)]})) + assert hook.called + row = PrinterSupplyAlert.query.filter_by(supplykey='black').first() + assert row.lasttier == 'warning' + + +def test_non_toner_supply_ignored(app, db, network_printer): + ip = network_printer + with app.app_context(): + waste = {'name': 'waste toner box', 'level': 1, 'color': 'none'} + with patch('shopdb.api.send_email') as email, \ + patch('shopdb.api.send_webhook'): + check_supplies(FakeZabbix({ip: [waste]})) + assert not email.called + + +def test_routes_to_support_team_webhook(app, db, network_printer): + ip = network_printer + with app.app_context(): + team = SupportTeam(teamname='Print Team', + webhookurl='https://teams.example/print') + db.session.add(team) + db.session.commit() + Setting.set('printers_alert_supportteamid', str(team.supportteamid)) + + with patch('shopdb.api.send_webhook') as webhook, \ + patch('shopdb.api.send_email'): + check_supplies(FakeZabbix({ip: [_toner(3)]})) + assert webhook.called + assert webhook.call_args.kwargs['url'] == 'https://teams.example/print'