513 lines
21 KiB
Python
513 lines
21 KiB
Python
###
|
|
# Copyright 2016-2023 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 -*-
|
|
"""Get Command for RDMC"""
|
|
from collections import OrderedDict
|
|
|
|
import six
|
|
import logging
|
|
|
|
import redfish.ris
|
|
from redfish.ris.utils import iterateandclear
|
|
|
|
try:
|
|
from rdmc_helper import (
|
|
UI,
|
|
InvalidCommandLineError,
|
|
InvalidCommandLineErrorOPTS,
|
|
NoContentsFoundForOperationError,
|
|
ReturnCodes,
|
|
)
|
|
except ImportError:
|
|
from ilorest.rdmc_helper import (
|
|
UI,
|
|
InvalidCommandLineError,
|
|
InvalidCommandLineErrorOPTS,
|
|
NoContentsFoundForOperationError,
|
|
ReturnCodes,
|
|
)
|
|
try:
|
|
from rdmc_helper import HARDCODEDLIST
|
|
except ImportError:
|
|
from ilorest.rdmc_helper import HARDCODEDLIST
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class GetCommand:
|
|
"""Constructor"""
|
|
|
|
def __init__(self):
|
|
self.ident = {
|
|
"name": "get",
|
|
"usage": None,
|
|
"description": "To retrieve all"
|
|
" the properties run without arguments. \n\t*Note*: "
|
|
"a type will need to be selected or this will return an "
|
|
"error.\n\texample: get\n\n\tTo retrieve multiple "
|
|
"properties use the following example\n\texample: "
|
|
"get Temperatures/ReadingCelsius Fans/Name --selector=Thermal."
|
|
"\n\n\tTo change output style format provide"
|
|
" the json flag\n\texample: get --json",
|
|
"summary": "Displays the current value(s) of a " "property(ies) within a selected type.",
|
|
"aliases": [],
|
|
"auxcommands": ["LogoutCommand"],
|
|
}
|
|
self.cmdbase = None
|
|
self.rdmc = None
|
|
self.auxcommands = dict()
|
|
|
|
def run(self, line, help_disp=False):
|
|
"""Main get worker function
|
|
|
|
:param line: command line input
|
|
:type line: string.
|
|
"""
|
|
if help_disp:
|
|
self.parser.print_help()
|
|
return ReturnCodes.SUCCESS
|
|
try:
|
|
(options, args) = self.rdmc.rdmc_parse_arglist(self, line)
|
|
except (InvalidCommandLineErrorOPTS, SystemExit):
|
|
if ("-h" in line) or ("--help" in line):
|
|
return ReturnCodes.SUCCESS
|
|
else:
|
|
raise InvalidCommandLineErrorOPTS("")
|
|
|
|
if getattr(options, "json"):
|
|
self.rdmc.json = True
|
|
|
|
selector = options.selector.lower() if options.selector else self.rdmc.app.selector.lower()
|
|
|
|
if "securityservice" in selector:
|
|
self.cmdbase.login_validation(self, options)
|
|
else:
|
|
self.getvalidation(options)
|
|
|
|
filtr = (None, None)
|
|
if options.filter:
|
|
try:
|
|
if (str(options.filter)[0] == str(options.filter)[-1]) and str(options.filter).startswith(("'", '"')):
|
|
options.filter = options.filter[1:-1]
|
|
|
|
(sel, val) = options.filter.split("=")
|
|
filtr = (sel.strip(), val.strip())
|
|
|
|
except (InvalidCommandLineError, SystemExit):
|
|
raise InvalidCommandLineError("Invalid filter" " parameter format [filter_attribute]=[filter_value]")
|
|
|
|
self.getworkerfunction(
|
|
args,
|
|
options,
|
|
results=None,
|
|
uselist=True,
|
|
filtervals=filtr,
|
|
readonly=options.noreadonly,
|
|
)
|
|
|
|
self.cmdbase.logout_routine(self, options)
|
|
# Return code
|
|
return ReturnCodes.SUCCESS
|
|
|
|
def getworkerfunction(
|
|
self,
|
|
args,
|
|
options,
|
|
readonly=False,
|
|
filtervals=(None, None),
|
|
results=None,
|
|
uselist=False,
|
|
):
|
|
"""main get worker function
|
|
|
|
:param args: command line arguments
|
|
:type args: list.
|
|
:param options: command line options
|
|
:type options: list.
|
|
:param line: command line input
|
|
:type line: string.
|
|
:param readonly: remove readonly properties
|
|
:type readonly: bool
|
|
:param filtervals: filter key value pair (Key,Val)
|
|
:type filtervals: tuple
|
|
:param results: current results collected
|
|
:type results: string.
|
|
:param uselist: use reserved properties list to filter results
|
|
:type uselist: boolean.
|
|
"""
|
|
nocontent = set()
|
|
instances = None
|
|
arg = None
|
|
|
|
# For rest redfish compatibility of bios.
|
|
if hasattr(options, "selector") and options.selector:
|
|
self.rdmc.app.selector = options.selector
|
|
args = [args] if args and isinstance(args, six.string_types) else args
|
|
if self.rdmc.app.selector and "." not in self.rdmc.app.selector:
|
|
self.rdmc.app.selector = self.rdmc.app.selector + "."
|
|
args = (
|
|
[
|
|
(
|
|
"Attributes/" + arg
|
|
if self.rdmc.app.selector
|
|
and self.rdmc.app.selector.lower().startswith("bios.")
|
|
and not (
|
|
arg.lower() in ["id", "name", "attributeregistry", "attributes"] or "@odata" in arg.lower()
|
|
)
|
|
else arg
|
|
)
|
|
for arg in args
|
|
]
|
|
if args
|
|
else args
|
|
)
|
|
if filtervals[0]:
|
|
instances = self.rdmc.app.select(selector=self.rdmc.app.selector, fltrvals=filtervals)
|
|
|
|
try:
|
|
if "securityservice" in self.rdmc.app.selector.lower():
|
|
url = "/redfish/v1/Managers/1/SecurityService/"
|
|
LOGGER.info("Fetching SecurityService data from: %s", url)
|
|
contents = self.rdmc.app.get_handler(url, service=True, silent=True).dict
|
|
security_contents = []
|
|
if not args and not options.json:
|
|
LOGGER.debug("Printing human-readable security service data.")
|
|
UI().print_out_human_readable(contents)
|
|
elif options and options.json and contents:
|
|
LOGGER.debug("Printing JSON security service data.")
|
|
UI().print_out_json(contents)
|
|
else:
|
|
attr = args[0]
|
|
contents_lower = {k.lower(): v for k, v in contents.items()}
|
|
security_contents.append({attr: contents_lower[attr.lower()]})
|
|
if security_contents:
|
|
UI().print_out_human_readable(security_contents)
|
|
elif (
|
|
"componentintegrity"
|
|
in self.rdmc.app.selector.lower()
|
|
# or "bios." in self.rdmc.app.selector.lower()
|
|
):
|
|
url = ""
|
|
if "componentintegrity" in self.rdmc.app.selector.lower():
|
|
url = "/redfish/v1/ComponentIntegrity/?$expand=."
|
|
# elif "bios." in self.rdmc.app.selector.lower():
|
|
# url = "/redfish/v1/systems/1/bios/?$expand=."
|
|
|
|
LOGGER.info("Fetching %s data from: %s", self.rdmc.app.selector, url)
|
|
contents = self.get_content(url, args, uselist, options)
|
|
else:
|
|
LOGGER.debug("Handling general selector case.")
|
|
skipnonsettingflag = True
|
|
if "networkadapter" in self.rdmc.app.selector.lower():
|
|
skipnonsettingflag = False
|
|
|
|
if "selector" in options:
|
|
LOGGER.info("Fetching properties with selector: %s", options.selector)
|
|
contents = self.rdmc.app.getprops(
|
|
props=args,
|
|
remread=readonly,
|
|
selector=options.selector,
|
|
nocontent=nocontent,
|
|
insts=instances,
|
|
skipnonsetting=skipnonsettingflag,
|
|
)
|
|
|
|
if "networkadapter" in self.rdmc.app.selector.lower():
|
|
unique_adapters = {}
|
|
for entry in contents:
|
|
# Remove keys @odata.id & Id if it contains "settings"
|
|
adapter_id = entry.get("Id", "")
|
|
odata_id = entry.get("@odata.id", "")
|
|
if "settings" in odata_id.lower() or "settings" in adapter_id.lower():
|
|
continue
|
|
# Preventing deduplicates by Id
|
|
if adapter_id:
|
|
unique_adapters[adapter_id] = entry
|
|
|
|
if unique_adapters:
|
|
contents = list(unique_adapters.values())
|
|
|
|
if (
|
|
contents
|
|
and "logentrycollection" in self.rdmc.app.selector.lower()
|
|
and "@odata.id" in args
|
|
and not uselist
|
|
):
|
|
# as per documentation
|
|
# collection of LogEntry resource instances
|
|
log_contents = self.rdmc.app.getprops(
|
|
props=["entries/@odata.id"],
|
|
remread=readonly,
|
|
selector="logservice",
|
|
nocontent=nocontent,
|
|
insts=instances,
|
|
skipnonsetting=skipnonsettingflag,
|
|
)
|
|
if log_contents:
|
|
log_contents = [
|
|
val["Entries"]
|
|
for val in log_contents
|
|
if "Entries" in val and "@odata.id" in val["Entries"]
|
|
]
|
|
|
|
if log_contents:
|
|
contents.extend(log_contents)
|
|
# Removing duplicates
|
|
contents = list({val["@odata.id"]: val for val in contents}.values())
|
|
|
|
else:
|
|
LOGGER.info("Fetching properties without selector.")
|
|
contents = self.rdmc.app.getprops(
|
|
props=args, remread=readonly, nocontent=nocontent, insts=instances
|
|
)
|
|
uselist = False if readonly else uselist
|
|
except redfish.ris.rmc_helper.EmptyRaiseForEAFP:
|
|
LOGGER.error("EmptyRaiseForEAFP exception encountered, retrying getprops without remread.")
|
|
contents = self.rdmc.app.getprops(props=args, nocontent=nocontent)
|
|
for ind, content in enumerate(contents):
|
|
if "bios." in self.rdmc.app.selector.lower() and "Attributes" in list(content.keys()):
|
|
content.update(content["Attributes"])
|
|
del content["Attributes"]
|
|
LOGGER.debug("Flattened BIOS attributes in response.")
|
|
contents[ind] = OrderedDict(sorted(list(content.items()), key=lambda x: x[0]))
|
|
if len(args) > 0 and "Members" in args[0]:
|
|
uselist = False
|
|
if uselist:
|
|
for item in contents:
|
|
self.removereserved(item)
|
|
if results:
|
|
LOGGER.info("Returning retrieved contents.")
|
|
return contents
|
|
|
|
contents = contents[0] if len(contents) == 1 else contents
|
|
|
|
if options and options.json and contents:
|
|
LOGGER.debug("Printing output in JSON format.")
|
|
UI().print_out_json(contents)
|
|
elif contents:
|
|
LOGGER.debug("Printing output in human-readable format.")
|
|
UI().print_out_human_readable(contents)
|
|
else:
|
|
try:
|
|
if nocontent or not any(next(iter(contents))):
|
|
raise Exception()
|
|
except Exception:
|
|
strtoprint = ", ".join(str(val) for val in nocontent)
|
|
if not strtoprint and arg:
|
|
strtoprint = arg
|
|
LOGGER.error("No contents found for entry: %s", strtoprint)
|
|
raise NoContentsFoundForOperationError("No get contents found for entry: %s" % strtoprint)
|
|
else:
|
|
LOGGER.error("No contents found for selected type.")
|
|
raise NoContentsFoundForOperationError("No get contents found for " "selected type.")
|
|
if options.logout:
|
|
LOGGER.info("Logging out.")
|
|
self.auxcommands["logout"].run("")
|
|
|
|
def get_content(self, url, args, uselist, options):
|
|
"""Get data from url"""
|
|
contents = self.rdmc.app.get_handler(url, service=True, silent=True).dict
|
|
if uselist:
|
|
# for get command removing @odata.* properties
|
|
args = [arg.lower() for arg in args if "@odata." not in arg.lower()]
|
|
else:
|
|
args = [arg.lower() for arg in args]
|
|
|
|
if contents and not args:
|
|
contents = [contents]
|
|
elif contents and args:
|
|
collect_data = []
|
|
for key, value in contents.items():
|
|
# Collect top-level keys based on args
|
|
if any(key.lower() == arg for arg in args):
|
|
collect_data.append({key: value})
|
|
|
|
if "attributes" == key.lower() and isinstance(value, dict) and "attributes" not in args:
|
|
attr_vals = {k.lower(): v for k, v in value.items()}
|
|
val_dict = {arg: attr_vals[arg.lower()] for arg in args if arg.lower() in attr_vals}
|
|
if val_dict:
|
|
collect_data.append(val_dict)
|
|
|
|
if key == "Members" and isinstance(value, list):
|
|
for member in value:
|
|
# Loop through each dict of Members
|
|
m_dict = {k.lower(): v for k, v in member.items()}
|
|
val_dict = {arg: m_dict[arg.lower()] for arg in args if arg.lower() in m_dict}
|
|
if val_dict:
|
|
collect_data.append(val_dict)
|
|
|
|
# collect_data = [dict(t) for t in {tuple(d.items()) for d in collect_data}]
|
|
contents = collect_data
|
|
else:
|
|
contents = []
|
|
|
|
return contents
|
|
|
|
def find_key_recursive(self, data, target_key):
|
|
"""Recursively searches for a key in a nested dictionary (case-insensitive) and returns its value(s)."""
|
|
target_key_lower = target_key.lower() # Convert target_key to lowercase for case-insensitive matching
|
|
|
|
if isinstance(data, dict):
|
|
for key, value in data.items():
|
|
if key.lower() == target_key_lower: # Convert dictionary key to lowercase before comparison
|
|
yield value # Use yield to return multiple matches
|
|
elif isinstance(value, (dict, list)):
|
|
yield from self.find_key_recursive(value, target_key)
|
|
|
|
elif isinstance(data, list):
|
|
for item in data:
|
|
yield from self.find_key_recursive(item, target_key)
|
|
|
|
def removereserved(self, entry):
|
|
"""function to remove reserved properties
|
|
:param entry: dictionary to remove reserved properties from
|
|
:type entry: dict.
|
|
"""
|
|
|
|
for key, val in list(entry.items()):
|
|
if key.lower() in HARDCODEDLIST or "@odata" in key.lower():
|
|
del entry[key]
|
|
elif isinstance(val, list):
|
|
for item in entry[key]:
|
|
if isinstance(item, dict) and item:
|
|
self.removereserved(item)
|
|
if all([True if not test else False for test in entry[key]]):
|
|
del entry[key]
|
|
elif isinstance(val, dict):
|
|
self.removereserved(val)
|
|
if all([True if not test else False for test in entry[key]]):
|
|
del entry[key]
|
|
|
|
def checktoprint(self, options, contents, nocontent, arg):
|
|
"""function to decide what/how to print
|
|
:param options: list of options
|
|
:type options: list.
|
|
:param contents: dictionary value returned by getprops.
|
|
:type contents: dict.
|
|
:param nocontent: props not found are added to the list.
|
|
:type nocontent: list.
|
|
:param arg: string of args
|
|
:type arg: string
|
|
"""
|
|
if options and options.json and contents:
|
|
self.rdmc.ui.print_out_json(contents)
|
|
elif contents:
|
|
self.rdmc.ui.print_out_human_readable(contents)
|
|
else:
|
|
try:
|
|
if nocontent or not any(next(iter(contents))):
|
|
raise Exception()
|
|
except Exception:
|
|
strtoprint = ", ".join(str(val) for val in nocontent)
|
|
if not strtoprint and arg:
|
|
strtoprint = arg
|
|
raise NoContentsFoundForOperationError("No get contents " "found for entry: %s" % strtoprint)
|
|
else:
|
|
raise NoContentsFoundForOperationError("No get contents " "found for selected type.")
|
|
|
|
def collectandclear(self, contents, key, values):
|
|
"""function to find and remove unneeded values from contents dictionary
|
|
:param contents: dictionary value returned by getprops
|
|
:type contents: dict.
|
|
:param key: string of keys
|
|
:type key: string.
|
|
:param values: list of values
|
|
:type values: list.
|
|
"""
|
|
clearcontent = contents[0][key]
|
|
if isinstance(clearcontent, dict):
|
|
keyslist = list(clearcontent.keys())
|
|
else:
|
|
keyslist = [clearcontent]
|
|
clearedlist = keyslist
|
|
for arg in values:
|
|
for keys in keyslist:
|
|
if str(keys).lower() == str(arg).lower():
|
|
clearedlist.remove(arg)
|
|
contents = iterateandclear(contents, clearedlist)
|
|
return contents
|
|
|
|
def getvalidation(self, options):
|
|
"""get method validation function
|
|
|
|
:param options: command line options
|
|
:type options: list.
|
|
"""
|
|
self.cmdbase.login_select_validation(self, options)
|
|
|
|
def definearguments(self, customparser):
|
|
"""Wrapper function for new command main function
|
|
|
|
:param customparser: command line input
|
|
:type customparser: parser.
|
|
"""
|
|
if not customparser:
|
|
return
|
|
|
|
self.cmdbase.add_login_arguments_group(customparser)
|
|
|
|
customparser.add_argument(
|
|
"--selector",
|
|
dest="selector",
|
|
help="Optionally include this flag to select a type to run"
|
|
" the current command on. Use this flag when you wish to"
|
|
" select a type without entering another command, or if you"
|
|
" wish to work with a type that is different from the one"
|
|
" you currently have selected.",
|
|
default=None,
|
|
)
|
|
|
|
customparser.add_argument(
|
|
"--filter",
|
|
dest="filter",
|
|
help="Optionally set a filter value for a filter attribute."
|
|
" This uses the provided filter for the currently selected"
|
|
" type. Note: Use this flag to narrow down your results. For"
|
|
" example, selecting a common type might return multiple"
|
|
" objects that are all of that type. If you want to modify"
|
|
" the properties of only one of those objects, use the filter"
|
|
" flag to narrow down results based on properties."
|
|
"\t\t\t\t\t Usage: --filter [ATTRIBUTE]=[VALUE]",
|
|
default=None,
|
|
)
|
|
customparser.add_argument(
|
|
"-j",
|
|
"--json",
|
|
dest="json",
|
|
action="store_true",
|
|
help="Optionally include this flag if you wish to change the"
|
|
" displayed output to JSON format. Preserving the JSON data"
|
|
" structure makes the information easier to parse.",
|
|
default=False,
|
|
)
|
|
customparser.add_argument(
|
|
"--noreadonly",
|
|
dest="noreadonly",
|
|
action="store_true",
|
|
help="Optionally include this flag if you wish to only show"
|
|
" properties that are not read-only. This is useful to see what "
|
|
"is configurable with the selected type(s).",
|
|
default=False,
|
|
)
|
|
customparser.add_argument(
|
|
"--refresh",
|
|
dest="ref",
|
|
action="store_true",
|
|
help="Optionally reload the data of selected type and clear " "patches from current selection.",
|
|
default=False,
|
|
)
|