shopdb-flask/tools/setup_zabbix_fixture.py

"""Provision the local docker Zabbix (tools/docker-compose.zabbix.yml) with a
printer host shaped like the real shop Zabbix, so ZabbixService can be tested
against a live server instead of the mock.

Creates:
  - an API token (printed at the end; put it in .env as ZABBIX_TOKEN)
  - host group "Printers"
  - host named by IP "10.20.30.40" with an SNMP-less agent interface
  - trapper items tagged component=supplies, type=level, color=<c>
  - an icmpping item

Item values are pushed afterwards with zabbix_sender (see seed_values()).
Run:  python tools/setup_zabbix_fixture.py
"""

import sys
import requests

BASE = "http://localhost:8888/api_jsonrpc.php"
HOST_IP = "10.20.30.40"
ADMIN_USER = "Admin"
ADMIN_PASS = "zabbix"

# name, color tag, key (must be unique per host)
SUPPLY_ITEMS = [
    ("Black Toner Level", "black", "supply.black"),
    ("Cyan Toner Level", "cyan", "supply.cyan"),
    ("Magenta Toner Level", "magenta", "supply.magenta"),
    ("Yellow Toner Level", "yellow", "supply.yellow"),
    ("Waste Cartridge Level", "", "supply.waste"),
]


def call(method, params, auth=None):
    headers = {"Content-Type": "application/json-rpc"}
    if auth:
        headers["Authorization"] = f"Bearer {auth}"
    resp = requests.post(
        BASE,
        json={"jsonrpc": "2.0", "method": method, "params": params, "id": 1},
        headers=headers,
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    if "error" in data:
        raise RuntimeError(f"{method}: {data['error']}")
    return data["result"]


def login():
    # user.login returns a session token; usable as Bearer in 7.0
    return call("user.login", {"username": ADMIN_USER, "password": ADMIN_PASS})


def make_api_token(sess):
    # a real, persistent API token (survives logout, matches prod usage)
    existing = call("token.get", {"filter": {"name": "shopdb-flask-test"}}, sess)
    if existing:
        tokenid = existing[0]["tokenid"]
    else:
        created = call("token.create", {
            "name": "shopdb-flask-test",
            "userid": call("user.get", {"output": ["userid"],
                                        "filter": {"username": ADMIN_USER}}, sess)[0]["userid"],
        }, sess)
        tokenid = created["tokenids"][0]
    return call("token.generate", [tokenid], sess)[0]["token"]


def ensure_hostgroup(sess):
    found = call("hostgroup.get", {"filter": {"name": "Printers"}}, sess)
    if found:
        return found[0]["groupid"]
    return call("hostgroup.create", {"name": "Printers"}, sess)["groupids"][0]


def ensure_host(sess, groupid):
    found = call("host.get", {"filter": {"host": [HOST_IP]}, "output": ["hostid"]}, sess)
    if found:
        hostid = found[0]["hostid"]
        # wipe existing items so re-runs are clean
        items = call("item.get", {"hostids": hostid, "output": ["itemid"]}, sess)
        if items:
            call("item.delete", [i["itemid"] for i in items], sess)
        return hostid
    created = call("host.create", {
        "host": HOST_IP,
        "groups": [{"groupid": groupid}],
        "interfaces": [{
            "type": 1, "main": 1, "useip": 1,
            "ip": HOST_IP, "dns": "", "port": "10050",
        }],
    }, sess)
    return created["hostids"][0]


def create_items(sess, hostid):
    for name, color, key in SUPPLY_ITEMS:
        tags = [
            {"tag": "component", "value": "supplies"},
            {"tag": "type", "value": "level"},
        ]
        if color:
            tags.append({"tag": "color", "value": color})
        call("item.create", {
            "name": name,
            "key_": key,
            "hostid": hostid,
            "type": 2,            # Zabbix trapper, lets zabbix_sender push values
            "value_type": 3,      # unsigned int
            "tags": tags,
        }, sess)
    # ping item, untagged, key icmpping
    call("item.create", {
        "name": "ICMP ping",
        "key_": "icmpping",
        "hostid": hostid,
        "type": 2,
        "value_type": 3,
    }, sess)


def main():
    sess = login()
    token = make_api_token(sess)
    groupid = ensure_hostgroup(sess)
    hostid = ensure_host(sess, groupid)
    create_items(sess, hostid)
    print("OK")
    print(f"hostid={hostid}")
    print(f"ZABBIX_TOKEN={token}")


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        print(f"FAILED: {exc}", file=sys.stderr)
        sys.exit(1)