wip

Signed-off-by: Maciej Pijanowski <[email protected]>

Author: macpijan

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/osfv_cli/rte_robot.py

@@ -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.1"
+name = "osfv"
+version = "0.4.4"
 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"
@@ -21,3 +21,8 @@ requests = "^2.31.0"
 unidecode = "^1.3.6"
 importlib-resources = "^6.1.1"
 voluptuous = "^0.14.2"
+
+# # Include your packages here
+# osfv_cli = {path = "./src/osfv_cli"}
+# osfv_libs = {path = "./src/osfv_libs"}
+# osfv_rf = {path = "./src/osfv_rf"}

osfv_cli/pyproject.toml

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

osfv_cli/osfv_cli/__init__.py osfv_cli/src/osfv/__init__.py

@@ -8,10 +8,10 @@
 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}")
@@ -466,7 +466,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(
@@ -622,8 +622,8 @@ def main():
     set_rel_parser.add_argument(
         "state",
         choices=[
-            "high",
-            "low",
+            "on",
+            "off",
         ],
         help="Relay state",
     )

osfv_cli/src/osfv/__main__.py

@@ -6,8 +6,8 @@
 from importlib_resources import files
 from voluptuous import Any, Optional, Required, Schema
 
-from .rtectrl_api import rtectrl
-from .sonoff_api import SonoffDevice
+from osfv.libs.rtectrl_api import rtectrl
+from osfv.libs.sonoff_api import SonoffDevice
 
 
 class RTE(rtectrl):
@@ -38,7 +38,7 @@ def __init__(self, rte_ip, dut_model, snipeit_api):
         self.sonoff, self.sonoff_ip = self.init_sonoff()
 
     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):
@@ -125,10 +125,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 == 'low':
+            gpio_state = 'off'
+        self.gpio_set(self.GPIO_RELAY, gpio_state)
 
     def reset_cmos(self):
         self.gpio_set(self.GPIO_CMOS, "low")
@@ -163,9 +174,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)
 
@@ -176,9 +187,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)
 

osfv_cli/src/osfv/cli/__init__.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/osfv_cli/osfv_cli.py osfv_cli/src/osfv/cli/cli.py

@@ -0,0 +1,96 @@
+import robot.api.logger
+from robot.api.deco import keyword
+from osfv.libs.rte import RTE
+from osfv.libs.snipeit_api import SnipeIT
+
+
+class RobotRTE:
+    def __init__(self, rte_ip):
+        snipeit_api = SnipeIT()
+        asset_id = snipeit_api.get_asset_id_by_rte_ip(rte_ip)
+        status, dut_model_name = snipeit_api.get_asset_model_name(asset_id)
+        if status:
+            robot.api.logger.info(f"DUT model retrieved from snipeit: {dut_model_name}")
+        else:
+            raise AssertionError(
+                f"Failed to retrieve model name from Snipe-IT. Check again arguments, or try providing model manually."
+            )
+        self.rte = RTE(rte_ip, dut_model_name, snipeit_api)
+
+    @keyword(types=None)
+    def rte_flash_read(self, fw_file):
+        """Reads DUT flash chip content into ``fw_file``  path
+        """
+        robot.api.logger.info(f"Reading from flash...")
+        self.rte.flash_read(fw_file)
+        robot.api.logger.info(f"Read flash content saved to {fw_file}")
+
+    @keyword(types=None)
+    def rte_flash_write(self, fw_file):
+        """Writes file from ``fw_file`` path into DUT flash chip
+        """
+        robot.api.logger.info(f"Writing {fw_file} to flash...")
+        self.rte.flash_write(fw_file)
+        robot.api.logger.info(f"Flash written")
+
+    @keyword(types=None)
+    def rte_flash_probe(self):
+        robot.api.logger.info(f"Probing flash...")
+        self.rte.flash_probe()
+    
+    @keyword(types=None)
+    def rte_flash_erase(self):
+        robot.api.logger.info(f"Erasing DUT flash...")
+        self.rte.flash_erase()
+        robot.api.logger.info(f"Flash erased")
+
+    @keyword(types=None)
+    def rte_relay_toggle(self):
+        state_str = self.rte.relay_get()
+        if state_str == "low":
+            new_state_str = "high"
+        else:
+            new_state_str = "low"
+        self.rte.relay_set(new_state_str)
+        state = self.rte.relay_get()
+        robot.api.logger.info(f"Relay state toggled. New state: {state}")
+    
+    @keyword(types=None)
+    def rte_relay_set(self, state):
+        self.rte.relay_set(state)
+        state = self.rte.relay_get()
+        robot.api.logger.info(f"Relay state set to {state}")
+        return state
+    
+    @keyword(types=None)
+    def rte_relay_get(self):
+        state = self.rte.relay_get()
+        robot.api.logger.info(f"Relay state: {state}")
+        return state
+
+    @keyword(types=None)
+    def rte_power_on(self, time=1):
+        robot.api.logger.info(f"Powering on...")
+        self.rte.power_on(time)
+    
+    @keyword(types=None)
+    def rte_power_off(self, time=6):
+        robot.api.logger.info(f"Powering off...")
+        self.rte.power_off(time)
+    
+    @keyword(types=None)
+    def rte_reset(self, time=1):
+        robot.api.logger.info(f"Pressing reset button...")
+        self.rte.reset(time)
+    
+    @keyword(types=None)
+    def rte_gpio_get(self, gpio_no):
+        state = self.rte.gpio_get(gpio_no)
+        robot.api.logger.info(f"GPIO {gpio_no} state: {state}")
+        return state
+    
+    @keyword(types=None)
+    def rte_gpio_set(self, gpio_no, state):
+        self.rte.gpio_set(gpio_no, state)
+        state = self.rte.gpio_get(gpio_no)
+        robot.api.logger.info(f"GPIO {gpio_no} state set to {state}")

osfv_cli/src/osfv/libs/__init__.py

@@ -1,9 +1,8 @@
 import robot.api.logger
-from snipeit_api import SnipeIT
+from osfv.libs.snipeit_api import SnipeIT
 
 snipeit_api = SnipeIT()
 
-
 def snipeit_checkout(rte_ip):
     asset_id = snipeit_api.get_asset_id_by_rte_ip(rte_ip)
     success, data = snipeit_api.check_out_asset(asset_id)
@@ -15,7 +14,6 @@ def snipeit_checkout(rte_ip):
             f"Error checking out asset {asset_id}. Response data: {data}"
         )
 
-
 def snipeit_checkin(rte_ip):
     asset_id = snipeit_api.get_asset_id_by_rte_ip(rte_ip)
     success, data = snipeit_api.check_in_asset(asset_id)
@@ -26,3 +24,20 @@ def snipeit_checkin(rte_ip):
         raise AssertionError(
             f"Error checking in asset {asset_id}. Response data: {data}"
         )
+
+def snipeit_get_sonoff_ip(rte_ip):
+    return snipeit_api.get_sonoff_ip_by_rte_ip(rte_ip)
+
+def snipeit_get_pikvm_ip(rte_ip):
+    return snipeit_api.get_pikvm_ip_by_rte_ip(rte_ip)
+
+def snipeit_get_asset_model(rte_ip):
+    asset_id = snipeit_api.get_asset_id_by_rte_ip(rte_ip)
+    success, data = snipeit_api.get_asset_model_name(asset_id)
+    if success:
+        robot.api.logger.info(f"Asset {asset_id} model is: {data} in.")
+        return data
+    else:
+        raise AssertionError(
+                f"Error getting model name of asset: {asset_id}. Response data: {data}"
+        )