yager: handle usb info though the HOST api

Prior to this commit yanger looked at the running system using
os.path..., this meant that the static reply / capture data wasn't
used properly. This resulted in strange behavior during unit testing
on GitHub runners where USB ports are missing.

Signed-off-by: Richard Alpe <[email protected]>

Author: rical

src/statd/python/yanger/host.py

@@ -116,6 +116,11 @@ def read(self, path):
 
         return None
 
+    def exists(self, path: str) -> bool:
+        try:
+            return os.path.exists(path)
+        except OSError:
+            return False
 
 class Remotehost(Localhost):
     def __init__(self, prefix, capdir):
@@ -159,6 +164,18 @@ def run(self, cmd, default=None, log=True):
 
         return out
 
+    def exists(self, path: str) -> bool:
+        if not self._run(("ls", path), default="", log=False):
+            return False
+
+        if self.capdir:
+            dirname = os.path.join(self.capdir, "rootfs", os.path.dirname(path[1:]))
+            filname = os.path.join(self.capdir, "rootfs", path[1:])
+            os.makedirs(dirname, exist_ok=True)
+            open(filname, "w", encoding='utf-8').close()  # Create empty file
+
+        return True
+
     def read(self, path):
         out = self._run(("cat", path), default="", log=False)
 
@@ -199,6 +216,13 @@ def run(self, cmd, default=None, log=True):
                 common.LOG.error(f"No recording found for run \"{path}\"")
             raise
 
+    def exists(self, path: str) -> bool:
+        path = os.path.join(self.replaydir, "rootfs", path[1:])
+        try:
+            return os.path.exists(path)
+        except OSError:
+            return False
+
     def read(self, path):
         path = os.path.join(self.replaydir, "rootfs", path[1:])
         try:

src/statd/python/yanger/ietf_hardware.py

@@ -61,18 +61,16 @@ def usb_port_components(systemjson):
 
             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)
+                if HOST.exists(path):
+                    names.append(usb_port["name"])
+                    data = int(HOST.read(path))
+                    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