From 8812ae70523f7b196f998892e2c7360a02f8e0ca Mon Sep 17 00:00:00 2001 From: Tobias Waldekranz Date: Thu, 9 Jan 2025 14:31:09 +0100 Subject: [PATCH] yanger: Move ietf-hardware implementation to separate module --- src/statd/python/yanger/ietf_hardware.py | 90 +++++++++++++++++++ src/statd/python/yanger/yanger.py | 81 +---------------- .../cli/system-output/rootfs/run/system.json | 1 + 3 files changed, 93 insertions(+), 79 deletions(-) create mode 100644 src/statd/python/yanger/ietf_hardware.py create mode 100644 test/case/cli/system-output/rootfs/run/system.json diff --git a/src/statd/python/yanger/ietf_hardware.py b/src/statd/python/yanger/ietf_hardware.py new file mode 100644 index 00000000..70a4aa29 --- /dev/null +++ b/src/statd/python/yanger/ietf_hardware.py @@ -0,0 +1,90 @@ +import datetime +import os + +from .common import insert +from .host import HOST + + +def vpd_vendor_extensions(data): + vendor_extensions = [] + for ext in data: + vendor_extension = {} + vendor_extension["iana-enterprise-number"] = ext[0] + vendor_extension["extension-data"] = ext[1] + vendor_extensions.append(vendor_extension) + return vendor_extensions + + +def vpd_component(vpd): + component = {} + component["name"] = vpd.get("board") + component["infix-hardware:vpd-data"] = {} + + if vpd.get("data"): + component["class"] = "infix-hardware:vpd" + if vpd["data"].get("manufacture-date"): + mfgdate = datetime.datetime.strptime(vpd["data"]["manufacture-date"], + "%m/%d/%Y %H:%M:%S") + component["mfg-date"] = mfgdate.strftime("%Y-%m-%dT%H:%M:%SZ") + if vpd["data"].get("manufacter"): + component["mfg-name"] = vpd["data"]["manufacturer"] + if vpd["data"].get("product-name"): + component["model-name"] = vpd["data"]["product-name"] + if vpd["data"].get("serial-number"): + component["serial-num"] = vpd["data"]["serial-number"] + + # Set VPD-data entrys + for k, v in vpd["data"].items(): + if vpd["data"].get(k): + if k != "vendor-extension": + component["infix-hardware:vpd-data"][k] = v + else: + vendor_extensions=vpd_vendor_extensions(v) + component["infix-hardware:vpd-data"]["infix-hardware:vendor-extension"] = vendor_extensions + return component + + +def vpd_components(systemjson): + return [vpd_component(vpd) for vpd in systemjson.get("vpd", {}).values()] + + +def usb_port_components(systemjson): + usb_ports = systemjson.get("usb-ports", []) + + ports=[] + names=[] + for usb_port in usb_ports: + port={} + if usb_port.get("path"): + if usb_port["name"] in names: + continue + + path = usb_port["path"] + if os.path.basename(path) == "authorized_default": + # TODO: Untestable. Should be done via the host API + if os.path.exists(path): + with open(path, "r") as f: + names.append(usb_port["name"]) + data = int(f.readline().strip()) + enabled = "unlocked" if data == 1 else "locked" + port["state"] = {} + port["state"]["admin-state"] = enabled + port["name"] = usb_port["name"] + port["class"] = "infix-hardware:usb" + port["state"]["oper-state"] = "enabled" + ports.append(port) + + return ports + + +def operational(): + systemjson = HOST.read_json("/run/system.json") + + return { + "ietf-hardware:hardware": { + "component": + vpd_components(systemjson) + + usb_port_components(systemjson) + + [], + }, + } diff --git a/src/statd/python/yanger/yanger.py b/src/statd/python/yanger/yanger.py index 7d233739..ab3c2aa0 100644 --- a/src/statd/python/yanger/yanger.py +++ b/src/statd/python/yanger/yanger.py @@ -175,79 +175,6 @@ def iface_is_dsa(iface_in): return True -def get_vpd_vendor_extensions(data): - vendor_extensions = [] - for ext in data: - vendor_extension = {} - vendor_extension["iana-enterprise-number"] = ext[0] - vendor_extension["extension-data"] = ext[1] - vendor_extensions.append(vendor_extension) - return vendor_extensions - - -def get_vpd_data(vpd): - component = {} - component["name"] = vpd.get("board") - component["infix-hardware:vpd-data"] = {} - - if vpd.get("data"): - component["class"] = "infix-hardware:vpd" - if vpd["data"].get("manufacture-date"): - component["mfg-date"] = datetime.strptime(vpd["data"]["manufacture-date"],"%m/%d/%Y %H:%M:%S").strftime("%Y-%m-%dT%H:%M:%SZ") - if vpd["data"].get("manufacter"): - component["mfg-name"] = vpd["data"]["manufacturer"] - if vpd["data"].get("product-name"): - component["model-name"] = vpd["data"]["product-name"] - if vpd["data"].get("serial-number"): - component["serial-num"] = vpd["data"]["serial-number"] - - # Set VPD-data entrys - for k, v in vpd["data"].items(): - if vpd["data"].get(k): - if k != "vendor-extension": - component["infix-hardware:vpd-data"][k] = v - else: - vendor_extensions=get_vpd_vendor_extensions(v) - component["infix-hardware:vpd-data"]["infix-hardware:vendor-extension"] = vendor_extensions - return component - -def get_usb_ports(usb_ports): - ports=[] - names=[] - for usb_port in usb_ports: - port={} - if usb_port.get("path"): - if usb_port["name"] in names: - continue - - path = usb_port["path"] - if os.path.basename(path) == "authorized_default": - if os.path.exists(path): - with open(path, "r") as f: - names.append(usb_port["name"]) - data = int(f.readline().strip()) - enabled = "unlocked" if data == 1 else "locked" - port["state"] = {} - port["state"]["admin-state"] = enabled - port["name"] = usb_port["name"] - port["class"] = "infix-hardware:usb" - port["state"]["oper-state"] = "enabled" - ports.append(port) - - return ports - - -def add_hardware(hw_out): - data = run_json_cmd(['cat', "/run/system.json"], "system.json") - components = [] - for _, vpd in data.get("vpd").items(): - component = get_vpd_data(vpd) - components.append(component) - if data.get("usb-ports", None): - components.extend(get_usb_ports(data["usb-ports"])) - insert(hw_out, "component", components) - - def uptime2datetime(uptime): """ Convert uptime to YANG format (YYYY-MM-DDTHH:MM:SS+00:00) @@ -1083,12 +1010,8 @@ def main(): add_ospf(yang_data['ietf-routing:routing']['control-plane-protocols']) elif args.model == 'ietf-hardware': - yang_data = { - "ietf-hardware:hardware": { - } - } - add_hardware(yang_data["ietf-hardware:hardware"]) - + from . import ietf_hardware + yang_data = ietf_hardware.operational() elif args.model == 'infix-containers': from . import infix_containers yang_data = infix_containers.operational() diff --git a/test/case/cli/system-output/rootfs/run/system.json b/test/case/cli/system-output/rootfs/run/system.json new file mode 100644 index 00000000..b95bd5af --- /dev/null +++ b/test/case/cli/system-output/rootfs/run/system.json @@ -0,0 +1 @@ +{"vendor": "QEMU", "product-name": "VM", "part-number": null, "serial-number": null, "mac-address": "02:00:00:00:00:00", "factory-password-hash": "$5$mI/zpOAqZYKLC2WU$i7iPzZiIjOjrBF3NyftS9CCq8dfYwHwrmUK097Jca9A", "vpd": {"product": {"board": "product", "available": false, "trusted": true, "data": {}}}, "usb-ports": [{"name": "USB", "path": "/sys/bus/usb/devices/usb1/authorized"}, {"name": "USB", "path": "/sys/bus/usb/devices/usb1/authorized_default"}, {"name": "USB2", "path": "/sys/bus/usb/devices/usb2/authorized"}, {"name": "USB2", "path": "/sys/bus/usb/devices/usb2/authorized_default"}]} \ No newline at end of file