osfv integration

.codespellx

@@ -0,0 +1 @@
+            f"{self.cfg_api_url}/hardware/{asset_id}/checkin",

osfv_cli/Makefile

@@ -8,7 +8,7 @@ build:
 	poetry build
 
 install: build
-	pip install dist/$(PACKAGE_NAME)-$(VERSION)-py3-none-any.whl
+	pip install --force-reinstall dist/$(PACKAGE_NAME)-$(VERSION)-py3-none-any.whl
 	test -f $(SNIPEIT_CONFIG_FILE) || install -D -m 644 snipeit.yml $(SNIPEIT_CONFIG_FILE)
 	test -f $(ZABBIX_CONFIG_FILE) || install -D -m 644 zabbix.yml $(ZABBIX_CONFIG_FILE)
 

osfv_cli/pyproject.toml

@@ -3,14 +3,14 @@ requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry.scripts]
-osfv_cli = "osfv_cli.osfv_cli:main"
+osfv_cli = "osfv.cli.cli:main"
 
 [tool.poetry]
-name = "osfv_cli"
-version = "0.4.4"
+name = "osfv"
+version = "0.5.0"
 description = "Open Source Firmware Validation Command Line Interface Tool"
 authors = ["Maciej Pijanowski <[email protected]>"]
-include = ["osfv_cli/models/*.yml"]
+include = ["src/models/*.yml"]
 
 [tool.poetry.dependencies]
 python = "^3.8"

osfv_cli/src/osfv/__main__.py

@@ -0,0 +1,6 @@
+import sys
+
+if __name__ == "__main__":
+    from osfv.osfv_cli.osfv_cli import main
+
+    sys.exit(main())

osfv_cli/osfv_cli/osfv_cli.py → osfv_cli/src/osfv/cli/cli.py

@@ -5,13 +5,13 @@
 from copy import copy
 from importlib import metadata
 
+import osfv.libs.utils as utils
 import pexpect
 import requests
-
-from .rte import RTE
-from .snipeit_api import SnipeIT
-from .sonoff_api import SonoffDevice
-from .zabbix import Zabbix
+from osfv.libs.rte import RTE
+from osfv.libs.snipeit_api import SnipeIT
+from osfv.libs.sonoff_api import SonoffDevice
+from osfv.libs.zabbix import Zabbix
 
 
 # Check out an asset
@@ -155,10 +155,10 @@ def print_asset_details_for_zabbix(asset):
 
 def relay_toggle(rte, args):
     state_str = rte.relay_get()
-    if state_str == "low":
-        new_state_str = "high"
+    if state_str == "off":
+        new_state_str = "on"
     else:
-        new_state_str = "low"
+        new_state_str = "off"
     rte.relay_set(new_state_str)
     state = rte.relay_get()
     print(f"Relay state toggled. New state: {state}")
