infamy: Add a simple USB test

test/case/all.yaml

@@ -19,5 +19,7 @@
 - name:  infix-dhcp
   suite: infix_dhcp/all.yaml
 
+- name:  ietf-hardware
+  suite: ietf_hardware/all.yaml
 #- name:  infix-services
 #  suite: infix_services/all.yaml

test/case/ietf_hardware/all.yaml

@@ -0,0 +1,2 @@
+---
+- case: usb.py

test/case/ietf_hardware/usb.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+
+import infamy
+import copy
+import infamy.usb as usb
+from infamy.util import until
+
+with infamy.Test() as test:
+    with test.step("Initialize"):
+        env = infamy.Env(infamy.std_topology("1x1"))
+        target = env.attach("target", "mgmt")
+        available=usb.get_usb_ports(target)
+        if len(available) < 1:
+            test.skip()
+
+    with test.step("Lock USB ports"):
+        components=[]
+        for port in available:
+            component = {
+                "name": port,
+                "state": {
+                    "admin-state": "locked"
+                }
+            }
+            components.append(component)
+
+        target.put_config_dict("ietf-hardware", {
+            "hardware": {
+                "component": components
+            }
+        })
+
+    with test.step("Verify USB ports locked"):
+        for port in available:
+            until(lambda: usb.get_usb_state(target, port) == "locked")
+
+
+    with test.step("Unlock USB ports"):
+        components=[]
+        for port in available:
+            component = {
+                "name": port,
+                "state": {
+                    "admin-state": "unlocked"
+                }
+            }
+            components.append(component)
+
+        target.put_config_dict("ietf-hardware", {
+            "hardware": {
+                "component": components
+            }
+        })
+
+
+    with test.step("Verify USB ports unlocked"):
+        for port in available:
+            until(lambda: usb.get_usb_state(target, port) == "unlocked")
+
+    if len(available) > 1:
+        with test.step("Lock one port"):
+            components=[]
+            component = {
+                "name": available[1],
+                "class": "infix-hardware:usb",
+                "state": {
+                    "admin-state": "locked"
+                }
+            }
+            components.append(component)
+
+        target.put_config_dict("ietf-hardware", {
+            "hardware": {
+                "component": components
+            }
+        })
+        with test.step("Verify one port is locked and one unlocked"):
+            until(lambda: usb.get_usb_state(target, available[1]) == "locked")
+            until(lambda: usb.get_usb_state(target, available[0]) == "unlocked")
+
+    with test.step("Remove all hardware configuration"):
+            running = target.get_config_dict("/ietf-hardware:hardware")
+            new = copy.deepcopy(running)
+            new["hardware"].clear()
+            target.put_diff_dicts("ietf-hardware",running,new)
+
+    with test.step("Verify USB ports locked"):
+        for port in available:
+            until(lambda: usb.get_usb_state(target, port) == "locked")
+
+
+    with test.step("Unlock USB ports"):
+        components=[]
+        for port in available:
+            component = {
+                "name": port,
+                "class": "infix-hardware:usb",
+                "state": {
+                    "admin-state": "unlocked"
+                }
+            }
+            components.append(component)
+
+        target.put_config_dict("ietf-hardware", {
+            "hardware": {
+                "component": components
+            }
+        })
+
+    with test.step("Verify USB ports unlocked"):
+        for port in available:
+            until(lambda: usb.get_usb_state(target, port) == "unlocked")
+
+    test.succeed()

test/infamy/usb.py

@@ -0,0 +1,18 @@
+def _get_hardware(target):
+    xpath="/ietf-hardware:hardware"
+    return target.get_data(xpath)["hardware"]
+
+def get_usb_ports(target):
+    hardware=_get_hardware(target)
+    ports=[]
+    for component in hardware["component"]:
+        if component.get("class") == "infix-hardware:usb":
+            ports.append(component["name"])
+
+    return ports
+
+def get_usb_state(target, name):
+    hardware=_get_hardware(target)
+    for component in hardware["component"]:
+        if component.get("name") == name and component.get("class") == "infix-hardware:usb":
+            return component["state"]["admin-state"]