diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ad9bde3..3684e68 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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 diff --git a/.robotidy b/.robotidy index b660836..539af78 100644 --- a/.robotidy +++ b/.robotidy @@ -1,7 +1,7 @@ spacecount = 4 indent = 4 continuation-indent = 4 -line-length = 120 +line-length = 80 lineseparator = "unix" separator = "space" diff --git a/osfv_cli/src/osfv/cli/cli.py b/osfv_cli/src/osfv/cli/cli.py index d882835..b891c8e 100755 --- a/osfv_cli/src/osfv/cli/cli.py +++ b/osfv_cli/src/osfv/cli/cli.py @@ -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": diff --git a/osfv_cli/src/osfv/libs/rte.py b/osfv_cli/src/osfv/libs/rte.py index fb44ae3..540b229 100644 --- a/osfv_cli/src/osfv/libs/rte.py +++ b/osfv_cli/src/osfv/libs/rte.py @@ -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) diff --git a/osfv_cli/src/osfv/libs/rtectrl_api.py b/osfv_cli/src/osfv/libs/rtectrl_api.py index b3db3dd..2bb71ed 100644 --- a/osfv_cli/src/osfv/libs/rtectrl_api.py +++ b/osfv_cli/src/osfv/libs/rtectrl_api.py @@ -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 diff --git a/osfv_cli/src/osfv/libs/snipeit_api.py b/osfv_cli/src/osfv/libs/snipeit_api.py index 2bb623b..16552fb 100644 --- a/osfv_cli/src/osfv/libs/snipeit_api.py +++ b/osfv_cli/src/osfv/libs/snipeit_api.py @@ -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( diff --git a/osfv_cli/src/osfv/libs/zabbix.py b/osfv_cli/src/osfv/libs/zabbix.py index 4e0f94b..78c7147 100644 --- a/osfv_cli/src/osfv/libs/zabbix.py +++ b/osfv_cli/src/osfv/libs/zabbix.py @@ -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, } diff --git a/osfv_cli/src/osfv/rf/rte_robot.py b/osfv_cli/src/osfv/rf/rte_robot.py index 866217a..56b73f2 100644 --- a/osfv_cli/src/osfv/rf/rte_robot.py +++ b/osfv_cli/src/osfv/rf/rte_robot.py @@ -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): """ diff --git a/osfv_cli/src/osfv/rf/sonoff_robot.py b/osfv_cli/src/osfv/rf/sonoff_robot.py index 1402e9d..2a3450c 100644 --- a/osfv_cli/src/osfv/rf/sonoff_robot.py +++ b/osfv_cli/src/osfv/rf/sonoff_robot.py @@ -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}" + )