dasharo/osfv-scripts
0
0
Fork 0
mirror of https://github.com/Dasharo/osfv-scripts.git synced 2026-06-13 19:16:17 -07:00
Code Issues Packages Projects Releases Wiki Activity

*.py: wrap on 80 characters as it in PEP8

Browse Source 
Signed-off-by: Piotr Król <piotr.krol@3mdeb.com>
This commit is contained in:
Piotr Król
2024-12-30 00:25:47 +01:00
parent 1407274fd1
commit 75bec57193
9 changed files with 209 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pre-commit-config.yaml
+1
View File
@@ -59,6 +59,7 @@ repos:
rev: 24.4.2
hooks:
- id: black
args: ["--line-length", "79"]
- repo: https://github.com/MarketSquare/robotframework-tidy
rev: 4.8.1
.robotidy
+1 -1
View File
@@ -1,7 +1,7 @@
spacecount = 4
indent = 4
continuation-indent = 4
line-length = 120
line-length = 80
lineseparator = "unix"
separator = "space"
osfv_cli/src/osfv/cli/cli.py
+97 -33
View File
@@ -49,7 +49,9 @@ def check_in_asset(snipeit_api, asset_id):
# List used assets
def list_used_assets(snipeit_api, args):
all_assets = snipeit_api.get_all_assets()
used_assets = [asset for asset in all_assets if asset["assigned_to"] is not None]
used_assets = [
asset for asset in all_assets if asset["assigned_to"] is not None
]
if not used_assets:
print("No used assets found.")
@@ -73,7 +75,9 @@ def get_my_assets(snipeit_api):
List of assets assigned to the current user
"""
all_assets = snipeit_api.get_all_assets()
used_assets = [asset for asset in all_assets if asset["assigned_to"] is not None]
used_assets = [
asset for asset in all_assets if asset["assigned_to"] is not None
]
return [
asset
for asset in used_assets
@@ -130,7 +134,9 @@ def check_in_my(snipeit_api, args):
return
if not args.yes:
print(f"Are you sure you want to check in {len(my_assets)} assets? [y/N]")
print(
f"Are you sure you want to check in {len(my_assets)} assets? [y/N]"
)
if input() != "y":
print(f"Checking in {len(my_assets)} assets aborted.")
return
@@ -149,7 +155,9 @@ def check_in_my(snipeit_api, args):
# List unused assets
def list_unused_assets(snipeit_api, args):
all_assets = snipeit_api.get_all_assets()
unused_assets = [asset for asset in all_assets if asset["assigned_to"] is None]
unused_assets = [
asset for asset in all_assets if asset["assigned_to"] is None
]
if not unused_assets:
print("No unused assets found.")
@@ -214,7 +222,9 @@ def get_zabbix_compatible_assets_from_asset(asset):
if field_name in ["RTE IP", "Sonoff IP", "PiKVM IP"]:
field_value = field_data.get("value")
if field_value:
key = f'{asset["asset_tag"]}_{field_name}'.replace(" ", "_")
key = f'{asset["asset_tag"]}_{field_name}'.replace(
" ", "_"
)
result[key] = field_value
return result
@@ -422,7 +432,9 @@ def update_zabbix_assets(snipeit_api):
if all_assets:
for asset in all_assets:
snipeit_assets.update(get_zabbix_compatible_assets_from_asset(asset))
snipeit_assets.update(
get_zabbix_compatible_assets_from_asset(asset)
)
snipeit_assets_keys = list(snipeit_assets.keys())
@@ -475,7 +487,9 @@ def update_zabbix_assets(snipeit_api):
snipeit_configuration_error = True
# check for forbidden symbols in asset names
if any(symbol in snipeit_assets_keys[i] for symbol in forbidden_symbols):
if any(
symbol in snipeit_assets_keys[i] for symbol in forbidden_symbols
):
print(
f"{snipeit_assets_keys[i]} contains forbidden symbols! They are going to be changed to '_'."
)
@@ -483,7 +497,9 @@ def update_zabbix_assets(snipeit_api):
for s in forbidden_symbols:
new_key = new_key.replace(s, "_")
snipeit_assets[new_key] = snipeit_assets.pop(snipeit_assets_keys[i])
snipeit_assets[new_key] = snipeit_assets.pop(
snipeit_assets_keys[i]
)
if snipeit_configuration_error:
print(
@@ -505,14 +521,18 @@ def update_zabbix_assets(snipeit_api):
):
update_available = True
common_keys = set(snipeit_assets.keys()) & set(current_zabbix_assets.keys())
common_keys = set(snipeit_assets.keys()) & set(
current_zabbix_assets.keys()
)
if keys_not_present_in_zabbix.__len__() > 0:
print("Assets not present in Zabbix (these will be added):")
print("\n".join(keys_not_present_in_zabbix))
if keys_not_present_in_snipeit.__len__() > 0:
print("\nAssets present in Zabbix but not in SnipeIT (these will be removed):")
print(
"\nAssets present in Zabbix but not in SnipeIT (these will be removed):"
)
print("\n".join(keys_not_present_in_snipeit))
print("")
@@ -560,13 +580,18 @@ def update_zabbix_assets(snipeit_api):
# Main function
def main():
parser = argparse.ArgumentParser(description="Open Source Firmware Validation CLI")
parser = argparse.ArgumentParser(
description="Open Source Firmware Validation CLI"
)
parser.add_argument(
"-v", "--version", action="version", version=metadata.version("osfv")
)
parser.add_argument(
"-j", "--json", action="store_true", help="Output as JSON (if applicable)"
"-j",
"--json",
action="store_true",
help="Output as JSON (if applicable)",
)
subparsers = parser.add_subparsers(
@@ -580,7 +605,9 @@ def main():
# Sonoff subcommands
sonoff_group = sonoff_parser.add_mutually_exclusive_group(required=True)
sonoff_group.add_argument("--sonoff_ip", type=str, help="Sonoff IP address")
sonoff_group.add_argument(
"--sonoff_ip", type=str, help="Sonoff IP address"
)
sonoff_group.add_argument("--rte_ip", type=str, help="RTE IP address")
sonoff_subparsers = sonoff_parser.add_subparsers(
title="subcommands", dest="sonoff_cmd", help="Sonoff subcommands"
@@ -608,7 +635,9 @@ def main():
"list_unused", help="List all unused assets"
)
list_all_parser = snipeit_subparsers.add_parser("list_all", help="List all assets")
list_all_parser = snipeit_subparsers.add_parser(
"list_all", help="List all assets"
)
list_zabbix_parser = snipeit_subparsers.add_parser(
"list_for_zabbix",
@@ -621,13 +650,17 @@ def main():
)
check_out_parser = snipeit_subparsers.add_parser(
"check_out", help="Check out an asset by providing the Asset ID or RTE IP"
"check_out",
help="Check out an asset by providing the Asset ID or RTE IP",
)
check_out_group = check_out_parser.add_mutually_exclusive_group(
required=True
)
check_out_group = check_out_parser.add_mutually_exclusive_group(required=True)
check_out_group.add_argument("--asset_id", type=int, help="Asset ID")
check_out_group.add_argument("--rte_ip", type=str, help="RTE IP")
check_out_parser = snipeit_subparsers.add_parser(
"user_add", help="Add a new user by providing user First Name and Last Name"
"user_add",
help="Add a new user by providing user First Name and Last Name",
)
check_out_parser.add_argument(
"--first-name", type=str, help="User First Name", required=True
@@ -639,7 +672,8 @@ def main():
"--company-name", type=str, default="3mdeb", help="Company Name"
)
check_out_parser = snipeit_subparsers.add_parser(
"user_del", help="Delete new user by providing user First Name and Last Name"
"user_del",
help="Delete new user by providing user First Name and Last Name",
)
check_out_parser.add_argument(
"--first-name", type=str, help="User First Name", required=True
@@ -649,9 +683,12 @@ def main():
)
check_in_parser = snipeit_subparsers.add_parser(
"check_in", help="Check in an asset by providing the Asset ID or RTE IP"
"check_in",
help="Check in an asset by providing the Asset ID or RTE IP",
)
check_in_group = check_in_parser.add_mutually_exclusive_group(
required=True
)
check_in_group = check_in_parser.add_mutually_exclusive_group(required=True)
check_in_group.add_argument("--asset_id", type=int, help="Asset ID")
check_in_group.add_argument("--rte_ip", type=str, help="RTE IP address")
@@ -663,7 +700,9 @@ def main():
)
# RTE subcommands
rte_parser.add_argument("--rte_ip", type=str, help="RTE IP address", required=True)
rte_parser.add_argument(
"--rte_ip", type=str, help="RTE IP address", required=True
)
rte_parser.add_argument(
"--model",
type=str,
@@ -675,15 +714,23 @@ def main():
)
rel_parser = rte_subparsers.add_parser("rel", help="Control RTE relay")
gpio_parser = rte_subparsers.add_parser("gpio", help="Control RTE GPIO")
pwr_parser = rte_subparsers.add_parser("pwr", help="Control DUT power via RTE")
spi_parser = rte_subparsers.add_parser("spi", help="Control SPI lines of RTE")
pwr_parser = rte_subparsers.add_parser(
"pwr", help="Control DUT power via RTE"
)
spi_parser = rte_subparsers.add_parser(
"spi", help="Control SPI lines of RTE"
)
serial_parser = rte_subparsers.add_parser(
"serial", help="Open DUT serial via telnet"
)
flash_parser = rte_subparsers.add_parser("flash", help="DUT flash operations")
flash_parser = rte_subparsers.add_parser(
"flash", help="DUT flash operations"
)
# Power subcommands
pwr_subparsers = pwr_parser.add_subparsers(title="subcommands", dest="pwr_cmd")
pwr_subparsers = pwr_parser.add_subparsers(
title="subcommands", dest="pwr_cmd"
)
power_on_parser = pwr_subparsers.add_parser(
"on", help="Short power button press, to power on DUT"
)
@@ -719,10 +766,14 @@ def main():
)
psu_subparsers.add_parser("on", help="Turn the power supply on")
psu_subparsers.add_parser("off", help="Turn the power supply off")
psu_subparsers.add_parser("get", help="Display information on DUT's power state")
psu_subparsers.add_parser(
"get", help="Display information on DUT's power state"
)
# GPIO subcommands
gpio_subparsers = gpio_parser.add_subparsers(title="subcommands", dest="gpio_cmd")
gpio_subparsers = gpio_parser.add_subparsers(
title="subcommands", dest="gpio_cmd"
)
get_gpio_parser = gpio_subparsers.add_parser("get", help="Get GPIO state")
get_gpio_parser.add_argument("gpio_no", type=int, help="GPIO number")
set_gpio_parser = gpio_subparsers.add_parser("set", help="Set GPIO state")
@@ -730,11 +781,17 @@ def main():
set_gpio_parser.add_argument(
"state", choices=["high", "low", "high-z"], help="GPIO state"
)
set_gpio_parser = gpio_subparsers.add_parser("list", help="List GPIO states")
set_gpio_parser = gpio_subparsers.add_parser(
"list", help="List GPIO states"
)
# Relay subcommands
rel_subparsers = rel_parser.add_subparsers(title="subcommands", dest="rel_cmd")
tgl_rel_parser = rel_subparsers.add_parser("tgl", help="Toggle relay state")
rel_subparsers = rel_parser.add_subparsers(
title="subcommands", dest="rel_cmd"
)
tgl_rel_parser = rel_subparsers.add_parser(
"tgl", help="Toggle relay state"
)
get_rel_parser = rel_subparsers.add_parser("get", help="Get relay state")
set_rel_parser = rel_subparsers.add_parser("set", help="Set relay state")
set_rel_parser.add_argument(
@@ -747,10 +804,15 @@ def main():
)
# RTE SPI subcommands
spi_subparsers = spi_parser.add_subparsers(title="subcommands", dest="spi_cmd")
spi_subparsers = spi_parser.add_subparsers(
title="subcommands", dest="spi_cmd"
)
spi_on_parser = spi_subparsers.add_parser("on", help="Enable SPI lines")
spi_on_parser.add_argument(
"--voltage", type=str, default="1.8V", help="SPI voltage (default: 1.8V)"
"--voltage",
type=str,
default="1.8V",
help="SPI voltage (default: 1.8V)",
)
spi_off_parser = spi_subparsers.add_parser("off", help="Disable SPI lines")
@@ -825,7 +887,9 @@ def main():
elif args.snipeit_cmd == "check_in_my":
check_in_my(snipeit_api, args)
elif args.snipeit_cmd == "user_add":
snipeit_api.user_add(args.first_name, args.last_name, args.company_name)
snipeit_api.user_add(
args.first_name, args.last_name, args.company_name
)
elif args.snipeit_cmd == "user_del":
snipeit_api.user_del(args.first_name, args.last_name)
elif args.snipeit_cmd == "update_zabbix":
osfv_cli/src/osfv/libs/rte.py
+21 -7
View File
@@ -40,21 +40,29 @@ class RTE(rtectrl):
self.sonoff = sonoff
if not self.sonoff_sanity_check():
raise SonoffNotFound(
exit(f"Missing value for 'sonoff_ip' or Sonoff not found in SnipeIT")
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"), "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):
raise UnsupportedDUTModel(f"The {file_path} model is not yet supported")
raise UnsupportedDUTModel(
f"The {file_path} model is not yet supported"
)
# Load the YAML file
with open(file_path, "r") as file:
data = yaml.safe_load(file)
voltage_validator = Any("1.8V", "3.3V")
programmer_name_validator = Any("rte_1_1", "rte_1_0", "ch341a", "dediprog")
programmer_name_validator = Any(
"rte_1_1", "rte_1_0", "ch341a", "dediprog"
)
flashing_power_state_validator = Any("G3", "S5")
schema = Schema(
@@ -69,7 +77,9 @@ class RTE(rtectrl):
Required("pwr_ctrl"): {
Required("sonoff"): bool,
Required("relay"): bool,
Required("flashing_power_state"): flashing_power_state_validator,
Required(
"flashing_power_state"
): flashing_power_state_validator,
},
Optional("reset_cmos", default=False): bool,
Optional("disable_wp", default=False): bool,
@@ -98,7 +108,9 @@ class RTE(rtectrl):
if key in current_field:
current_field = current_field[key]
else:
exit(f"Required field '{field}' is missing in model config.")
exit(
f"Required field '{field}' is missing in model config."
)
# Return the loaded data
return data
@@ -428,7 +440,9 @@ class RTE(rtectrl):
args = self.flash_create_args("--wp-disable --wp-range=0x0,0x0")
self.flash_cmd(args)
if bios:
args = self.flash_create_args(f"-i bios --ifd -w {self.FW_PATH_WRITE}")
args = self.flash_create_args(
f"-i bios --ifd -w {self.FW_PATH_WRITE}"
)
else:
args = self.flash_create_args(f"-w {self.FW_PATH_WRITE}")
rc = self.flash_cmd(args, write_file=write_file)
osfv_cli/src/osfv/libs/rtectrl_api.py
+3 -1
View File
@@ -154,7 +154,9 @@ class rtectrl:
HTTPError: If the response contains an HTTP error status code.
"""
url = BASE_URL_TEMPLATE.format(rte_ip=self.rte_ip)
response = requests.patch(f"{url}{endpoint}", json=data, headers=headers)
response = requests.patch(
f"{url}{endpoint}", json=data, headers=headers
)
response.raise_for_status()
return response
osfv_cli/src/osfv/libs/snipeit_api.py
+51 -15
View File
@@ -68,7 +68,9 @@ class SnipeIT:
break
page += 1
else:
print(f"Error retrieving assets. Status code: {response.status_code}")
print(
f"Error retrieving assets. Status code: {response.status_code}"
)
print(response.json())
break
@@ -209,7 +211,10 @@ class SnipeIT:
)
response_json = response.json()
if response.status_code == 200 and response_json.get("status") != "error":
if (
response.status_code == 200
and response_json.get("status") != "error"
):
return True, response_json, False
else:
return False, response_json, False
@@ -223,17 +228,25 @@ class SnipeIT:
)
response_json = response.json()
if response.status_code == 200 and response_json.get("status") != "error":
if (
response.status_code == 200
and response_json.get("status") != "error"
):
return True, response_json
else:
return False, response_json
def get_asset(self, asset_id):
response = requests.get(
f"{self.cfg_api_url}/hardware/{asset_id}", headers=self.headers, timeout=10
f"{self.cfg_api_url}/hardware/{asset_id}",
headers=self.headers,
timeout=10,
)
response_json = response.json()
if response.status_code == 200 and response_json.get("status") != "error":
if (
response.status_code == 200
and response_json.get("status") != "error"
):
return True, response_json
else:
return False, response_json
@@ -257,7 +270,9 @@ class SnipeIT:
return company["id"]
return None
else:
print(f"Error retrieving companies. Status code: {response.status_code}")
print(
f"Error retrieving companies. Status code: {response.status_code}"
)
print(response.json())
return None
@@ -272,7 +287,9 @@ class SnipeIT:
return group["id"]
return None
else:
print(f"Error retrieving user groups. Status code: {response.status_code}")
print(
f"Error retrieving user groups. Status code: {response.status_code}"
)
print(response.json())
return None
@@ -300,7 +317,9 @@ class SnipeIT:
break
page += 1
else:
print(f"Error retrieving users. Status code: {response.status_code}")
print(
f"Error retrieving users. Status code: {response.status_code}"
)
print(response.json())
break
@@ -316,7 +335,9 @@ class SnipeIT:
def user_add(self, first_name, last_name, company_name):
email = f"{unidecode.unidecode(first_name.lower())}.{unidecode.unidecode(last_name.lower())}@3mdeb.com"
username = f"{first_name[0].lower()}{unidecode.unidecode(last_name.lower())}"
username = (
f"{first_name[0].lower()}{unidecode.unidecode(last_name.lower())}"
)
password = self.generate_password()
users = self.get_users()
@@ -352,10 +373,16 @@ class SnipeIT:
print(data)
response = requests.post(
f"{self.cfg_api_url}/users", headers=self.headers, json=data, timeout=10
f"{self.cfg_api_url}/users",
headers=self.headers,
json=data,
timeout=10,
)
response_json = response.json()
if response.status_code == 200 and response_json.get("status") != "error":
if (
response.status_code == 200
and response_json.get("status") != "error"
):
user_info = response.json()["payload"]
user_id = user_info["id"]
print(f"User created successfully!")
@@ -363,12 +390,16 @@ class SnipeIT:
print(f"Password: {password}")
print(f"User ID: {user_id}")
else:
print(f"Failed to create user. Status code: {response.status_code}")
print(
f"Failed to create user. Status code: {response.status_code}"
)
print(response_json)
def user_del(self, first_name, last_name):
email = f"{unidecode.unidecode(first_name.lower())}.{unidecode.unidecode(last_name.lower())}@3mdeb.com"
username = f"{first_name[0].lower()}{unidecode.unidecode(last_name.lower())}"
username = (
f"{first_name[0].lower()}{unidecode.unidecode(last_name.lower())}"
)
user_id = self.get_user_id(username)
if not user_id:
@@ -376,10 +407,15 @@ class SnipeIT:
return
response = requests.delete(
f"{self.cfg_api_url}/users/{user_id}", headers=self.headers, timeout=10
f"{self.cfg_api_url}/users/{user_id}",
headers=self.headers,
timeout=10,
)
response_json = response.json()
if response.status_code == 200 and response_json.get("status") != "error":
if (
response.status_code == 200
and response_json.get("status") != "error"
):
print(f"User {username} deleted successfully!")
else:
print(
osfv_cli/src/osfv/libs/zabbix.py
+11 -3
View File
@@ -29,7 +29,10 @@ class Zabbix:
payload = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {"user": self.api_username, "password": self.api_password},
"params": {
"user": self.api_username,
"password": self.api_password,
},
"id": 1,
"auth": None,
}
@@ -41,7 +44,9 @@ class Zabbix:
return result["result"]
elif "error" in result:
error_message = result["error"]["message"]
raise ValueError(f"Zabbix API authentication failed: {error_message}")
raise ValueError(
f"Zabbix API authentication failed: {error_message}"
)
else:
raise ValueError("Invalid response from Zabbix API authentication")
@@ -49,7 +54,10 @@ class Zabbix:
payload = {
"jsonrpc": "2.0",
"method": "host.get",
"params": {"selectInterfaces": ["ip"], "output": ["hostid", "host"]},
"params": {
"selectInterfaces": ["ip"],
"output": ["hostid", "host"],
},
"id": 1,
"auth": self.auth_token,
}
osfv_cli/src/osfv/rf/rte_robot.py
+9 -3
View File
@@ -37,7 +37,9 @@ class RobotRTE:
if snipeit:
self.snipeit_api = SnipeIT()
asset_id = self.snipeit_api.get_asset_id_by_rte_ip(rte_ip)
status, dut_model_name = self.snipeit_api.get_asset_model_name(asset_id)
status, dut_model_name = self.snipeit_api.get_asset_model_name(
asset_id
)
if status:
robot.api.logger.info(
f"DUT model retrieved from snipeit: {dut_model_name}"
@@ -51,8 +53,12 @@ class RobotRTE:
)
self.rte = RTE(rte_ip, dut_model_name, self.sonoff)
else:
self.sonoff, self.sonoff_ip = utils.init_sonoff(sonoff_ip, self.rte_ip)
self.rte = RTE(rte_ip, self.cli_model_from_osfv(config), self.sonoff)
self.sonoff, self.sonoff_ip = utils.init_sonoff(
sonoff_ip, self.rte_ip
)
self.rte = RTE(
rte_ip, self.cli_model_from_osfv(config), self.sonoff
)
def cli_model_from_osfv(self, osfv_model):
"""
osfv_cli/src/osfv/rf/sonoff_robot.py
+15 -5
View File
@@ -15,7 +15,9 @@ class Sonoff:
response = self.sonoff.turn_on()
robot.api.logger.info(response)
except requests.exceptions.RequestException as e:
robot.api.logger.info(f"Failed to turn on Sonoff relay. Error: {e}")
robot.api.logger.info(
f"Failed to turn on Sonoff relay. Error: {e}"
)
@keyword(types=None)
def sonoff_off(self):
@@ -24,7 +26,9 @@ class Sonoff:
response = self.sonoff.turn_off()
robot.api.logger.info(response)
except requests.exceptions.RequestException as e:
robot.api.logger.info(f"Failed to turn off Sonoff relay. Error: {e}")
robot.api.logger.info(
f"Failed to turn off Sonoff relay. Error: {e}"
)
@keyword(types=None)
def sonoff_get(self):
@@ -34,7 +38,9 @@ class Sonoff:
state = self.sonoff.get_state()
robot.api.logger.info(f"Sonoff relay state: {state}")
except requests.exceptions.RequestException as e:
robot.api.logger.info(f"Failed to get Sonoff relay state. Error: {e}")
robot.api.logger.info(
f"Failed to get Sonoff relay state. Error: {e}"
)
return state
@keyword(types=None)
@@ -51,6 +57,10 @@ class Sonoff:
response = self.sonoff.turn_on()
robot.api.logger.info("Sonoff relay state toggled on.")
else:
robot.api.logger.info(f"Unexpected Sonoff relay state: {current_state}")
robot.api.logger.info(
f"Unexpected Sonoff relay state: {current_state}"
)
except requests.exceptions.RequestException as e:
robot.api.logger.info(f"Failed to toggle Sonoff relay state. Error: {e}")
robot.api.logger.info(
f"Failed to toggle Sonoff relay state. Error: {e}"
)