ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

printers: low-toner alerts with configurable thresholds + support-team routing

Browse Source 
Poll Zabbix for toner levels on a schedule and email/webhook on a downward
crossing. Warning fires at or below the warning threshold (default 5%),
critical at the critical threshold (default 0%); both thresholds are settings.
State lives in printersupplyalerts so an alert fires once per crossing and
re-arms after a refill.

Recipients mirror the printedparts pattern: plugin-scoped shopdb users +
roles + free-text emails (falling back to the site alert_recipients), and a
chosen support team's webhook (falling back to the site alert_webhook_url).

- PrinterSupplyAlert model + migration printers0002supplyalerts
- alerttier(remaining, warning, critical) + check_supplies poller
- flask printers check-toner-alerts CLI (run via scheduled task/cron)
- printers alert settings + Low-Toner Alerts settings page
- 7 tests: tier boundaries, once-per-crossing + re-arm, toner-only scope,
  custom thresholds, support-team webhook routing
This commit is contained in:
cproudlock
2026-07-22 14:47:59 -04:00
parent fb188fd302
commit b211e817d5
10 changed files with 739 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
plugins/printers/frontend/routes.js
View File

@@ -20,6 +20,12 @@ export default [
component: () => import('./views/ZabbixSettings.vue'),
meta: { requiresAuth: true, requiresAdmin: true, plugin: 'printers' }
},
{
path: 'settings/printer-alerts',
name: 'printer-alerts',
component: () => import('./views/PrinterAlertsSettings.vue'),
meta: { requiresAuth: true, requiresAdmin: true, plugin: 'printers' }
},
{
path: 'printers',
name: 'printers',

192
plugins/printers/frontend/views/PrinterAlertsSettings.vue Normal file
View File

@@ -0,0 +1,192 @@
<template>
<div>
<div class="page-header">
<h2>Low-Toner Alerts</h2>
</div>
<div class="card form-card">
<div v-if="message" class="settings-success">{{ message }}</div>
<div v-if="error" class="error-message">{{ error }}</div>
<p class="field-hint">
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.
</p>
<div class="form-row">
<div class="form-group">
<label>Warning threshold (% remaining)</label>
<input v-model.number="values.printers_alert_warning_threshold"
type="number" min="0" max="100" class="form-control" />
</div>
<div class="form-group">
<label>Critical threshold (% remaining)</label>
<input v-model.number="values.printers_alert_critical_threshold"
type="number" min="0" max="100" class="form-control" />
</div>
</div>
<div class="form-group">
<label>Alert shopdb users</label>
<div class="user-picker">
<label v-for="candidate in users" :key="candidate.userid" class="user-row">
<input type="checkbox" :value="String(candidate.userid)"
v-model="selectedUserids" />
<span>{{ candidate.username }}</span>
<span class="user-email">{{ candidate.email }}</span>
</label>
<p v-if="users.length === 0" class="field-hint">No users loaded</p>
</div>
<p class="field-hint">
Selected users receive low-toner alerts at their account email.
</p>
</div>
<div class="form-group">
<label>Alert roles</label>
<div class="user-picker">
<label v-for="role in roles" :key="role.roleid" class="user-row">
<input type="checkbox" :value="String(role.roleid)"
v-model="selectedRoleids" />
<span>{{ role.rolename }}</span>
<span class="user-email">{{ role.description }}</span>
</label>
<p v-if="roles.length === 0" class="field-hint">No roles loaded</p>
</div>
<p class="field-hint">
Every active member of a selected role receives low-toner alerts.
</p>
</div>
<div class="form-group">
<label>Additional alert emails</label>
<input v-model="values.printers_alert_email" type="text"
class="form-control" placeholder="print-team@example.com, lead@example.com" />
<p class="field-hint">
Comma-separated. Empty uses the site-wide alert recipients
(Settings &gt; System &gt; Email).
</p>
</div>
<div class="form-group">
<label>Alert support team (Teams webhook)</label>
<select v-model="values.printers_alert_supportteamid" class="form-control">
<option value="">Site default webhook</option>
<option v-for="team in supportTeams" :key="team.supportteamid" :value="String(team.supportteamid)">
{{ team.teamname }}{{ team.webhookurl ? '' : ' (no webhook set)' }}
</option>
</select>
<p class="field-hint">
Low-toner alerts post to this team's webhook (set on Settings &gt;
Support Teams). Empty uses the site-wide alert webhook.
</p>
</div>
<button class="btn btn-primary" :disabled="saving" @click="save">
{{ saving ? 'Saving...' : 'Save' }}
</button>
</div>
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { settingsApi, usersApi, supportteamsApi } from '@/api'
const KEYS = [
'printers_alert_email',
'printers_alert_userids',
'printers_alert_roleids',
'printers_alert_supportteamid',
'printers_alert_warning_threshold',
'printers_alert_critical_threshold'
]
const values = ref({
printers_alert_email: '',
printers_alert_userids: '',
printers_alert_roleids: '',
printers_alert_supportteamid: '',
printers_alert_warning_threshold: 5,
printers_alert_critical_threshold: 0
})
const supportTeams = ref([])
const users = ref([])
const selectedUserids = ref([])
const roles = ref([])
const selectedRoleids = ref([])
const saving = ref(false)
const message = ref('')
const error = ref('')
onMounted(async () => {
try {
const response = await settingsApi.list({ category: 'printers' })
const rows = response.data.data || []
for (const row of rows) {
if (KEYS.includes(row.key)) values.value[row.key] = row.value
}
values.value.printers_alert_warning_threshold =
parseInt(values.value.printers_alert_warning_threshold, 10) || 0
values.value.printers_alert_critical_threshold =
parseInt(values.value.printers_alert_critical_threshold, 10) || 0
selectedUserids.value = (values.value.printers_alert_userids || '')
.split(',').map(id => id.trim()).filter(Boolean)
const usersResponse = await usersApi.list()
users.value = (usersResponse.data.data || []).filter(
candidate => candidate.isactive && candidate.email)
selectedRoleids.value = (values.value.printers_alert_roleids || '')
.split(',').map(id => id.trim()).filter(Boolean)
const rolesResponse = await usersApi.roles.list()
roles.value = rolesResponse.data.data || []
const teamsResponse = await supportteamsApi.list()
supportTeams.value = teamsResponse.data.data || []
} catch (loadError) {
error.value = 'Could not load settings'
console.error(loadError)
}
})
async function save() {
saving.value = true
message.value = ''
error.value = ''
try {
values.value.printers_alert_userids = selectedUserids.value.join(',')
values.value.printers_alert_roleids = selectedRoleids.value.join(',')
for (const key of KEYS) {
await settingsApi.update(key, String(values.value[key] ?? ''))
}
message.value = 'Settings saved'
} catch (saveError) {
error.value = saveError.response?.data?.data?.error?.message || 'Save failed'
} finally {
saving.value = false
}
}
</script>
<style scoped>
.field-hint { color: var(--text-light); font-size: 0.85rem; margin-top: 0.25rem; }
.form-row { display: flex; gap: 1rem; }
.form-row .form-group { flex: 1; }
.user-picker {
max-height: 12rem;
overflow-y: auto;
border: 1px solid var(--border);
border-radius: 0.35rem;
padding: 0.5rem;
display: flex;
flex-direction: column;
gap: 0.35rem;
}
.user-row {
display: flex;
align-items: center;
gap: 0.5rem;
cursor: pointer;
}
.user-email { color: var(--text-light); font-size: 0.85rem; }
</style>

