Zabbix supply backend rebuild + data-driven model supplies

Rewrite the printer Zabbix integration (Bearer auth, host-by-IP, tag-based
supply lookup, ping) and replace the hardcoded toner table with a
modelsupplies table + CRUD + seed. Add mock Zabbix server, live test
harness, and the Playwright screenshot tooling.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-06-26 08:35:02 -04:00
parent 9d78b363dc
commit f1b3b65532
14 changed files with 1411 additions and 103 deletions

View File

@@ -0,0 +1,140 @@
"""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)