shopdb-flask/tests/test_plugins/test_printers_supply_alerts.py

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