From b3fe0d58d875366ad6a6c1955f7be4ad9a22e46a Mon Sep 17 00:00:00 2001
From: Richard Alpe <richard@bit42.se>
Date: Tue, 5 Aug 2025 15:05:19 +0200
Subject: [PATCH] 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 <richard@bit42.se>
---
 src/statd/python/yanger/host.py          | 24 ++++++++++++++++++++++++
 src/statd/python/yanger/ietf_hardware.py | 22 ++++++++++------------
 2 files changed, 34 insertions(+), 12 deletions(-)

diff --git a/src/statd/python/yanger/host.py b/src/statd/python/yanger/host.py
index 1fcceb90..8fd4f01c 100644
--- a/src/statd/python/yanger/host.py
+++ b/src/statd/python/yanger/host.py
@@ -116,6 +116,11 @@ class Localhost(Host):
 
         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 @@ class Remotehost(Localhost):
 
         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 @@ class Replayhost(Host):
                 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:
diff --git a/src/statd/python/yanger/ietf_hardware.py b/src/statd/python/yanger/ietf_hardware.py
index 70a4aa29..94a74f65 100644
--- a/src/statd/python/yanger/ietf_hardware.py
+++ b/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