@@ -465,7 +465,7 @@ def update_zabbix_assets(snipeit_api):
 def main():
     parser = argparse.ArgumentParser(description="Open Source Firmware Validation CLI")
     parser.add_argument(
-        "-v", "--version", action="version", version=metadata.version("osfv_cli")
+        "-v", "--version", action="version", version=metadata.version("osfv")
     )
 
     parser.add_argument(
@@ -621,8 +621,8 @@ def main():
     set_rel_parser.add_argument(
         "state",
         choices=[
-            "high",
-            "low",
+            "on",
+            "off",
         ],
         help="Relay state",
     )
@@ -717,7 +717,8 @@ def main():
                 exit(
                     f"Failed to retrieve model name from Snipe-IT. Check again arguments, or try providing model manually."
                 )
-        rte = RTE(args.rte_ip, dut_model_name, snipeit_api)
+        sonoff, sonoff_ip = utils.init_sonoff(None, args.rte_ip, snipeit_api)
+        rte = RTE(args.rte_ip, dut_model_name, sonoff)
 
         if args.rte_cmd == "rel":
             # Handle RTE relay related commands

osfv_cli/osfv_cli/rte.py → osfv_cli/src/osfv/libs/rte.py

@@ -4,11 +4,9 @@
 import paramiko
 import yaml
 from importlib_resources import files
+from osfv.libs.rtectrl_api import rtectrl
 from voluptuous import Any, Optional, Required, Schema
 
-from .rtectrl_api import rtectrl
-from .sonoff_api import SonoffDevice
-
 
 class RTE(rtectrl):
     GPIO_SPI_ON = 1
@@ -30,15 +28,18 @@ class RTE(rtectrl):
     PROGRAMMER_CH341A = "ch341a_spi"
     FLASHROM_CMD = "flashrom -p {programmer} {args}"
 
-    def __init__(self, rte_ip, dut_model, snipeit_api):
+    def __init__(self, rte_ip, dut_model, sonoff):
         self.rte_ip = rte_ip
         self.dut_model = dut_model
         self.dut_data = self.load_model_data()
-        self.snipeit_api = snipeit_api
-        self.sonoff, self.sonoff_ip = self.init_sonoff()
+        self.sonoff = sonoff
+        if not self.sonoff_sanity_check():
+            raise SonoffNotFound(
+                exit(f"Missing value for 'sonoff_ip' or Sonoff not found in SnipeIT")
+            )
 
     def load_model_data(self):
-        file_path = os.path.join(files("osfv_cli"), "models", f"{self.dut_model}.yml")
+        file_path = os.path.join(files("osfv"), "models", f"{self.dut_model}.yml")
 
         # Check if the file exists
         if not os.path.isfile(file_path):
@@ -99,20 +100,6 @@ def load_model_data(self):
         # Return the loaded data
         return data
 
-    def init_sonoff(self):
-        sonoff_ip = ""
-        sonoff = None
-
-        if self.dut_data["pwr_ctrl"]["sonoff"] is True:
-            sonoff_ip = self.snipeit_api.get_sonoff_ip_by_rte_ip(self.rte_ip)
-            if not sonoff_ip:
-                raise SonoffNotFound(
-                    exit(f"Sonoff IP not found in SnipeIT for RTE: {self.rte_ip}")
-                )
-            sonoff = SonoffDevice(sonoff_ip)
-
-        return sonoff, sonoff_ip
-
     def power_on(self, sleep=1):
         self.gpio_set(self.GPIO_POWER, "low", sleep)
         time.sleep(sleep)
@@ -126,10 +113,21 @@ def reset(self, sleep=1):
         time.sleep(sleep)
 
     def relay_get(self):
-        return self.gpio_get(self.GPIO_RELAY)
-
-    def relay_set(self, state):
-        self.gpio_set(self.GPIO_RELAY, state)
+        gpio_state = self.gpio_get(self.GPIO_RELAY)
+        relay_state = None
+        if gpio_state == "high":
+            relay_state = "on"
+        if gpio_state == "low":
+            relay_state = "off"
+        return relay_state
+
+    def relay_set(self, relay_state):
+        gpio_state = None
+        if relay_state == "on":
+            gpio_state = "high"
+        if relay_state == "off":
+            gpio_state = "low"
+        self.gpio_set(self.GPIO_RELAY, gpio_state)
 
     def reset_cmos(self):
         self.gpio_set(self.GPIO_CMOS, "low")
@@ -164,9 +162,9 @@ def pwr_ctrl_on(self):
             if state != "ON":
                 raise Exception("Failed to power control ON")
         elif self.dut_data["pwr_ctrl"]["relay"] is True:
-            self.relay_set("high")
+            self.relay_set("on")
             state = self.relay_get()
-            if state != "high":
+            if state != "on":
                 raise Exception("Failed to power control ON")
         time.sleep(5)
 
@@ -177,9 +175,9 @@ def pwr_ctrl_off(self):
             if state != "OFF":
                 raise Exception("Failed to power control OFF")
         elif self.dut_data["pwr_ctrl"]["relay"] is True:
-            self.relay_set("low")
+            self.relay_set("off")
             state = self.relay_get()
-            if state != "low":
+            if state != "off":
                 raise Exception("Failed to power control OFF")
         time.sleep(2)
 
@@ -309,6 +307,12 @@ def flash_write(self, write_file):
             if self.dut_data["reset_cmos"] == True:
                 self.reset_cmos()
 
+    def sonoff_sanity_check(self):
+        """
+        Verify that if DUT is powered by Sonoff, Sonoff IP is not None
+        """
+        return not self.dut_data["pwr_ctrl"]["sonoff"] or self.sonoff.sonoff_ip
+
 
 class IncompleteModelData(Exception):
     pass

osfv_cli/osfv_cli/snipeit_api.py → osfv_cli/src/osfv/libs/snipeit_api.py

@@ -117,6 +117,29 @@ def get_sonoff_ip_by_rte_ip(self, rte_ip):
         # No asset found with matching RTE IP
         return None
 
+    def get_pikvm_ip_by_rte_ip(self, rte_ip):
+        # Retrieve all assets
+        all_assets = self.get_all_assets()
+
+        # Search for asset with matching RTE IP
+        for asset in all_assets:
+            custom_fields = asset.get("custom_fields", {})
+            if custom_fields:
+                rte_ip_field = next(
+                    (
+                        field_data["value"]
+                        for field_name, field_data in custom_fields.items()
+                        if field_name == "RTE IP"
+                    ),
+                    None,
+                )
+                if rte_ip_field == rte_ip:
+                    if custom_fields["PiKVM IP"]:
+                        return custom_fields["PiKVM IP"]["value"]
+
+        # No asset found with matching PiKVM IP
+        return None
+
     # Check out an asset
     def check_out_asset(self, asset_id):
         status, asset_data = self.get_asset(asset_id)

osfv_cli/src/osfv/libs/utils.py

@@ -0,0 +1,12 @@
+from osfv.libs.sonoff_api import SonoffDevice
+
+
+def init_sonoff(init_sonoff_ip, rte_ip, snipeit_api=None):
+    sonoff_ip = ""
+    sonoff = None
+    if not snipeit_api:
+        sonoff_ip = init_sonoff_ip
+    else:
+        sonoff_ip = snipeit_api.get_sonoff_ip_by_rte_ip(rte_ip)
+    sonoff = SonoffDevice(sonoff_ip)
+    return sonoff, sonoff_ip

osfv_cli/osfv_cli/models/Optiplex 7010.yml → osfv_cli/src/osfv/models/Optiplex 7010.yml

@@ -14,7 +14,7 @@ pwr_ctrl:
   sonoff: true
   # whether power is controller via on-board RTE relay (required)
   relay: false
-  init_on: false
+  flashing_power_state: "OFF"
 
 # whether CMOS reset is required after flashing (optional - defaults to false)
 reset_cmos: false