47
plugins/printers/migrations/versions/0002_printersupplyalerts.py Normal file
View File

@@ -0,0 +1,47 @@
"""printers: printersupplyalerts table (toner alert crossing state).
Stores the last tier alerted per (printer, supply) so the poller fires once
on a downward crossing and re-arms after a refill. Idempotent create.
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'printers0002supplyalerts'
down_revision = 'printers0001anchor'
branch_labels = None
depends_on = None
def upgrade():
bind = op.get_bind()
inspector = sa.inspect(bind)
if 'printersupplyalerts' in inspector.get_table_names():
return
op.create_table(
'printersupplyalerts',
sa.Column('printersupplyalertid', sa.Integer(), primary_key=True),
sa.Column('printerid', sa.Integer(), nullable=False),
sa.Column('supplykey', sa.String(length=64), nullable=False),
sa.Column('lasttier', sa.String(length=16), nullable=False,
server_default='ok'),
sa.Column('createddate', sa.DateTime(), nullable=False),
sa.Column('modifieddate', sa.DateTime(), nullable=False),
sa.Column('isactive', sa.Boolean(), nullable=False,
server_default=sa.true()),
sa.ForeignKeyConstraint(['printerid'], ['printers.printerid'],
ondelete='CASCADE'),
sa.UniqueConstraint('printerid', 'supplykey',
name='uq_printersupplyalert_printer_key'),
)
op.create_index('ix_printersupplyalerts_printerid',
'printersupplyalerts', ['printerid'])
def downgrade():
bind = op.get_bind()
inspector = sa.inspect(bind)
if 'printersupplyalerts' not in inspector.get_table_names():
return
op.drop_table('printersupplyalerts')

2
plugins/printers/models/__init__.py
View File

@@ -8,12 +8,14 @@ from .model_supply import ( # data-driven model -> toner/drum/waste mapping
SUPPLY_COLORS,
CAPACITY_TIERS,
)
from .supply_alert import PrinterSupplyAlert # per-printer toner alert state
__all__ = [
'Printer',
'PrinterType',
'PrinterDriver',
'ModelSupply',
'PrinterSupplyAlert',
'SUPPLY_TYPES',
'SUPPLY_COLORS',
'CAPACITY_TIERS',

41
plugins/printers/models/supply_alert.py Normal file
View File

@@ -0,0 +1,41 @@
"""Per-printer supply alert state.
The supply poller (flask printers check-toner-alerts) is stateless: it reads
live levels from Zabbix each run. To alert once per downward crossing (and
re-arm after a refill) it needs to remember the last tier it alerted for each
printer + supply. One row per (printerid, supplykey); supplykey is the toner
color (black/cyan/magenta/yellow) or the raw item name when color is unknown.
"""
from shopdb.extensions import db
from shopdb.core.models.base import BaseModel
class PrinterSupplyAlert(BaseModel):
"""Last-alerted tier for one printer supply. lasttier in ok/warning/critical."""
__tablename__ = 'printersupplyalerts'
printersupplyalertid = db.Column(db.Integer, primary_key=True)
printerid = db.Column(
db.Integer,
db.ForeignKey('printers.printerid', ondelete='CASCADE'),
nullable=False,
index=True,
)
supplykey = db.Column(
db.String(64),
nullable=False,
comment='Toner color, or Zabbix item name when color is unknown',
)
lasttier = db.Column(
db.String(16),
nullable=False,
default='ok',
comment='Last tier alerted: ok, warning, or critical',
)
__table_args__ = (
db.UniqueConstraint('printerid', 'supplykey',
name='uq_printersupplyalert_printer_key'),
)

69
plugins/printers/plugin.py
View File

@@ -9,9 +9,11 @@ from flask import Flask, Blueprint
import click
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, AssetType
from shopdb.api import db, AssetType, Setting
from .models import Printer, PrinterType, ModelSupply, PrinterDriver
from .models import (
Printer, PrinterType, ModelSupply, PrinterDriver, PrinterSupplyAlert
)
from .api import printers_asset_bp
from .services import ZabbixService
@@ -76,6 +78,7 @@ class PrintersPlugin(BasePlugin):
PrinterType, # printer type classification
PrinterDriver, # driver links (SMB/HTTP)
ModelSupply, # model -> toner/drum/waste part numbers
PrinterSupplyAlert, # per-printer toner alert crossing state
]
def get_services(self) -> Dict[str, Type]:
@@ -103,8 +106,41 @@ class PrintersPlugin(BasePlugin):
with app.app_context():
self._ensure_asset_type()
self._ensure_printer_types()
self._seed_settings()
logger.info("Printers plugin installed")
def on_enable(self, app: Flask) -> None:
# Idempotent re-seed so settings added in later versions reach sites
# that installed earlier (enable runs on every upgrade cycle).
with app.app_context():
self._seed_settings()
def _seed_settings(self) -> None:
"""Seed low-toner alert recipient settings (idempotent)."""
defaults = [
('printers_alert_email', '', 'string',
'Comma-separated low-toner alert recipients; empty uses the '
'site alert_recipients'),
('printers_alert_userids', '', 'string',
'Comma-separated shopdb user ids whose account emails receive '
'low-toner alerts'),
('printers_alert_roleids', '', 'string',
'Comma-separated role ids; every active member of these roles '
'receives low-toner alerts'),
('printers_alert_supportteamid', '', 'string',
'Support team whose webhook receives low-toner alerts; empty '
'uses the site alert_webhook_url'),
('printers_alert_warning_threshold', '5', 'integer',
'Toner percent remaining at or below which a warning email fires'),
('printers_alert_critical_threshold', '0', 'integer',
'Toner percent remaining at or below which a critical email fires'),
]
for key, value, valuetype, description in defaults:
if Setting.get(key) is None:
Setting.set(key, value, valuetype=valuetype,
category='printers', description=description)
db.session.commit()
def _ensure_asset_type(self) -> None:
"""Ensure printer asset type exists."""
existing = AssetType.query.filter_by(assettype='printer').first()
@@ -151,6 +187,19 @@ class PrintersPlugin(BasePlugin):
"""Called when plugin is uninstalled."""
logger.info("Printers plugin uninstalled")
def get_settings_cards(self) -> List[dict]:
return [
{
'group': 'Printers',
'to': '/settings/printer-alerts',
'icon': 'bell',
'title': 'Low-Toner Alerts',
'description': 'Who gets warning (5%) and critical (0%) toner '
'emails, and which support team webhook',
'position': 48,
},
]
def get_cli_commands(self) -> List:
"""Return CLI commands for this plugin."""
@@ -159,6 +208,22 @@ class PrintersPlugin(BasePlugin):
"""Printers plugin commands."""
pass
@printerscli.command('check-toner-alerts')
def checktoneralerts():
"""Poll Zabbix for all printers and email/webhook low-toner crossings.
Run on a schedule (scheduled task / cron). Fires a warning at or
below 5 percent and a critical at 0 percent, once per crossing."""
from flask import current_app
from .services import check_supplies
with current_app.app_context():
summary = check_supplies()
click.echo(
f"Toner poll: {summary['polled']}/{summary['printers']} "
f"printers reachable, {summary['alerts']} alert(s) sent, "
f"{summary['rearmed']} re-armed.")
@printerscli.command('check-supplies')
@click.argument('ip')
def checksupplies(ip):

4
plugins/printers/services/__init__.py
View File

@@ -6,7 +6,9 @@ from .supply_parts import (
derivesupplytype,
derivecolor,
lookupsupplies,
alerttier,
)
from .supply_alerts import check_supplies
from .seed_supplies import seedsupplies
__all__ = [
@@ -15,5 +17,7 @@ __all__ = [
'derivesupplytype',
'derivecolor',
'lookupsupplies',
'alerttier',
'check_supplies',
'seedsupplies',
]

190
plugins/printers/services/supply_alerts.py Normal file
View File

@@ -0,0 +1,190 @@
"""Low-toner email/webhook alerting.
The report shows live levels; this fires an alert when a toner crosses a
threshold on its way DOWN. A warning email fires at or below 5 percent, a
critical email at 0 percent (empty). State lives in printersupplyalerts so an
alert fires once per crossing and re-arms only after a refill.
Recipients mirror the printedparts pattern: a plugin-scoped set of shopdb
users + roles + a free-text email list, falling back to the site-wide
alert_recipients; the webhook routes to a chosen support team's webhook,
falling back to the site-wide alert_webhook_url.
"""
import logging
from shopdb.api import db, Setting
from ..models import Printer, PrinterSupplyAlert
from .supply_parts import derivesupplytype, derivecolor, alerttier, TIER_RANK
logger = logging.getLogger(__name__)
def alert_recipients():
"""Merge selected shopdb users' account emails with the free-text list.
Empty result means fall back to the site-wide alert_recipients."""
from shopdb.api import User, Role
recipients = []
userids = (Setting.get('printers_alert_userids') or '').strip()
for rawid in userids.split(','):
rawid = rawid.strip()
if not rawid.isdigit():
continue
user = db.session.get(User, int(rawid))
if user and user.isactive and user.email:
recipients.append(user.email)
roleids = (Setting.get('printers_alert_roleids') or '').strip()
for rawid in roleids.split(','):
rawid = rawid.strip()
if not rawid.isdigit():
continue
role = db.session.get(Role, int(rawid))
if role:
for user in role.users:
if user.isactive and user.email:
recipients.append(user.email)
freetext = Setting.get('printers_alert_email') or ''
for addr in freetext.replace(';', ',').split(','):
addr = addr.strip()
if addr:
recipients.append(addr)
# de-dupe, preserve order
seen = set()
unique = []
for addr in recipients:
low = addr.lower()
if low not in seen:
seen.add(low)
unique.append(addr)
return unique
def _threshold(key, default):
"""Read a numeric threshold setting, tolerating blank/garbage values."""
try:
raw = Setting.get(key)
return float(raw) if raw not in (None, '') else float(default)
except (ValueError, TypeError):
return float(default)
def site_alert_recipients():
"""Site-wide alert_recipients setting as a clean list (email fallback)."""
raw = Setting.get('alert_recipients') or ''
return [r.strip() for r in raw.replace(';', ',').split(',') if r.strip()]
def alert_team_webhook():
"""Webhook URL of the support team chosen for toner alerts, or None.
printers_alert_supportteamid selects a SupportTeam; alerts route to that
team's webhook. None -> send_webhook uses the site-wide default."""
team_id = Setting.get('printers_alert_supportteamid')
if not team_id:
return None
try:
from shopdb.core.models import SupportTeam
team = db.session.get(SupportTeam, int(team_id))
except (ValueError, TypeError):
return None
return (team.webhookurl or None) if team else None
def _printer_ipaddress(printer):
"""Primary (else any) communication IP for a printer's asset, or None."""
from shopdb.api import Communication
asset = printer.asset
if not asset:
return None
comm = Communication.query.filter_by(
assetid=asset.assetid, isprimary=True).first() \
or Communication.query.filter_by(assetid=asset.assetid).first()
return comm.ipaddress if comm else None
def _fire_alert(printer, color, remaining, tier):
"""Best-effort email + webhook for one toner crossing. Never raises."""
from shopdb.api import send_email, send_webhook
name = printer.asset.name or printer.asset.assetnumber if printer.asset \
else f'Printer {printer.printerid}'
label = 'CRITICAL - empty' if tier == 'critical' else 'Low'
subject = f'{label} toner: {name} ({color}) - {remaining:.0f}% remaining'
base = (Setting.get('site_base_url') or '').rstrip('/')
link = f'{base}/printers/{printer.printerid}' if base else ''
link_html = f'<p><a href="{link}">View {name}</a></p>' if link else ''
html = (f'<p><strong>{name}</strong> {color} toner is at '
f'<strong>{remaining:.0f}%</strong> ({tier}).</p>{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

26
plugins/printers/services/supply_parts.py
View File

@@ -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."""

165
tests/test_plugins/test_printers_supply_alerts.py Normal file
View File

@@ -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'