shopdb-flask/tests/test_plugins/test_zabbix_live.py

"""End-to-end test of ZabbixService against the mock Zabbix JSON-RPC server.

Exercises the real HTTP path: Bearer auth, host.get by IP, tag-filtered
item.get, supply parsing, ping, and the low-supplies roll-up.
"""

import socket

import pytest

from tools.mock_zabbix import serve_in_thread


def _free_port():
    sock = socket.socket()
    sock.bind(('127.0.0.1', 0))
    port = sock.getsockname()[1]
    sock.close()
    return port


@pytest.fixture
def mock_zabbix(app):
    """Boot the mock server and point app config at it."""
    port = _free_port()
    server = serve_in_thread(port, 'testtoken')
    app.config['ZABBIX_ENABLED'] = True
    app.config['ZABBIX_URL'] = f'http://127.0.0.1:{port}'
    app.config['ZABBIX_TOKEN'] = 'testtoken'
    yield
    server.shutdown()


def test_service_reachable_and_configured(app, db, mock_zabbix):
    from plugins.printers.services import ZabbixService
    with app.app_context():
        service = ZabbixService()
        assert service.isconfigured
        assert service.isreachable


def test_gethostid_by_ip(app, db, mock_zabbix):
    from plugins.printers.services import ZabbixService
    with app.app_context():
        service = ZabbixService()
        assert service.gethostidbyip('10.20.30.40') == '10501'
        assert service.gethostidbyip('1.2.3.4') is None


def test_supplies_parsed_with_color_and_status_filter(app, db, mock_zabbix):
    from plugins.printers.services import ZabbixService
    with app.app_context():
        supplies = ZabbixService().getsuppliesbyip('10.20.30.40')
        # disabled item (status=1) dropped, three active remain
        names = {s['name'] for s in supplies}
        assert 'Disabled Drum Level' not in names
        assert len(supplies) == 3
        black = next(s for s in supplies if s['name'].startswith('Black'))
        assert black['color'] == 'black'
        assert black['level'] == 4


def test_ping_status(app, db, mock_zabbix):
    from plugins.printers.services import ZabbixService
    with app.app_context():
        assert ZabbixService().getpingstatus('10.20.30.40') == '1'


def test_bad_token_returns_no_data(app, db):
    """Wrong token -> API errors -> service returns nothing, fails soft."""
    port = _free_port()
    server = serve_in_thread(port, 'rightsecret')
    app.config['ZABBIX_ENABLED'] = True
    app.config['ZABBIX_URL'] = f'http://127.0.0.1:{port}'
    app.config['ZABBIX_TOKEN'] = 'wrongsecret'
    try:
        with app.app_context():
            assert ZabbixService_gethost(app) is None
    finally:
        server.shutdown()


def ZabbixService_gethost(app):
    from plugins.printers.services import ZabbixService
    return ZabbixService().gethostidbyip('10.20.30.40')


def test_low_supplies_rollup_flags_waste_and_toner(app, db, mock_zabbix):
    """The mock host has a 4% black toner and a 97%-full waste -> both flagged."""
    from shopdb.core.models import (
        Vendor, Model, Asset, AssetType, Communication, CommunicationType
    )
    from plugins.printers.models import Printer
    from plugins.printers.api.asset_routes import _get_low_supplies_data
    from shopdb.extensions import cache

    with app.app_context():
        # an HP printer at the mock's known IP
        vendor = Vendor(vendor='HP')
        db.session.add(vendor)
        db.session.flush()
        model = Model(modelnumber='HP M454', vendorid=vendor.vendorid)
        db.session.add(model)

        atype = AssetType.query.filter_by(assettype='printer').first()
        if not atype:
            atype = AssetType(assettype='printer', pluginname='printers',
                              tablename='printers')
            db.session.add(atype)
            db.session.flush()
        asset = Asset(assetnumber='PRN-1', name='Test Printer',
                      assettypeid=atype.assettypeid)
        db.session.add(asset)
        db.session.flush()

        printer = Printer(assetid=asset.assetid, vendorid=vendor.vendorid,
                          modelnumberid=model.modelnumberid)
        db.session.add(printer)

        comtype = CommunicationType.query.filter_by(comtype='IP').first()
        if not comtype:
            comtype = CommunicationType(comtype='IP')
            db.session.add(comtype)
            db.session.flush()
        db.session.add(Communication(assetid=asset.assetid,
                                     comtypeid=comtype.comtypeid,
                                     ipaddress='10.20.30.40', isprimary=True))
        db.session.commit()

        cache.delete('printers_low_supplies')
        data = _get_low_supplies_data()

        assert data['summary']['total_checked'] == 1
        assert len(data['printers']) == 1
        row = data['printers'][0]
        statuses = {s['name']: s['status'] for s in row['supplies']}
        # 4% black toner is critical
        assert statuses['Black Toner Level'] == 'critical'
        # 97%-full waste (HP, non-inverted) -> 3% remaining -> critical
        assert statuses['Waste Cartridge Level'] == 'critical'
        # 60% cyan is FINE, so it is not listed at all. The report answers
        # "what needs replacing"; including healthy cartridges made the reader
        # hunt for the problem inside the row.
        assert 'Cyan Toner Level' not in statuses