1430 lines
44 KiB
Python
1430 lines
44 KiB
Python
###
|
|
# Copyright 2017-2025 Hewlett Packard Enterprise, Inc. All rights reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
###
|
|
|
|
# -*- coding: utf-8 -*-
|
|
"""This is the helper module for RDMC"""
|
|
|
|
# ---------Imports---------
|
|
from __future__ import unicode_literals
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
from ctypes import byref, c_char_p, create_string_buffer
|
|
|
|
try:
|
|
from logging_config_path import get_logging_config_path
|
|
except ModuleNotFoundError:
|
|
from ilorest.logging_config_path import get_logging_config_path
|
|
|
|
|
|
import pyaes
|
|
import six
|
|
from prompt_toolkit.completion import Completer, Completion
|
|
|
|
import redfish.hpilo.risblobstore2 as risblobstore2
|
|
import redfish.ris
|
|
import logging.config
|
|
|
|
try:
|
|
import versioning
|
|
except ImportError:
|
|
from ilorest import versioning
|
|
|
|
if os.name == "nt":
|
|
from six.moves import winreg
|
|
|
|
if six.PY2:
|
|
from win32con import HKEY_LOCAL_MACHINE
|
|
elif six.PY3:
|
|
from win32.lib.win32con import HKEY_LOCAL_MACHINE
|
|
|
|
# ---------End of imports---------
|
|
|
|
|
|
# Using hard coded list until better solution is found
|
|
HARDCODEDLIST = [
|
|
"modified",
|
|
"attributeregistry",
|
|
"links",
|
|
"settingsresult",
|
|
"actions",
|
|
"availableactions",
|
|
"extref",
|
|
]
|
|
|
|
# ---------Debug logger---------
|
|
# Centralized logging behaviour:
|
|
# - The JSON config in ilorest/logging_config.json defines handlers/formatters.
|
|
# - `setup_logging_from_json()` loads that config and uses `prepare_log_directory()` to
|
|
# resolve and create a writable `logdir` (trying the configured path first and then
|
|
# platform-specific fallbacks). This centralizes all directory-selection logic.
|
|
# - `setup_fallback_logging()` is to support fallback logging if the JSON config
|
|
# cannot be loaded or applied for some reason. It uses command line flags to retain old functionality.
|
|
|
|
|
|
class InfoFilter(logging.Filter):
|
|
"""Filter to allow only INFO and WARNING messages, blocking ERROR and above"""
|
|
|
|
def filter(self, record):
|
|
# Allow only records below ERROR (i.e., INFO and WARNING)
|
|
return record.levelno < logging.ERROR
|
|
|
|
|
|
# Logger instance - configuration will be handled by JSON config file
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def clear_existing_handlers():
|
|
"""Clear any existing log handlers to avoid conflicts."""
|
|
root_logger = logging.getLogger()
|
|
for handler in root_logger.handlers[:]:
|
|
root_logger.removeHandler(handler)
|
|
|
|
|
|
def setup_fallback_logging(reason="Unknown error", opts=None):
|
|
"""
|
|
Fallback logging setup when all else fails.
|
|
|
|
Args:
|
|
reason: The reason for falling back to fallback logging
|
|
opts: Optional command line options that may contain logging flags
|
|
"""
|
|
try:
|
|
clear_existing_handlers()
|
|
|
|
# Extract command line flags from opts or sys.argv
|
|
debug_flag = False
|
|
noinfo_flag = False
|
|
nostdout_flag = False
|
|
log_dir = os.path.join(os.getcwd(), "ilorest_logs")
|
|
|
|
if opts:
|
|
debug_flag = getattr(opts, "debug", False)
|
|
noinfo_flag = getattr(opts, "noinfolog", False)
|
|
nostdout_flag = getattr(opts, "nostdoutlog", False)
|
|
custom_logdir = getattr(opts, "logdir", None)
|
|
if custom_logdir:
|
|
log_dir = os.path.normpath(custom_logdir)
|
|
else:
|
|
# Check sys.argv as fallback
|
|
debug_flag = any(arg in ["-d", "--debug"] for arg in sys.argv)
|
|
noinfo_flag = any(arg in ["--noinfolog"] for arg in sys.argv)
|
|
nostdout_flag = any(arg in ["--nostdoutlog"] for arg in sys.argv)
|
|
|
|
# Create log directory
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
log_file_path = os.path.join(log_dir, "iLORest.log")
|
|
|
|
# Determine logging level (debug takes precedence over noinfo)
|
|
if noinfo_flag and not debug_flag:
|
|
file_level = logging.ERROR
|
|
elif debug_flag:
|
|
file_level = logging.DEBUG
|
|
else:
|
|
file_level = logging.INFO
|
|
|
|
# Configure file handler
|
|
file_handler = logging.FileHandler(log_file_path)
|
|
file_handler.setLevel(file_level)
|
|
file_formatter = logging.Formatter("%(levelname)s\t: %(message)s")
|
|
file_handler.setFormatter(file_formatter)
|
|
|
|
handlers = [file_handler]
|
|
|
|
# Configure stdout handler so that INFO (and below) do NOT go to stdout.
|
|
# The console should show CRITICAL by default, but if the
|
|
# global file level is higher (e.g. ERROR due to --noinfolog), keep
|
|
# stdout at that higher level to respect the flag.
|
|
if not nostdout_flag:
|
|
stdout_handler = logging.StreamHandler(sys.stdout)
|
|
# If user requested debug output on the command line, allow stdout to
|
|
# receive DEBUG messages. Otherwise keep stdout quieter (CRITICAL)
|
|
if debug_flag:
|
|
stdout_level = logging.DEBUG
|
|
else:
|
|
# set to default CRITICAL
|
|
stdout_level = logging.CRITICAL
|
|
stdout_handler.setLevel(stdout_level)
|
|
stdout_handler.setFormatter(file_formatter)
|
|
handlers.append(stdout_handler)
|
|
|
|
# Apply configuration to the root logger
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(file_level)
|
|
|
|
# Clear any existing handlers (already done above) then add ours
|
|
for h in handlers:
|
|
root_logger.addHandler(h)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.debug("Fallback to basic configurations for logging: %s (logdir=%s)", reason, log_dir)
|
|
return True
|
|
|
|
except Exception as fallback_error:
|
|
# If even fallback setup failed, write to stderr directly
|
|
sys.stderr.write(f"Critical: All fallback logging attempts failed: {reason} - {fallback_error}\n")
|
|
return False
|
|
|
|
|
|
def load_config_file(config_path):
|
|
"""Load and validate the JSON configuration file."""
|
|
if not os.path.isfile(config_path):
|
|
raise FileNotFoundError(f"JSON config file not found: {config_path}")
|
|
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
config = json.load(f)
|
|
|
|
# Validate basic structure
|
|
if not isinstance(config, dict):
|
|
raise ValueError("Config file must contain a JSON object")
|
|
|
|
if "handlers" not in config:
|
|
raise ValueError("Config file must contain 'handlers' section")
|
|
|
|
return config
|
|
|
|
|
|
def prepare_log_directory(config):
|
|
"""Determine and prepare a writable log directory based on config and platform.
|
|
|
|
Behavior (matching historical implementation):
|
|
- Try the JSON-configured path first (absolute).
|
|
- If that fails, fall back to a single directory:
|
|
- On Windows: use APPDATA if set
|
|
- Otherwise: use ~/.local/share
|
|
- Return the directory that is writable; raise PermissionError if neither works.
|
|
"""
|
|
primary = config.get("logdir", "./ilorest_logs")
|
|
if not os.path.isabs(primary):
|
|
primary = os.path.abspath(primary)
|
|
|
|
def _writable_dir(path):
|
|
try:
|
|
os.makedirs(path, exist_ok=True)
|
|
testfile = os.path.join(path, ".write_test")
|
|
with open(testfile, "w") as fh:
|
|
fh.write("ok")
|
|
try:
|
|
os.remove(testfile)
|
|
except Exception:
|
|
pass
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# Try primary configured path first
|
|
if _writable_dir(primary):
|
|
LOGGER.debug("Selected logdir (configured): %s", os.path.normpath(primary))
|
|
return os.path.normpath(primary)
|
|
|
|
# Single fallback to APPDATA or ~/.local/share (matching legacy behavior)
|
|
appdata = os.getenv("APPDATA")
|
|
if appdata:
|
|
fallback_dir = appdata
|
|
else:
|
|
fallback_dir = os.path.join(os.path.expanduser("~"), ".local", "share")
|
|
|
|
fallback_dir = os.path.join(fallback_dir, "ilorest_logs")
|
|
|
|
if _writable_dir(fallback_dir):
|
|
LOGGER.debug("Selected logdir (fallback): %s", os.path.normpath(fallback_dir))
|
|
return os.path.normpath(fallback_dir)
|
|
|
|
# Nothing worked
|
|
raise PermissionError(f"Unable to create or write to configured logdir '{primary}' or fallback '{fallback_dir}'")
|
|
|
|
|
|
def configure_file_handler(config, logdir):
|
|
"""Configure the file handler with proper path joining and validation."""
|
|
file_handler = config.get("handlers", {}).get("file", {})
|
|
if not file_handler:
|
|
return
|
|
|
|
original_filename = file_handler.get("filename", "iLORest.log")
|
|
if "%(logdir)s" in original_filename:
|
|
filename = original_filename.replace("%(logdir)s", logdir)
|
|
else:
|
|
filename = os.path.join(logdir, "iLORest.log")
|
|
|
|
file_handler["filename"] = os.path.normpath(filename)
|
|
|
|
# Test if we can write to the log file
|
|
try:
|
|
with open(file_handler["filename"], "a") as test_file:
|
|
pass # Just test if we can open for writing
|
|
except (IOError, OSError, PermissionError) as e:
|
|
raise ValueError(f"Cannot write to log file {file_handler['filename']}: {e}")
|
|
|
|
|
|
def set_logging_config_with_flags(config, nostdout=False, noinfo=False, debug=False):
|
|
"""Modify logging configuration based on command-line flags.
|
|
|
|
:param config: The logging configuration dictionary to modify
|
|
:type config: dict
|
|
:param nostdout: Flag to suppress stdout logging
|
|
:type nostdout: bool
|
|
:param noinfo: Flag to suppress info level logging
|
|
:type noinfo: bool
|
|
:param debug: Flag to enable debug level logging
|
|
:type debug: bool
|
|
:returns: Modified configuration dictionary
|
|
:rtype: dict
|
|
"""
|
|
# Handle priority: debug takes precedence over noinfo when both are set
|
|
if noinfo and debug:
|
|
noinfo = False
|
|
|
|
# Modify config based on early command-specific flags
|
|
if nostdout and "handlers" in config and "stdout" in config["handlers"]:
|
|
# Remove stdout handler from the configuration
|
|
del config["handlers"]["stdout"]
|
|
# Remove stdout from all logger handlers lists
|
|
for logger_name, logger_config in config.get("loggers", {}).items():
|
|
if "handlers" in logger_config and "stdout" in logger_config["handlers"]:
|
|
logger_config["handlers"] = [h for h in logger_config["handlers"] if h != "stdout"]
|
|
|
|
# Modify logging levels based on flags
|
|
if noinfo:
|
|
level = "ERROR"
|
|
elif debug:
|
|
level = "DEBUG"
|
|
else:
|
|
level = None
|
|
|
|
if level:
|
|
# Apply level to all loggers
|
|
for logger_name, logger_config in config.get("loggers", {}).items():
|
|
logger_config["level"] = level
|
|
# Also apply to handlers if needed
|
|
for handler_name, handler_config in config.get("handlers", {}).items():
|
|
if handler_name != "stderr": # Keep stderr at ERROR level
|
|
handler_config["level"] = level
|
|
|
|
return config
|
|
|
|
|
|
def setup_logging_from_json(opts=None):
|
|
"""Setup logging from JSON config file using dictConfig with comprehensive fallback.
|
|
|
|
This function attempts to load logging configuration from a JSON file and falls back
|
|
to existing configuration if the JSON config fails for any reason.
|
|
If `opts` is provided, it extracts logging flags from the options object.
|
|
|
|
"""
|
|
config_path = get_logging_config_path()
|
|
|
|
try:
|
|
# Load and validate configuration file
|
|
config = load_config_file(config_path)
|
|
|
|
# Extract logging flags from opts if provided and apply_flags is True
|
|
if opts:
|
|
nostdout = getattr(opts, "nostdoutlog", False)
|
|
noinfo = getattr(opts, "noinfolog", False)
|
|
debug = getattr(opts, "debug", False)
|
|
logdir = getattr(opts, "logdir", None)
|
|
|
|
# If caller passed an explicit logdir, use it to override the config value
|
|
if logdir:
|
|
# normalize and set
|
|
config["logdir"] = os.path.normpath(logdir)
|
|
|
|
# Modify config based on flags using the dedicated function
|
|
config = set_logging_config_with_flags(config, nostdout, noinfo, debug)
|
|
|
|
# Prepare log directory (this will create it if missing)
|
|
final_logdir = prepare_log_directory(config)
|
|
|
|
# Configure file handler paths (will replace %(logdir)s etc.)
|
|
configure_file_handler(config, final_logdir)
|
|
|
|
# Clear any existing handlers to avoid conflicts
|
|
clear_existing_handlers()
|
|
|
|
# Apply the logging configuration
|
|
logging.config.dictConfig(config)
|
|
|
|
# Test that logging is working (only if not suppressing info logs and flags applied)
|
|
if opts and not getattr(opts, "noinfolog", False):
|
|
logger = logging.getLogger(__name__)
|
|
logger.debug("Logging configuration loaded successfully from JSON (logdir=%s)", final_logdir)
|
|
except FileNotFoundError as e:
|
|
setup_fallback_logging(str(e), opts)
|
|
return None
|
|
except (PermissionError, IOError) as e:
|
|
setup_fallback_logging(f"File access error: {e}", opts)
|
|
return None
|
|
except json.JSONDecodeError as e:
|
|
setup_fallback_logging(f"Invalid JSON format: {e}", opts)
|
|
return None
|
|
except (KeyError, ValueError, TypeError) as e:
|
|
setup_fallback_logging(f"Invalid configuration structure: {e}", opts)
|
|
return None
|
|
except Exception as e:
|
|
setup_fallback_logging(f"Unexpected error loading config: {e}", opts)
|
|
return None
|
|
|
|
|
|
# ---------End of debug logger---------
|
|
|
|
|
|
class ReturnCodes(object):
|
|
"""Return code class to be used by all functions"""
|
|
|
|
SUCCESS = 0
|
|
|
|
# ****** RDMC ERRORS ******
|
|
CONFIGURATION_FILE_ERROR = 1
|
|
COMMAND_NOT_ENABLED_ERROR = 2
|
|
INVALID_COMMAND_LINE_ERROR = 3
|
|
INVALID_FILE_FORMATTING_ERROR = 4
|
|
USER_NOT_ADMIN = 5
|
|
NO_CONTENTS_FOUND_FOR_OPERATION = 6
|
|
INVALID_FILE_INPUT_ERROR = 7
|
|
NO_CHANGES_MADE_OR_FOUND = 8
|
|
NO_VALID_INFO_ERROR = 9
|
|
|
|
# ****** CLI ERRORS ******
|
|
UI_CLI_ERROR_EXCEPTION = 10
|
|
UI_CLI_WARN_EXCEPTION = 11
|
|
UI_CLI_USAGE_EXCEPTION = 12
|
|
UI_CLI_COMMAND_NOT_FOUND_EXCEPTION = 13
|
|
INVALID_PASSWORD_LENGTH_ERROR = 14
|
|
|
|
# ****** RMC/RIS ERRORS ******
|
|
RIS_UNDEFINED_CLIENT_ERROR = 21
|
|
RIS_CURRENTLY_LOGGED_IN_ERROR = 22
|
|
RIS_INSTANCE_NOT_FOUND_ERROR = 23
|
|
RIS_NOTHING_SELECTED_ERROR = 24
|
|
RIS_NOTHING_SELECTED_FILTER_ERROR = 25
|
|
RIS_NOTHING_SELECTED_SET_ERROR = 26
|
|
RIS_INVALID_SELECTION_ERROR = 27
|
|
RIS_VALIDATION_ERROR = 28
|
|
RIS_MISSING_ID_TOKEN = 29
|
|
RIS_SESSION_EXPIRED = 30
|
|
|
|
# ****** REST V1 ERRORS ******
|
|
V1_RETRIES_EXHAUSTED_ERROR = 31
|
|
V1_INVALID_CREDENTIALS_ERROR = 32
|
|
V1_SERVER_DOWN_OR_UNREACHABLE_ERROR = 33
|
|
V1_CHIF_DRIVER_MISSING_ERROR = 34
|
|
REST_ILOREST_CHIF_DLL_MISSING_ERROR = 35
|
|
REST_ILOREST_UNEXPECTED_RESPONSE_ERROR = 36
|
|
REST_ILOREST_ILO_ERROR = 37
|
|
REST_ILOREST_CREATE_BLOB_ERROR = 38
|
|
REST_ILOREST_READ_BLOB_ERROR = 39
|
|
|
|
# ****** RDMC ERRORS ******
|
|
SAME_SETTINGS_ERROR = 40
|
|
FIRMWARE_UPDATE_ERROR = 41
|
|
BOOT_ORDER_ENTRY_ERROR = 42
|
|
NIC_MISSING_OR_INVALID_ERROR = 43
|
|
NO_CURRENT_SESSION_ESTABLISHED = 44
|
|
FAILURE_DURING_COMMIT_OPERATION = 45
|
|
USERNAME_PASSWORD_REQUIRED_ERROR = 46
|
|
VNIC_NOT_ENABLED_ERROR = 47
|
|
MULTIPLE_SERVER_CONFIG_FAIL = 51
|
|
MULTIPLE_SERVER_INPUT_FILE_ERROR = 52
|
|
LOAD_SKIP_SETTING_ERROR = 53
|
|
INCOMPATIBLE_ILO_VERSION_ERROR = 54
|
|
INVALID_CLIST_FILE_ERROR = 55
|
|
UNABLE_TO_MOUNT_BB_ERROR = 56
|
|
BIRTHCERT_PARSE_ERROR = 57
|
|
INCOMPATIBLE_SERVER_TYPE = 58
|
|
ILO_LICENSE_ERROR = 59
|
|
RESOURCE_EXISTS_ERROR = 60
|
|
|
|
# ****** RMC/RIS ERRORS ******
|
|
RIS_VALUE_CHANGED_ERROR = 61
|
|
RIS_REF_PATH_NOT_FOUND_ERROR = 62
|
|
RIS_ILO_RESPONSE_ERROR = 63
|
|
RIS_ILO_INIT_ERROR = 64
|
|
RIS_SCHEMA_PARSE_ERROR = 65
|
|
RIS_ILO_CHIF_ACCESS_DENIED_ERROR = 66
|
|
RIS_CREATE_AND_PREPARE_CHANNEL_ERROR = 67
|
|
RIS_ILO_CHIF_PACKET_EXCHANGE_ERROR = 71
|
|
RIS_ILO_CHIF_NO_DRIVER_ERROR = 69
|
|
|
|
# ****** REST V1 ERRORS ******
|
|
REST_ILOREST_WRITE_BLOB_ERROR = 70
|
|
REST_ILOREST_BLOB_DELETE_ERROR = 68
|
|
REST_ILOREST_BLOB_FINALIZE_ERROR = 72
|
|
REST_ILOREST_BLOB_NOT_FOUND_ERROR = 73
|
|
JSON_DECODE_ERROR = 74
|
|
V1_SECURITY_STATE_ERROR = 75
|
|
REST_ILOREST_BLOB_OVERRIDE_ERROR = 76
|
|
REST_BLOB_RETRIES_EXHAUSETED_ERROR = 77
|
|
|
|
# ****** RDMC ERRORS ******
|
|
RESOURCE_ALLOCATION_ISSUES_ERROR = 80
|
|
ENCRYPTION_ERROR = 81
|
|
DRIVE_MISSING_ERROR = 82
|
|
PATH_UNAVAILABLE_ERROR = 83
|
|
ILO_RIS_CORRUPTION_ERROR = 84
|
|
RESOURCE_NOT_READY_ERROR = 85
|
|
INVALID_SMART_ARRAY_PAYLOAD = 86
|
|
|
|
# ****** RIS ERRORS ******
|
|
RIS_RIS_BIOS_UNREGISTERED_ERROR = 100
|
|
|
|
# ***** Upload/Download ERRORS ******
|
|
FAILED_TO_DOWNLOAD_COMPONENT = 101
|
|
UPDATE_SERVICE_BUSY = 102
|
|
FAILED_TO_UPLOAD_COMPONENT = 103
|
|
TASKQUEUE_ERROR = 104
|
|
DEVICE_DISCOVERY_IN_PROGRESS = 105
|
|
INSTALLSET_ERROR = 106
|
|
INVALID_TARGET_ERROR = 107
|
|
ILO_UNSUPPORTED_FLASH = 108
|
|
|
|
# **** ComputeOpsManagement Errors****
|
|
CLOUD_CONNECT_TIMEOUT = 111
|
|
CLOUD_CONNECT_FAILED = 112
|
|
CLOUD_ALREADY_CONNECTED = 113
|
|
PROXY_CONFIG_FAILED = 114
|
|
|
|
# **** scep error ****
|
|
SCEP_ENABLED_ERROR = 121
|
|
|
|
# *** authentication error ****
|
|
TFA_WRONG_OTP = 131
|
|
TFA_OTP_TIMEDOUT = 132
|
|
TFA_ENABLED_ERROR = 133
|
|
TFA_OTP_EMAILED = 134
|
|
|
|
# ****** GENERAL ERRORS ******
|
|
GENERAL_ERROR = 255
|
|
|
|
# ****** VNIC ERRORS ******
|
|
GENERAL_ACCOUNT_GENERATE_SAVE_ERROR = 141
|
|
VNIC_DOES_NOT_EXIST_ERROR = 142
|
|
ACCOUNT_DOES_NOT_EXIST_ERROR = 143
|
|
ACCOUNT_REMOVE_ERROR = 144
|
|
ACCOUNT_EXISTS_CHECK_ERROR = 145
|
|
VNIC_LOGIN_ERROR = 146
|
|
ACCOUNT_SAVE_ERROR_TPM = 147
|
|
ACCOUNT_SAVE_ERROR_ILO = 148
|
|
GEN_BEFORE_LOGIN_ERROR = 149
|
|
APPID_LIST_ERROR = 150
|
|
INACTIVE_APP_ACCOUNT_TOKEN = 151
|
|
REACTIVATE_APP_ACCOUNT_TOKEN_ERROR = 152
|
|
|
|
|
|
class RdmcError(Exception):
|
|
"""Baseclass for all rdmc exceptions"""
|
|
|
|
errcode = 1
|
|
|
|
def __init__(self, message=None):
|
|
Exception.__init__(self, message)
|
|
|
|
|
|
class ConfigurationFileError(RdmcError):
|
|
"""Raised when something is wrong in the config file"""
|
|
|
|
errcode = 3
|
|
|
|
|
|
class ProxyConfigFailedError(RdmcError):
|
|
"""Raised when ComputeOpsManagement connection fails"""
|
|
|
|
pass
|
|
|
|
|
|
class CloudConnectTimeoutError(RdmcError):
|
|
"""Raised when ComputeOpsManagement connection times out"""
|
|
|
|
pass
|
|
|
|
|
|
class CloudConnectFailedError(RdmcError):
|
|
"""Raised when ComputeOpsManagement connection fails"""
|
|
|
|
pass
|
|
|
|
|
|
class TfaEnablePreRequisiteError(RdmcError):
|
|
"""Raised when pre-requisites not met while enabling TFA"""
|
|
|
|
pass
|
|
|
|
|
|
class AlreadyCloudConnectedError(RdmcError):
|
|
"""Raised when ComputeOpsManagement is already connected"""
|
|
|
|
pass
|
|
|
|
|
|
class CommandNotEnabledError(RdmcError):
|
|
"""Raised when user tries to invoke a command that isn't enabled"""
|
|
|
|
pass
|
|
|
|
|
|
class iLORisCorruptionError(RdmcError):
|
|
"""Raised when user tries to invoke a command that isn't enabled"""
|
|
|
|
pass
|
|
|
|
|
|
class ResourceNotReadyError(RdmcError):
|
|
"""Raised when user tries to invoke a command that isn't enabled"""
|
|
|
|
pass
|
|
|
|
|
|
class UsernamePasswordRequiredError(RdmcError):
|
|
"""Raised when username and password are required for local chif login"""
|
|
|
|
pass
|
|
|
|
|
|
class PathUnavailableError(Exception):
|
|
"""Raised when the requested path is unavailable"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidCommandLineError(RdmcError):
|
|
"""Raised when user enter incorrect command line arguments"""
|
|
|
|
pass
|
|
|
|
|
|
class NoCurrentSessionEstablished(RdmcError):
|
|
"""Raised when user enter incorrect command line arguments"""
|
|
|
|
pass
|
|
|
|
|
|
class NoChangesFoundOrMadeError(RdmcError):
|
|
"""Raised when no changes were found or made on the commit function"""
|
|
|
|
pass
|
|
|
|
|
|
class StandardBlobErrorHandler(RdmcError):
|
|
"""Raised when error occured for blob operations"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidCommandLineErrorOPTS(RdmcError):
|
|
"""Raised when user enter incorrect command line arguments"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidFileInputError(RdmcError):
|
|
"""Raised when user enter an invalid file input"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidFileFormattingError(RdmcError):
|
|
"""Raised when user enter incorrect load file formatting"""
|
|
|
|
pass
|
|
|
|
|
|
class WindowsUserNotAdmin(RdmcError):
|
|
"""Raised when user is not running as admin"""
|
|
|
|
pass
|
|
|
|
|
|
class NoContentsFoundForOperationError(RdmcError):
|
|
"""Raised when no contents were found for the current operation"""
|
|
|
|
pass
|
|
|
|
|
|
class InfoMissingEntriesError(RdmcError):
|
|
"""Raised when no valid entries for info were found in the current
|
|
instance"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidOrNothingChangedSettingsError(RdmcError):
|
|
"""Raised when something is wrong with the settings"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidPasswordLengthError(RdmcError):
|
|
"""Raised when password length is invalid"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidSmartArrayConfigurationError(RdmcError):
|
|
"""Raised when no changes were found or made on the commit function"""
|
|
|
|
pass
|
|
|
|
|
|
class NoDifferencesFoundError(RdmcError):
|
|
"""Raised when no differences are found in the current configuration"""
|
|
|
|
pass
|
|
|
|
|
|
class NothingSelectedError(RdmcError):
|
|
"""Raised when type selection is reference but none have been provided"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidPropertyError(RdmcError):
|
|
"""Raised when one or more properties or attributes are in conflict with current or
|
|
specified configuration"""
|
|
|
|
pass
|
|
|
|
|
|
class MultipleServerConfigError(RdmcError):
|
|
"""Raised when one or more servers failed to load given configuration"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidMSCfileInputError(RdmcError):
|
|
"""Raised when servers input file for load has incorrect parameters"""
|
|
|
|
pass
|
|
|
|
|
|
class FirmwareUpdateError(RdmcError):
|
|
"""Raised when there is an error while updating firmware"""
|
|
|
|
pass
|
|
|
|
|
|
class FailureDuringCommitError(RdmcError):
|
|
"""Raised when there is an error during commit"""
|
|
|
|
pass
|
|
|
|
|
|
class BootOrderMissingEntriesError(RdmcError):
|
|
"""Raised when no entries were found for bios tools"""
|
|
|
|
pass
|
|
|
|
|
|
class NicMissingOrConfigurationError(RdmcError):
|
|
""" Raised when no entries are found for given NIC or all NICs are \
|
|
configured or when wrong inputs are presented for NIC entries"""
|
|
|
|
pass
|
|
|
|
|
|
class IncompatibleiLOVersionError(RdmcError):
|
|
"""Raised when the iLO version is above or below the required \
|
|
version"""
|
|
|
|
pass
|
|
|
|
|
|
class IncompatableServerTypeError(RdmcError):
|
|
"""Raised when the server type is incompatable with the requested\
|
|
command"""
|
|
|
|
pass
|
|
|
|
|
|
class IloLicenseError(RdmcError):
|
|
"""Raised when the proper iLO license is not available for a command"""
|
|
|
|
pass
|
|
|
|
|
|
class ScepenabledError(RdmcError):
|
|
"""Raised when the generation csr or deletion of https cert is issues when scep is enabled"""
|
|
|
|
pass
|
|
|
|
|
|
class ResourceExists(RdmcError):
|
|
"""Raised when the account to be added already exists"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidCListFileError(RdmcError):
|
|
"""Raised when an error occurs while reading the cfilelist \
|
|
within AHS logs"""
|
|
|
|
pass
|
|
|
|
|
|
class PartitionMoutingError(RdmcError):
|
|
"""Raised when there is error or iLO fails to respond to \
|
|
partition mounting request"""
|
|
|
|
pass
|
|
|
|
|
|
class DownloadError(RdmcError):
|
|
"""Raised when the component fails to download"""
|
|
|
|
pass
|
|
|
|
|
|
class UploadError(RdmcError):
|
|
"""Raised when the component fails to download"""
|
|
|
|
pass
|
|
|
|
|
|
class FlashUnsupportedByIloError(RdmcError):
|
|
"""Raised when the flashing of the component is not supported by iLO"""
|
|
|
|
pass
|
|
|
|
|
|
class TimeOutError(RdmcError):
|
|
"""Raised when the update service times out"""
|
|
|
|
pass
|
|
|
|
|
|
class LibHPsrvMissingError(RdmcError):
|
|
"""Raised when unable to obtain the libhpsrv handle"""
|
|
|
|
pass
|
|
|
|
|
|
class BirthcertParseError(RdmcError):
|
|
"""Raised when unable to parse the birthcert"""
|
|
|
|
pass
|
|
|
|
|
|
class InvalidKeyError(RdmcError):
|
|
"""Raised when an invalid encryption key is used"""
|
|
|
|
pass
|
|
|
|
|
|
class UnableToDecodeError(RdmcError):
|
|
"""Raised when the file is unable to be decoded using the given key"""
|
|
|
|
pass
|
|
|
|
|
|
class UnabletoFindDriveError(RdmcError):
|
|
"""Raised when there is an issue finding required label"""
|
|
|
|
pass
|
|
|
|
|
|
class TaskQueueError(RdmcError):
|
|
"""Raised when there is an issue with the current order of taskqueue"""
|
|
|
|
pass
|
|
|
|
|
|
class DeviceDiscoveryInProgress(RdmcError):
|
|
"""Raised when device discovery in progress"""
|
|
|
|
pass
|
|
|
|
|
|
class FallbackChifUse(RdmcError):
|
|
"""Fallback Chif Use"""
|
|
|
|
pass
|
|
|
|
|
|
class InstallsetError(RdmcError):
|
|
"""Error while deleting Recovery installset"""
|
|
|
|
pass
|
|
|
|
|
|
class GenerateAndSaveAccountError(RdmcError):
|
|
"""Raised when errors occurred while generating and saving app account"""
|
|
|
|
pass
|
|
|
|
|
|
class NoAppAccountError(RdmcError):
|
|
"""Raised when the app account to be deleted does not exist."""
|
|
|
|
pass
|
|
|
|
|
|
class RemoveAccountError(RdmcError):
|
|
"""Raised when errors occurred while removing app account"""
|
|
|
|
pass
|
|
|
|
|
|
class AppAccountExistsError(RdmcError):
|
|
"""Raised when errors occurred while removing app account"""
|
|
|
|
pass
|
|
|
|
|
|
class ReactivateAppAccountTokenError(RdmcError):
|
|
"""Raised when errors occurred while reactivating app account"""
|
|
|
|
pass
|
|
|
|
|
|
class InactiveAppAccountTokenError(RdmcError):
|
|
"""Raised when inactive app account token"""
|
|
|
|
pass
|
|
|
|
|
|
class VnicLoginError(RdmcError):
|
|
"""Raised when error occurs while VNIC login"""
|
|
|
|
pass
|
|
|
|
|
|
class VnicExistsError(RdmcError):
|
|
"""Raised when VNIC is not enabled"""
|
|
|
|
pass
|
|
|
|
|
|
class SavinginTPMError(RdmcError):
|
|
"""Raised when error occurs while saving app account in TPM"""
|
|
|
|
pass
|
|
|
|
|
|
class SavinginiLOError(RdmcError):
|
|
"""Raised when error occurs while saving app account in iLO"""
|
|
|
|
pass
|
|
|
|
|
|
class GenBeforeLoginError(RdmcError):
|
|
"""Raised when error occurs while getting iLO Gen before login"""
|
|
|
|
pass
|
|
|
|
|
|
class AppIdListError(RdmcError):
|
|
"""Raised when error occurs while retrieving list of apptokens and appaccounts"""
|
|
|
|
pass
|
|
|
|
|
|
class UI(object):
|
|
"""UI class handles all of our printing etc so we have
|
|
consistency across the project"""
|
|
|
|
def __init__(self, verbosity=1):
|
|
self.verbosity = verbosity
|
|
|
|
def printer(self, data, flush=True, excp=None, verbose_override=False):
|
|
"""Print wrapper for stdout vs fileout"""
|
|
if self.verbosity >= 0 or verbose_override:
|
|
if excp:
|
|
sys.stderr.write(str(data))
|
|
else:
|
|
try:
|
|
sys.stdout.write(str(data))
|
|
if flush:
|
|
sys.stdout.flush()
|
|
except IOError:
|
|
pass
|
|
|
|
def command_not_found(self, cmd):
|
|
"""Called when command was not found"""
|
|
self.printer(
|
|
("\nCommand '%s' not found. Use the help command to " "see a list of available commands\n" % cmd),
|
|
excp=True,
|
|
)
|
|
return ReturnCodes.UI_CLI_COMMAND_NOT_FOUND_EXCEPTION
|
|
|
|
def command_not_enabled(self, cmd, excp):
|
|
"""Called when command has not been enabled"""
|
|
self.printer(("\nCommand '%s' has not been enabled: %s\n" % (cmd, excp)), excp=excp)
|
|
|
|
def invalid_commmand_line(self, excp):
|
|
"""Called when user entered invalid command line entries"""
|
|
self.printer(("Error: %s\n" % excp), excp=excp)
|
|
|
|
def ilo_ris_corruption(self, excp):
|
|
"""Called when user entered invalid command line entries"""
|
|
self.printer(("\nError: %s\n" % excp), excp=excp)
|
|
|
|
def standard_blob_error(self, excp):
|
|
"""Called when user error encountered with blob"""
|
|
self.printer(("\nError: Blob operation failed with error code %s\n" % excp), excp=excp)
|
|
|
|
def invalid_file_formatting(self, excp):
|
|
"""Called when file formatting is unrecognizable"""
|
|
self.printer(("\nError: %s\n" % excp), excp=excp)
|
|
|
|
def user_not_admin(self):
|
|
"""Called when file formatting in unrecognizable"""
|
|
self.printer(
|
|
(
|
|
"\nBoth remote and local mode is accessible when %s "
|
|
"is run as administrator. Only remote mode is available for non-"
|
|
"admin user groups.\n" % versioning.__longname__
|
|
),
|
|
excp=True,
|
|
)
|
|
|
|
def no_contents_found_for_operation(self, excp):
|
|
"""Called when no contents were found for the current operation"""
|
|
self.printer(("\nError: %s\n" % excp), excp=True)
|
|
|
|
def nothing_selected(self):
|
|
"""Called when nothing has been select yet"""
|
|
self.printer(
|
|
"\nNo type currently selected. Please use the"
|
|
" 'types' command to\nget a list of types, or input"
|
|
" your type by using the '--selector' flag.\n",
|
|
excp=True,
|
|
)
|
|
|
|
def nothing_selected_filter(self):
|
|
"""Called when nothing has been select after a filter set"""
|
|
self.printer("\nNothing was found to match your provided filter.\n", excp=True)
|
|
|
|
def nothing_selected_set(self):
|
|
"""Called when nothing has been select yet"""
|
|
self.printer("\nNothing is selected or selection is read-only.\n", excp=True)
|
|
|
|
def no_differences_found(self, excp):
|
|
"""Called when no difference is found in the current configuration"""
|
|
self.printer(("Error: %s\n" % excp), excp=True)
|
|
|
|
def multiple_server_config_fail(self, excp):
|
|
"""Called when one or more servers failed to load given configuration"""
|
|
self.printer(("Error: %s\n" % excp), excp=True)
|
|
|
|
def multiple_server_config_input_file(self, excp):
|
|
"""Called when servers input file has incorrect information"""
|
|
self.printer(("Error: %s\n" % excp), excp=True)
|
|
|
|
def invalid_credentials(self, timeout):
|
|
"""Called user has entered invalid credentials
|
|
:param timeout: timeout given for failed login attempt
|
|
:type timeout: int.
|
|
"""
|
|
self.printer("Authenticating to iLO", excp=True)
|
|
|
|
timeout = 0
|
|
for _ in range(0, (int(str(timeout)) + 3)):
|
|
time.sleep(2)
|
|
self.printer(".", excp=True)
|
|
|
|
self.printer(
|
|
"\nInvalid credentials. Unable to authenticate to iLO. \n"
|
|
"Please ensure to specify correct username and password.\n",
|
|
excp=True,
|
|
)
|
|
|
|
def bios_unregistered_error(self):
|
|
"""Called when ilo/bios unregistered error occurs"""
|
|
self.printer(
|
|
"\nERROR 100: Bios provider is unregistered. Please"
|
|
" refer to the documentation for details on this issue.\n",
|
|
excp=True,
|
|
)
|
|
|
|
def error(self, msg, inner_except=None):
|
|
"""Used for general error handling
|
|
:param inner_except: raised exception to be logged
|
|
:type inner_except: exception.
|
|
:param msg: warning message
|
|
:type msg: string.
|
|
"""
|
|
if inner_except is not None:
|
|
LOGGER.error(msg, exc_info=True)
|
|
else:
|
|
LOGGER.error(msg)
|
|
|
|
def warn(self, msg, inner_except=None):
|
|
"""Used for general warning handling
|
|
:param inner_except: raised exception to be logged
|
|
:type inner_except: exception.
|
|
:param msg: warning message
|
|
:type msg: string.
|
|
"""
|
|
if inner_except is not None:
|
|
LOGGER.warning(msg, exc_info=True)
|
|
else:
|
|
LOGGER.warning(msg)
|
|
|
|
def retries_exhausted_attemps(self):
|
|
"""Called when url retries have been exhausted"""
|
|
LOGGER.error("Could not reach URL. Retries have been exhausted.")
|
|
|
|
def retries_exhausted_vnic_not_enabled(self):
|
|
"""Called when there is no VNIC is Enabled"""
|
|
self.printer("\nError: Could not reach URL, VNIC is not enabled. \n", excp=True)
|
|
|
|
def print_out_json(self, content):
|
|
"""Print out json content to std.out with sorted keys
|
|
:param content: content to be printed out
|
|
:type content: str.
|
|
"""
|
|
# stringify
|
|
content = json.dumps(content, indent=2, cls=redfish.ris.JSONEncoder, sort_keys=True)
|
|
self.printer(content, verbose_override=True)
|
|
self.printer("\n")
|
|
|
|
def print_out_json_ordered(self, content):
|
|
"""Print out sorted json content to std.out
|
|
:param content: content to be printed out
|
|
:type content: str.
|
|
"""
|
|
content = json.dumps(content, indent=2, cls=redfish.ris.JSONEncoder)
|
|
self.printer(content, verbose_override=True)
|
|
self.printer("\n")
|
|
|
|
def print_out_human_readable(self, content):
|
|
"""Print out human readable content to std.out
|
|
:param content: content to be printed out
|
|
:type content: str.
|
|
"""
|
|
self.pretty_human_readable(content, enterloop=True)
|
|
self.printer("\n")
|
|
|
|
def pretty_human_readable(self, content, indent=0, start=0, enterloop=False):
|
|
"""Convert content to human readable and print out to std.out
|
|
:param content: content to be printed out
|
|
:type content: str.
|
|
:param indent: indent string to be used as seperator
|
|
:type indent: str.
|
|
:param start: used to determine the indent level
|
|
:type start: int.
|
|
"""
|
|
space = "\n" + "\t" * indent + " " * start
|
|
if isinstance(content, list):
|
|
for item in content:
|
|
if item is None:
|
|
continue
|
|
|
|
self.pretty_human_readable(item, indent, start)
|
|
|
|
if content.index(item) != (len(content) - 1):
|
|
self.printer(space)
|
|
elif isinstance(content, dict):
|
|
for key, value in content.items():
|
|
if space and not enterloop:
|
|
self.printer(space)
|
|
|
|
enterloop = False
|
|
self.printer((str(key) + "="))
|
|
self.pretty_human_readable(value, indent, (start + len(key) + 2))
|
|
else:
|
|
content = content if isinstance(content, six.string_types) else str(content)
|
|
|
|
content = '""' if not content else content
|
|
# Changed to support py3, verify if there is a unicode prit issue.
|
|
|
|
self.printer(content)
|
|
|
|
|
|
class Encryption(object):
|
|
"""Encryption/Decryption object"""
|
|
|
|
@staticmethod
|
|
def check_fips_mode_os():
|
|
"""Function to check for the OS fips mode
|
|
:param key: string to encrypt with
|
|
:type key: str.
|
|
:returns: returns True if FIPS mode is active, False otherwise
|
|
"""
|
|
fips = False
|
|
if os.name == "nt":
|
|
reg = winreg.ConnectRegistry(None, HKEY_LOCAL_MACHINE)
|
|
try:
|
|
reg = winreg.OpenKey(
|
|
reg,
|
|
"System\\CurrentControlSet\\Control\\" "Lsa\\FipsAlgorithmPolicy",
|
|
)
|
|
winreg.QueryInfoKey(reg)
|
|
value, _ = winreg.QueryValueEx(reg, "Enabled")
|
|
if value:
|
|
fips = True
|
|
except:
|
|
fips = False
|
|
else:
|
|
try:
|
|
fipsfile = open("/proc/sys/crypto/fips_enabled")
|
|
result = fipsfile.readline()
|
|
if int(result) > 0:
|
|
fipsfile = True
|
|
fipsfile.close()
|
|
except:
|
|
fips = False
|
|
return fips
|
|
|
|
@staticmethod
|
|
def check_fips_mode_ssl():
|
|
"""Function to check for the SSL fips mode
|
|
Uses custom cpython ssl module API, if available. Otheriwse
|
|
probes using ctypes.cdll APIs.
|
|
:returns: returns True if FIPS mode is active, False otherwise
|
|
"""
|
|
import ssl
|
|
|
|
if hasattr(ssl, "FIPS_mode"):
|
|
return ssl.FIPS_mode()
|
|
|
|
from ctypes import cdll
|
|
|
|
libcrypto = cdll.LoadLibrary(ssl._ssl.__file__)
|
|
return libcrypto.FIPS_mode()
|
|
|
|
def encrypt_file(self, filetxt, key):
|
|
"""encrypt a file given a key
|
|
:param filetxt: content to be encrypted
|
|
:type content: str.
|
|
:param key: string to encrypt with
|
|
:type key: str.
|
|
"""
|
|
try:
|
|
filetxt = filetxt.encode()
|
|
except (UnicodeDecodeError, AttributeError):
|
|
pass # must be encoded already
|
|
try:
|
|
key = key.encode()
|
|
except (UnicodeDecodeError, AttributeError):
|
|
pass # must be encoded already
|
|
if Encryption.check_fips_mode_os():
|
|
raise CommandNotEnabledError("Encrypting of files is not available" " in FIPS mode.")
|
|
if len(key) not in [16, 24, 32]:
|
|
raise InvalidKeyError("")
|
|
else:
|
|
encryptedfile = pyaes.AESModeOfOperationCTR(key).encrypt(filetxt)
|
|
|
|
return encryptedfile
|
|
|
|
def decrypt_file(self, filetxt, key):
|
|
"""decrypt a file given a key
|
|
:param filetxt: content to be decrypted
|
|
:type content: str.
|
|
:param key: string to decrypt with
|
|
:type key: str.
|
|
:returns: returns the decrypted file
|
|
"""
|
|
try:
|
|
filetxt = filetxt.encode()
|
|
except (UnicodeDecodeError, AttributeError):
|
|
pass # must be encoded already
|
|
try:
|
|
key = key.encode()
|
|
except (UnicodeDecodeError, AttributeError):
|
|
pass # must be encoded already
|
|
if len(key) not in [16, 24, 32]:
|
|
raise InvalidKeyError("")
|
|
else:
|
|
decryptedfile = pyaes.AESModeOfOperationCTR(key).decrypt(filetxt)
|
|
try:
|
|
json.loads(decryptedfile)
|
|
except:
|
|
raise UnableToDecodeError(
|
|
"Unable to decrypt the file, make " "sure the key is the same as used in encryption."
|
|
)
|
|
|
|
return decryptedfile
|
|
|
|
@staticmethod
|
|
def decode_credentials(credential):
|
|
"""decode an encoded credential
|
|
:param credential: credential to be decoded
|
|
:type credential: str.
|
|
:returns: returns the decoded credential
|
|
"""
|
|
|
|
lib = risblobstore2.BlobStore2.gethprestchifhandle()
|
|
credbuff = create_string_buffer(credential.encode("utf-8"))
|
|
retbuff = create_string_buffer(128)
|
|
|
|
lib.decode_credentials.argtypes = [c_char_p]
|
|
|
|
lib.decode_credentials(credbuff, byref(retbuff))
|
|
|
|
risblobstore2.BlobStore2.unloadchifhandle(lib)
|
|
# try:
|
|
# if isinstance(retbuff.value, bytes):
|
|
# retbuff.value = retbuff.value.decode('utf-8', 'ignore')
|
|
# if not retbuff.value:
|
|
# raise UnableToDecodeError("")
|
|
# except:
|
|
# raise UnableToDecodeError("Unable to decode credential %s." % credential)
|
|
|
|
return retbuff.value
|
|
|
|
@staticmethod
|
|
def encode_credentials(credential):
|
|
"""encode a credential
|
|
:param credential: credential to be encoded
|
|
:type credential: str.
|
|
:returns: returns the encoded credential
|
|
"""
|
|
|
|
lib = risblobstore2.BlobStore2.gethprestchifhandle()
|
|
if isinstance(credential, bytes):
|
|
credential = credential.decode("utf-8")
|
|
credbuff = create_string_buffer(credential.encode("utf-8"))
|
|
|
|
retbuff = create_string_buffer(128)
|
|
|
|
lib.encode_credentials.argtypes = [c_char_p]
|
|
|
|
lib.encode_credentials(credbuff, byref(retbuff))
|
|
|
|
risblobstore2.BlobStore2.unloadchifhandle(lib)
|
|
try:
|
|
if six.PY2:
|
|
enc_val = retbuff.value.encode("utf-8")
|
|
elif six.PY3:
|
|
enc_val = retbuff.value.decode("utf-8") # .encode('utf-8')
|
|
if not retbuff.value:
|
|
raise UnableToDecodeError("")
|
|
except Exception:
|
|
raise UnableToDecodeError("Unable to decode credential %s." % credential)
|
|
|
|
return enc_val
|
|
|
|
|
|
class TabAndHistoryCompletionClass(Completer):
|
|
"""Tab and History Class used by interactive mode"""
|
|
|
|
def __init__(self, options):
|
|
self.options = options
|
|
self.toolbar_text = None
|
|
self.last_complete = None
|
|
|
|
def get_completions(self, document, complete_event):
|
|
"""Function to return the options for autocomplete"""
|
|
word = ""
|
|
self.toolbar_text = ""
|
|
lstoption = self.options
|
|
if document.text:
|
|
tokens = document.text.split()
|
|
# We aren't completing options yet
|
|
tokens = [token for token in tokens if not token.startswith("-")]
|
|
|
|
self.last_complete = tokens[-1]
|
|
nestedtokens = self.last_complete.split("/")
|
|
|
|
if not document.text.endswith(" "):
|
|
tokens.pop()
|
|
word = document.get_word_under_cursor()
|
|
else:
|
|
nestedtokens = []
|
|
if word == "/":
|
|
word = ""
|
|
|
|
if len(tokens) >= 1:
|
|
if tokens[0] == "select":
|
|
# only first type
|
|
if len(tokens) >= 2:
|
|
lstoption = []
|
|
else:
|
|
lstoption = self.options.get(tokens[0], {})
|
|
elif tokens[0] in ["get", "list", "info", "set"]:
|
|
# Match properties
|
|
nested_data = self.options.get("nestedprop", {})
|
|
nested_info = self.options.get("nestedinfo", {})
|
|
for token in nestedtokens:
|
|
try:
|
|
nested_data = nested_data[token]
|
|
if tokens[0] == "get" and isinstance(nested_data, dict):
|
|
for k in list(nested_data.keys()):
|
|
if (
|
|
k.lower() in HARDCODEDLIST
|
|
or "@odata" in k.lower()
|
|
or "@redfish.allowablevalues" in k.lower()
|
|
):
|
|
del nested_data[k]
|
|
if nested_info:
|
|
if "properties" in nested_info:
|
|
nested_info = nested_info["properties"]
|
|
if "AttributeName" not in nested_info[token]:
|
|
nested_info = (
|
|
nested_info["properties"][token]
|
|
if "properties" in nested_info
|
|
else nested_info[token]
|
|
)
|
|
else:
|
|
nested_info = nested_info[token]
|
|
except Exception:
|
|
break
|
|
nested_data = list(nested_data.keys()) if isinstance(nested_data, dict) else []
|
|
lstoption = nested_data
|
|
|
|
# Try to get info for help bar
|
|
help_text = nested_info.get("HelpText", "")
|
|
enum_tab = []
|
|
if "Type" in nested_info and nested_info["Type"].lower() == "enumeration":
|
|
help_text += "\nPossible Values:\n"
|
|
for value in nested_info["Value"]:
|
|
enum_tab.append(value["ValueName"])
|
|
help_text += six.u(str(value["ValueName"])) + " "
|
|
|
|
if not help_text:
|
|
try:
|
|
nested_info = nested_info["properties"]
|
|
except KeyError:
|
|
pass
|
|
help_text = nested_info.get("description", "")
|
|
if "enum" in nested_info:
|
|
help_text += "\nPossible Values:\n"
|
|
for value in nested_info["enum"]:
|
|
enum_tab.append(value)
|
|
help_text += six.u(str(value)) + " "
|
|
if isinstance(help_text, str):
|
|
help_text = help_text.replace(". ", ".\n")
|
|
self.toolbar_text = help_text
|
|
if tokens[0] in ["set"]:
|
|
lstoption = self.options.get("set")
|
|
else:
|
|
lstoption = {}
|
|
else:
|
|
for token in tokens:
|
|
# just match commands
|
|
lstoption = self.options.get(token, {})
|
|
|
|
for opt in lstoption:
|
|
if opt == word:
|
|
self.last_complete = opt
|
|
if opt.startswith(word):
|
|
yield Completion(opt + "", start_position=-len(word))
|
|
|
|
def bottom_toolbar(self):
|
|
return self.toolbar_text if self.toolbar_text else None
|
|
|
|
def updates_tab_completion_lists(self, options):
|
|
"""Function to update tab completion lists
|
|
:param options: options list
|
|
:type options: list.
|
|
"""
|
|
# Loop through options passed and add them to them
|
|
# to the current tab options list
|
|
for key, value in options.items():
|
|
self.options[key] = value
|