shopdb-flask/tools/mock_zabbix.py

"""Mock Zabbix 7.0 JSON-RPC server for testing the printer supply integration.

Speaks just enough of the Zabbix API (api_jsonrpc.php) to exercise our
ZabbixService end to end over real HTTP: Bearer token auth, host.get by IP,
and item.get returning printer supply items tagged the way the real Zabbix
templates tag them (component=supplies, type=level, color=<color>).

Run standalone:
    python tools/mock_zabbix.py --port 18080 --token testtoken

Then point the app at it:
    ZABBIX_ENABLED=true
    ZABBIX_URL=http://localhost:18080
    ZABBIX_TOKEN=testtoken

It is also imported by tests/test_plugins/test_zabbix_live.py, which boots it
in a background thread.
"""

import argparse
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


# one fake host, named by its IP (matches how the real shop Zabbix names hosts)
HOSTS = {
    '10.20.30.40': '10501',
}

# supply level items for host 10501, with Zabbix-style tags
SUPPLY_ITEMS = {
    '10501': [
        {'itemid': '1', 'name': 'Black Toner Level', 'lastvalue': '4',
         'status': '0', 'state': '0',
         'tags': [{'tag': 'component', 'value': 'supplies'},
                  {'tag': 'type', 'value': 'level'},
                  {'tag': 'color', 'value': 'black'}]},
        {'itemid': '2', 'name': 'Cyan Toner Level', 'lastvalue': '60',
         'status': '0', 'state': '0',
         'tags': [{'tag': 'component', 'value': 'supplies'},
                  {'tag': 'type', 'value': 'level'},
                  {'tag': 'color', 'value': 'cyan'}]},
        {'itemid': '3', 'name': 'Waste Cartridge Level', 'lastvalue': '97',
         'status': '0', 'state': '0',
         'tags': [{'tag': 'component', 'value': 'supplies'},
                  {'tag': 'type', 'value': 'level'}]},
        {'itemid': '4', 'name': 'Disabled Drum Level', 'lastvalue': '0',
         'status': '1', 'state': '0',
         'tags': [{'tag': 'component', 'value': 'supplies'},
                  {'tag': 'type', 'value': 'level'}]},
    ],
}

PING_ITEMS = {
    '10501': '1',
}


def build_handler(token):
    class ZabbixHandler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass  # quiet

        def _send(self, payload, code=200):
            body = json.dumps(payload).encode()
            self.send_response(code)
            self.send_header('Content-Type', 'application/json')
            self.send_header('Content-Length', str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            # reachability probe hits the endpoint with GET
            self._send({'jsonrpc': '2.0', 'error': {'code': -32600}, 'id': None})

        def do_POST(self):
            length = int(self.headers.get('Content-Length', 0))
            request = json.loads(self.rfile.read(length) or b'{}')
            method = request.get('method')
            params = request.get('params', {})
            reqid = request.get('id', 1)

            auth = self.headers.get('Authorization', '')
            if auth != f'Bearer {token}':
                self._send({'jsonrpc': '2.0',
                            'error': {'code': -32602, 'message': 'Not authorised'},
                            'id': reqid})
                return

            result = self._dispatch(method, params)
            self._send({'jsonrpc': '2.0', 'result': result, 'id': reqid})

        def _dispatch(self, method, params):
            if method in ('apiinfo.version',):
                return '7.0.0'
            if method == 'hostgroup.get':
                return [{'groupid': '1'}]
            if method == 'host.get':
                ips = (params.get('filter') or {}).get('host', [])
                out = []
                for ip in ips:
                    if ip in HOSTS:
                        out.append({'hostid': HOSTS[ip], 'host': ip, 'name': ip})
                return out
            if method == 'item.get':
                hostids = params.get('hostids')
                hostid = hostids[0] if isinstance(hostids, list) else hostids
                search = params.get('search') or {}
                if 'icmpping' in (search.get('key_') or ''):
                    value = PING_ITEMS.get(hostid)
                    return [{'lastvalue': value}] if value is not None else []
                return SUPPLY_ITEMS.get(hostid, [])
            return []

    return ZabbixHandler


def serve(port, token):
    server = ThreadingHTTPServer(('127.0.0.1', port), build_handler(token))
    return server


def serve_in_thread(port, token):
    """Start the mock in a daemon thread. Returns the server (call shutdown())."""
    server = serve(port, token)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Mock Zabbix JSON-RPC server')
    parser.add_argument('--port', type=int, default=18080)
    parser.add_argument('--token', default='testtoken')
    args = parser.parse_args()
    httpd = serve(args.port, args.token)
    print(f"Mock Zabbix on http://127.0.0.1:{args.port}/api_jsonrpc.php "
          f"(token: {args.token})")
    print(f"Known host: {list(HOSTS)[0]} -> supplies + icmpping")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        httpd.shutdown()