1813 lines
74 KiB
Python
1813 lines
74 KiB
Python
###
|
|
# Copyright 2020 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 -*-
|
|
|
|
"""A convenience layer that combines multiple lower level classes and functions into one."""
|
|
|
|
import copy
|
|
import hashlib
|
|
import logging
|
|
import shutil
|
|
|
|
# ---------Imports---------
|
|
import sys
|
|
import time
|
|
|
|
import six
|
|
|
|
try:
|
|
from collections import OrderedDict
|
|
except ImportError:
|
|
from collections.abc import OrderedDict
|
|
|
|
import jsonpatch
|
|
import jsonpointer
|
|
|
|
import redfish.ris.gen_compat
|
|
import redfish.ris.validation
|
|
from redfish.rest.v1 import RestClient
|
|
from redfish.ris.resp_handler import ResponseHandler
|
|
from redfish.ris.ris import RisMonolith, SchemaValidationError, SessionExpired
|
|
from redfish.ris.rmc_helper import (
|
|
EmptyRaiseForEAFP,
|
|
IloResponseError,
|
|
IncompatibleiLOVersionError,
|
|
InstanceNotFoundError,
|
|
LoadSkipSettingError,
|
|
NothingSelectedError,
|
|
NothingSelectedSetError,
|
|
RmcFileCacheManager,
|
|
UndefinedClientError,
|
|
ValidationError,
|
|
ValueChangedError,
|
|
)
|
|
from redfish.ris.utils import (
|
|
checkallowablevalues,
|
|
diffdict,
|
|
getattributeregistry,
|
|
iterateandclear,
|
|
merge_dict,
|
|
navigatejson,
|
|
print_handler,
|
|
skipnonsettingsinst,
|
|
validate_headers,
|
|
warning_handler,
|
|
)
|
|
from redfish.ris.validation import Typepathforval, ValidationManager
|
|
from redfish.hpilo.vnichpilo import AppAccount
|
|
|
|
# ---------End of imports---------
|
|
|
|
# ---------Debug logger---------
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------End of debug logger---------
|
|
|
|
|
|
class RmcApp(object):
|
|
"""A convenience class that combines the client, compatibility, validation, caching,
|
|
and monolith into one class.
|
|
|
|
:param showwarnings: Flag to print warnings to std.out (True) or log in log file (False)
|
|
:type showwarnings: bool
|
|
:param cache_dir: Cache directory to save cache data to, if None, RmcApp will not cache data.
|
|
Cache can allow your RmcApp to persist between scripts.
|
|
:type cache_dir: str
|
|
"""
|
|
|
|
def __init__(self, showwarnings=False, cache_dir=None):
|
|
self.logger = LOGGER
|
|
self.redfishinst = None
|
|
self.sessioninst = None
|
|
self._cm = RmcFileCacheManager(self)
|
|
self.monolith = None
|
|
self._iloversion = None
|
|
self._validationmanager = None
|
|
self._selector = None
|
|
self._cachedir = cache_dir
|
|
self.verbose = 1
|
|
self._sessionid = None
|
|
self.typepath = redfish.ris.gen_compat.Typesandpathdefines()
|
|
Typepathforval(typepathobj=self.typepath)
|
|
|
|
if not showwarnings:
|
|
self.logger.setLevel(logging.WARNING)
|
|
if self.logger.handlers and self.logger.handlers[0].name == "lerr":
|
|
self.logger.handlers.remove(self.logger.handlers[0])
|
|
|
|
@property
|
|
def monolith(self):
|
|
"""Get the monolith from the current client"""
|
|
return self._monolith
|
|
|
|
@monolith.setter
|
|
def monolith(self, monolith):
|
|
"""Set the monolith"""
|
|
self._monolith = monolith
|
|
|
|
@property
|
|
def current_client(self):
|
|
"""Get the current client"""
|
|
if self.redfishinst:
|
|
return self.redfishinst
|
|
raise UndefinedClientError()
|
|
|
|
@property
|
|
def validationmanager(self):
|
|
"""Get the valdation manager"""
|
|
if self.getiloversion():
|
|
if self._validationmanager:
|
|
self._validationmanager.reset_errors_warnings()
|
|
else:
|
|
monolith = self.monolith
|
|
self._validationmanager = ValidationManager(monolith, defines=self.typepath)
|
|
self._validationmanager.updatevalidationdata()
|
|
else:
|
|
self._validationmanager = None
|
|
return self._validationmanager
|
|
|
|
@property
|
|
def selector(self):
|
|
"""The selector that will be used to gather data if no selector or instance argument is
|
|
passed."""
|
|
return self._selector
|
|
|
|
@selector.setter
|
|
def selector(self, sel):
|
|
"""Set the selector"""
|
|
self._selector = sel
|
|
|
|
@property
|
|
def cachedir(self):
|
|
"""The cache directory that is used to cache app data to a file,
|
|
None if not caching data."""
|
|
return self._cachedir
|
|
|
|
@cachedir.setter
|
|
def cachedir(self, cache_dir):
|
|
"""Set the cachedir"""
|
|
self._cachedir = cache_dir
|
|
|
|
@property
|
|
def cache(self):
|
|
"""True if we are caching data, False if we are not"""
|
|
return True if self.cachedir else False
|
|
|
|
def restore(self, creds=None, enc=False):
|
|
"""Restores the monolith from cache. Used to load a monolith data back into a new app
|
|
class. Keyword arguments are only needed in a local client when in a high security mode.
|
|
|
|
:param creds: Credentials to create the client with.
|
|
:type creds: str
|
|
:param enc: Flag to determine if encoding functions are being used. True if being used false
|
|
if not.
|
|
:type enc: bool
|
|
"""
|
|
self._cm.uncache_rmc(creds=creds, enc=enc)
|
|
|
|
def set_encode_funct(self, funct):
|
|
"""Set the encoding function for cache to use. Can be used to protect sensitive data when
|
|
it is at rest.
|
|
|
|
:param funct: The function to use for encoding data
|
|
:type funct: function
|
|
"""
|
|
self._cm.encodefunct = funct
|
|
|
|
def set_decode_funct(self, funct):
|
|
"""Set the decoding function for cache to use. Is used in conjunction with the
|
|
`set_encode_funct` to turn the encoded data back into a usable string.
|
|
|
|
:param funct: The function to use for decoding data
|
|
:type funct: function
|
|
"""
|
|
self._cm.decodefunct = funct
|
|
|
|
def save(self):
|
|
"""Updates the cache with the latest monolith data."""
|
|
self._cm.cache_rmc()
|
|
|
|
def login(
|
|
self,
|
|
username=None,
|
|
password=None,
|
|
sessionid=None,
|
|
base_url="blobstore://.",
|
|
path=None,
|
|
skipbuild=False,
|
|
includelogs=False,
|
|
biospassword=None,
|
|
is_redfish=False,
|
|
proxy=None,
|
|
ssl_cert=None,
|
|
user_ca_cert_data=None,
|
|
json_out=False,
|
|
login_otp=None,
|
|
log_dir=None,
|
|
session_location=None,
|
|
):
|
|
"""Performs a login on a the server specified by the keyword arguments. Will also create
|
|
a monolith, client, and update the compatibility classes for the app instance. If base_url
|
|
is not included the login is assumed to be locally on the OS.
|
|
|
|
:param username: The user name required to login to server.
|
|
:type: str
|
|
:param password: The password credentials required to login.
|
|
:type password: str
|
|
:param base_url: The redfish host name or ip address to login to.
|
|
:type base_url: str
|
|
:param path: The path to initiate the monolith crawl from. If None, it will start from the
|
|
root. See monolith documentation on how the path is used.
|
|
:type path: str
|
|
:param proxy: The proxy required for connection (if any).
|
|
:type proxy: str
|
|
:param ssl_cert: The path to the CA bundle or SSL certificate to use with connection
|
|
(if any).
|
|
:type ssl_cert: str
|
|
:param user_ca_cert_data: Dictionary of user certificate data for iLO Certificate-based
|
|
authentication including iLO User TLS certificate, iLO User CA Root Key, \
|
|
iLO User CA Root Key Password (for encrypted CAs)
|
|
:type: user_ca_pass: str
|
|
:param skipbuild: The flag to determine monolith download. If True, monolith will be
|
|
initiated empty, if False will build the monolith.
|
|
:type skipbuild: bool
|
|
:param includelogs: The flag to determine id logs should be downloaded in the crawl.
|
|
:type includelogs: bool
|
|
:param biospassword: The BIOS password for the server if set.
|
|
:type biospassword: str
|
|
:param is_redfish: If True, a Redfish specific header (OData) will be
|
|
added to every request. Only required if the system has both LegacyREST and Redfish.
|
|
:type is_redfish: bool
|
|
"""
|
|
|
|
self.typepath.getgen(
|
|
url=base_url,
|
|
username=username,
|
|
password=password,
|
|
sessionid=sessionid,
|
|
ca_cert_data=user_ca_cert_data,
|
|
proxy=proxy,
|
|
isredfish=is_redfish,
|
|
login_otp=login_otp,
|
|
log_dir=log_dir,
|
|
session_location=session_location,
|
|
)
|
|
if user_ca_cert_data and self.typepath.iloversion < 5.23:
|
|
raise IncompatibleiLOVersionError(
|
|
"Certificate based login is incompatible with this " "iLO version: %s\n" % self.typepath.iloversion
|
|
)
|
|
is_redfish = self.typepath.updatedefinesflag(redfishflag=is_redfish)
|
|
|
|
if self.redfishinst and self.redfishinst.session_key:
|
|
self.logout()
|
|
|
|
self.redfishinst = RestClient(
|
|
base_url=base_url,
|
|
username=username,
|
|
password=password,
|
|
session_key=sessionid,
|
|
default_prefix=self.typepath.defs.startpath,
|
|
biospassword=biospassword,
|
|
is_redfish=is_redfish,
|
|
proxy=proxy,
|
|
ca_cert_data=user_ca_cert_data,
|
|
login_otp=login_otp,
|
|
log_dir=log_dir,
|
|
session_location=session_location,
|
|
)
|
|
|
|
self.current_client.login(self.current_client.auth_type)
|
|
resp = self.get_handler(self.typepath.defs.managerpath, service=False, silent=True).dict
|
|
ver = resp["FirmwareVersion"]
|
|
if "iLO" not in ver:
|
|
self.typepath.ilover = ver.split(" ")[0]
|
|
self.typepath.ilover = float(self.typepath.ilover.replace(".", ""))
|
|
inittime = time.time()
|
|
|
|
self._build_monolith(path=path, includelogs=includelogs, skipbuild=skipbuild, json_out=json_out)
|
|
endtime = time.time()
|
|
|
|
if self.verbose > 1:
|
|
sys.stdout.write("Monolith build process time: %s\n" % (endtime - inittime))
|
|
self.save()
|
|
if not self.monolith:
|
|
self.monolith.update_member(
|
|
resp=self.current_client.root,
|
|
path=self.typepath.defs.startpath,
|
|
init=False,
|
|
)
|
|
|
|
return self.current_client
|
|
|
|
def logout(self, url=None):
|
|
"""Performs a logout of the server and prepares the app for another system, setting app
|
|
variables to default values.
|
|
|
|
:param url: The URL for the logout request. Only needed when using a cache.
|
|
:type url: str
|
|
"""
|
|
sessionlocs = []
|
|
self._validationmanager = None
|
|
self._iloversion = None
|
|
|
|
try:
|
|
self.monolith.killthreads()
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
self.current_client.logout()
|
|
except Exception:
|
|
sessionlocs = self._cm.logout_del_function(url)
|
|
else:
|
|
self._cm.logout_del_function(url)
|
|
|
|
for session in sessionlocs:
|
|
try:
|
|
self.delete_handler(session[0], silent=True, service=True)
|
|
except:
|
|
pass
|
|
self.redfishinst = None
|
|
|
|
cachedir = self.cachedir
|
|
if cachedir:
|
|
try:
|
|
shutil.rmtree(cachedir)
|
|
except Exception:
|
|
pass
|
|
|
|
def select(self, selector=None, fltrvals=(None, None), path_refresh=False):
|
|
"""Selects instances based on selector and filter values. The select specified is saved in
|
|
the app for further use. If another selector is sent, it overwrites the current one.
|
|
|
|
:param selector: The type (@odata.type for Redfish) to select.
|
|
:type selector: str
|
|
:param fltrvals: The filter values for the select operation (Key,Val). If a selector returns
|
|
multiple instances fltrvals can filter the instances by a key/value pair,
|
|
limiting the returned instances to the one you want.
|
|
:type fltrvals: tuple
|
|
:param path_refresh: The flag to reload the selected instances. If True, each instance will be
|
|
grabbed again from the server to make sure responses are up to date.
|
|
:type path_refresh: bool
|
|
:returns: A list of selected monolith member instances.
|
|
:rtype: RisMonolithMemberv100
|
|
"""
|
|
if not selector:
|
|
selector = self.selector
|
|
# Still nothing selected and selector is NULL
|
|
if not selector:
|
|
raise NothingSelectedError()
|
|
selector = self.typepath.modifyselectorforgen(selector)
|
|
instances = self._getinstances(selector=selector, path_refresh=path_refresh)
|
|
val = fltrvals[1].strip("'\"") if isinstance(fltrvals[1], six.string_types) else fltrvals[1]
|
|
instances = [
|
|
inst
|
|
for inst in instances
|
|
if not fltrvals[0] or navigatejson(fltrvals[0].split("/"), copy.deepcopy(inst.dict), val)
|
|
]
|
|
if any(instances):
|
|
self.selector = selector
|
|
self.save()
|
|
return instances
|
|
|
|
errmsg = (
|
|
"Unable to locate instance for '{0}' and filter '{1}={2}'".format(selector, fltrvals[0], fltrvals[1])
|
|
if fltrvals[0] and fltrvals[1]
|
|
else "Unable to locate instance for {}\n".format(selector)
|
|
)
|
|
|
|
raise InstanceNotFoundError(errmsg)
|
|
|
|
def types(self, fulltypes=False):
|
|
"""Returns a list of types available to be queried and selected with monolith.
|
|
|
|
:param fulltypes: Flag to determine if types return Redfish full name, if False will return
|
|
a shortened version of the type string.
|
|
:type fulltypes: bool
|
|
:returns: A list of type strings.
|
|
:rtype: list
|
|
"""
|
|
instances = list()
|
|
|
|
if self.monolith:
|
|
monolith = self.monolith
|
|
rdirtype = next(monolith.gettypename(self.typepath.defs.resourcedirectorytype), None)
|
|
|
|
if not rdirtype:
|
|
for inst in monolith.iter():
|
|
if not any([x for x in ["ExtendedError", "object", "string"] if x in inst.type]):
|
|
instances.append(inst.type)
|
|
else:
|
|
for instance in monolith.iter(rdirtype):
|
|
for item in instance.resp.dict["Instances"]:
|
|
if (
|
|
item
|
|
and instance._typestring in list(item.keys())
|
|
and "ExtendedError" not in item[instance._typestring]
|
|
):
|
|
if not fulltypes and instance._typestring == "@odata.type":
|
|
tval = item["@odata.type"].split("#")
|
|
tval = tval[-1].split(".")[:-1]
|
|
tval = ".".join(tval)
|
|
instances.append(tval)
|
|
elif item:
|
|
instances.append(item[instance._typestring])
|
|
return instances
|
|
|
|
def getprops(
|
|
self,
|
|
selector=None,
|
|
props=None,
|
|
nocontent=None,
|
|
skipnonsetting=True,
|
|
remread=False,
|
|
insts=None,
|
|
):
|
|
"""Gets properties from a specified selector. If no selector is specified, uses the selector
|
|
property in the app class. Instead of a selector a list of instances to search can be used
|
|
instead. If both **selector** and **insts** are passed, **insts** is used.
|
|
|
|
Specific values for multi-level dictionaries can be returned by passing each key
|
|
separated by a "/" Ex: Key/Sub-Key/Sub-Sub-Key
|
|
|
|
:param selector: The type selection for the get operation.
|
|
:type selector: str
|
|
:param skipnonsetting: Flag to remove non settings path.
|
|
:type skipnonsetting: bool
|
|
:param nocontent: Keys not found are added to the list provided.
|
|
:type nocontent: list
|
|
:param remread: Flag to remove readonly properties.
|
|
:type remread: bool
|
|
:param props: The keys to search for within current selection.
|
|
:type props: list
|
|
:param insts: List of RisMonolithMemberv100 to be searched for specific keys.
|
|
:type insts: list
|
|
:returns: A list of properties found in dictionary form.
|
|
:rtype: list
|
|
"""
|
|
results = list()
|
|
nocontent = set() if nocontent is None else nocontent
|
|
if props:
|
|
if isinstance(props, list):
|
|
noprop = {prop: False for prop in props}
|
|
else:
|
|
noprop = {props: False}
|
|
else:
|
|
noprop = {}
|
|
instances = insts if insts else self._getinstances(selector=selector)
|
|
instances = skipnonsettingsinst(instances) if skipnonsetting else instances
|
|
|
|
if selector != "UpdateService.":
|
|
if not instances or len(instances) == 0:
|
|
if not selector:
|
|
raise NothingSelectedError()
|
|
|
|
for instance in instances:
|
|
currdict = instance.dict
|
|
for patch in instance.patches:
|
|
currdict = jsonpatch.apply_patch(currdict, patch)
|
|
_ = self.removereadonlyprops(currdict, emptyraise=True) if remread else None
|
|
temp_dict = dict()
|
|
if props:
|
|
if isinstance(props, six.string_types):
|
|
props = [props]
|
|
for prop in props:
|
|
copydict = copy.deepcopy(currdict)
|
|
propsdict = navigatejson(prop.split("/"), copydict)
|
|
if propsdict is None:
|
|
continue
|
|
noprop[prop] = True
|
|
merge_dict(temp_dict, propsdict)
|
|
if temp_dict:
|
|
if temp_dict not in results:
|
|
results.append(temp_dict)
|
|
else:
|
|
results.append(currdict)
|
|
if props:
|
|
_ = [nocontent.add(prop) for prop in props if not noprop[prop]]
|
|
return results
|
|
|
|
def info(self, selector=None, props=None, dumpjson=True, latestschema=False):
|
|
"""Gets schema information for properties from a specified selector. If no selector is
|
|
specified, uses the selector property in the app class. If no properties are specified
|
|
the entire schema dictionary is returned in a list.
|
|
|
|
:param selector: The type selection for the info operation.
|
|
:type selector: str
|
|
:param props: The keys to gather schema data for within current selection.
|
|
:type props: str or list
|
|
:param dumpjson: Flag to determine if output should be human readable or json schema.
|
|
:type dumpjson: bool
|
|
:param latestschema: Flag to determine if we should drop the schema version when we try to
|
|
match schema information. If True, the version will be dropped.
|
|
:type latestschema: bool
|
|
:returns: A list of property schema information if dumpjson is True or string if dumpjson is
|
|
False.
|
|
:rtype: list or string
|
|
"""
|
|
model = None
|
|
outdata = ""
|
|
nokey = False
|
|
results = None
|
|
typestring = self.typepath.defs.typestring
|
|
iloversion = self.getiloversion()
|
|
if not iloversion:
|
|
return results
|
|
instances = self._getinstances(selector)
|
|
attributeregistry = getattributeregistry(instances)
|
|
instances = skipnonsettingsinst(instances)
|
|
|
|
if not instances or len(instances) == 0:
|
|
raise NothingSelectedError()
|
|
|
|
for inst in instances:
|
|
bsmodel = None
|
|
currdict = inst.resp.dict
|
|
proppath = inst.resp.getheader("Link").split(";")[0].strip("<>") if inst.resp.getheader("Link") else None
|
|
seldict = {}
|
|
if not props:
|
|
model, bsmodel = self.get_model(currdict, attributeregistry, latestschema, proppath=proppath)
|
|
results = model
|
|
break
|
|
if isinstance(props, six.string_types):
|
|
props = props.split("/") if "/" in props else props
|
|
props = [props] if not isinstance(props, (list, tuple)) else props
|
|
seldict = navigatejson(props, copy.deepcopy(currdict))
|
|
if seldict is None:
|
|
nokey = True
|
|
continue
|
|
if self.typepath.defs.typestring in currdict:
|
|
seldict[typestring] = currdict[typestring]
|
|
model, bsmodel = self.get_model(
|
|
currdict,
|
|
attributeregistry,
|
|
latestschema,
|
|
newarg=props[:-1],
|
|
proppath=proppath,
|
|
)
|
|
if not model and not bsmodel:
|
|
errmsg = "/".join(props)
|
|
warning_handler(
|
|
"Unable to locate registry model or " "No data available for entry: {}\n".format(errmsg)
|
|
)
|
|
continue
|
|
found = model.get_validator(props[-1]) if model else None
|
|
found = bsmodel.get_validator(props[-1]) if not found and bsmodel else found
|
|
outdata = found if found and dumpjson else found.print_help(props[-1]) if found else outdata
|
|
|
|
if outdata or results:
|
|
return outdata if outdata else results
|
|
|
|
errmsg = (
|
|
"Entry {} not found in current selection\n".format("/".join(props))
|
|
if nokey
|
|
else "Entry {} not found in current" " selection\n".format("/".join(props))
|
|
)
|
|
warning_handler(errmsg)
|
|
|
|
def loadset(
|
|
self,
|
|
seldict=None,
|
|
fltrvals=(None, None),
|
|
diffonly=False,
|
|
latestschema=False,
|
|
uniqueoverride=False,
|
|
selector=None,
|
|
):
|
|
"""Creates json patches in monolith if the supplied dictionary passes schema validation.
|
|
In the event schemas are unavailable the patches are always added. Patches that are created
|
|
this way are not sent to the server until the :meth:`commit` function is called, sending the
|
|
patches to the server. A list of patches that have not been sent to the server can be
|
|
returned with the :meth:`status` function.
|
|
|
|
:param selector: The type selection for the loadset operation.
|
|
:type selector: str
|
|
:param seldict: Dictionary with the patches to apply to the selected instances.
|
|
:type seldict: dict
|
|
:param fltrvals: The filter values for the operation (Key,Val). If a selector returns
|
|
multiple instances fltrvals can filter the instances by a key/value pair,
|
|
limiting the returned instances to the one you want. If no filter is
|
|
supplied the patch dictionary will be applied to all instances.
|
|
:type fltrvals: tuple
|
|
:param latestschema: Flag to determine if we should drop the schema version when we try to
|
|
match schema information. If True, the version will be dropped.
|
|
:type latestschema: bool
|
|
:param diffonly: flag to differentiate only existing properties.
|
|
:type diffonly: bool
|
|
:param uniqueoverride: Flag to determine if system unique properties should also be patched.
|
|
If this is True, then unique properties will be patched.
|
|
:type uniqueoverride: bool
|
|
:returns: returns a list of properties that have successfully been set
|
|
"""
|
|
results = list()
|
|
nochangesmade = False
|
|
settingskipped = [False]
|
|
|
|
selector = self.selector if not selector else selector
|
|
instances = self.select(selector=selector, fltrvals=fltrvals)
|
|
attributeregistry = getattributeregistry(instances=instances)
|
|
instances = skipnonsettingsinst(instances=instances)
|
|
|
|
if not instances or len(instances) == 0:
|
|
raise NothingSelectedSetError()
|
|
|
|
for instance in instances:
|
|
if validate_headers(instance, verbose=self.verbose):
|
|
continue
|
|
else:
|
|
nochangesmade = True
|
|
|
|
currdict = instance.resp.dict
|
|
if "@odata.id" in seldict:
|
|
if currdict["@odata.id"] != seldict["@odata.id"]:
|
|
continue
|
|
diff_resp = diffdict(
|
|
newdict=copy.deepcopy(seldict),
|
|
oridict=copy.deepcopy(currdict),
|
|
settingskipped=settingskipped,
|
|
)
|
|
|
|
iloversion = self.getiloversion()
|
|
if iloversion:
|
|
proppath = (
|
|
instance.resp.getheader("Link").split(";")[0].strip("<>")
|
|
if instance.resp.getheader("Link")
|
|
else None
|
|
)
|
|
try:
|
|
self._validatechanges(
|
|
instance=instance,
|
|
attributeregistry=attributeregistry,
|
|
newdict=diff_resp,
|
|
oridict=currdict,
|
|
unique=uniqueoverride,
|
|
latestschema=latestschema,
|
|
proppath=proppath,
|
|
)
|
|
except SchemaValidationError:
|
|
LOGGER.error("Cannot validate changes, error found in schema.")
|
|
|
|
patches = jsonpatch.make_patch(currdict, diff_resp)
|
|
|
|
if patches:
|
|
torem = []
|
|
_ = [torem.append(patch) for patch in patches.patch if patch["op"] == "remove"]
|
|
_ = [patches.patch.remove(patch) for patch in torem]
|
|
|
|
for ind, item in enumerate(instance.patches):
|
|
ppath = item.patch[0]["path"] if hasattr(item, "patch") else item[0]["path"]
|
|
# ppath = ["path"](getattr(item, "patch"), item)[0]["path"]
|
|
jpath = jsonpointer.JsonPointer(ppath.lower())
|
|
jval = jpath.resolve(seldict, default="kasjdk?!")
|
|
if not jval == "kasjdk?!":
|
|
del instance.patches[ind]
|
|
|
|
if patches:
|
|
for patch in patches.patch:
|
|
forprint = patch["value"] if "value" in patch else (patch["op"] + " " + patch["from"])
|
|
results.append({patch["path"][1:]: forprint})
|
|
if "Managers/1/EthernetInterfaces/1" not in instance.path:
|
|
self.monolith.path(instance.path).patches.append(patches)
|
|
else:
|
|
nochangesmade = True
|
|
|
|
if not nochangesmade:
|
|
return results
|
|
elif settingskipped[0] is True:
|
|
raise LoadSkipSettingError()
|
|
else:
|
|
return results
|
|
|
|
def status(self):
|
|
"""Returns all pending changes that have not been committed yet."""
|
|
iloversion = self.getiloversion()
|
|
|
|
finalresults = list()
|
|
monolith = self.monolith
|
|
(_, _) = self.get_selection(setenable=True)
|
|
attrreg = getattributeregistry([ele for ele in monolith.iter() if ele])
|
|
for instance in monolith.iter():
|
|
results = list()
|
|
|
|
if not (instance.patches and len(instance.patches) > 0):
|
|
continue
|
|
for item in instance.patches:
|
|
if isinstance(item, list):
|
|
results.extend(jsonpatch.JsonPatch(item))
|
|
else:
|
|
results.extend(item)
|
|
|
|
currdict = instance.resp.dict
|
|
itemholder = list()
|
|
for mainitem in results:
|
|
item = copy.deepcopy(mainitem)
|
|
|
|
if iloversion:
|
|
_, bsmodel = self.get_model(currdict, attrreg)
|
|
if bsmodel:
|
|
prop = item["path"][1:].split("/")[-1]
|
|
validator = bsmodel.get_validator(prop)
|
|
if validator:
|
|
if isinstance(validator, redfish.ris.validation.PasswordValidator):
|
|
item["value"] = "******"
|
|
|
|
itemholder.append(item)
|
|
if itemholder:
|
|
finalresults.append({instance.maj_type + "(" + instance.path + ")": itemholder})
|
|
|
|
return finalresults
|
|
|
|
def commit(self):
|
|
"""Applies all pending json patches to the server.
|
|
|
|
:yields: Two strings.
|
|
|
|
1. Path being PATCHed
|
|
2. True if an error occurred during the PATCH, False if no error.
|
|
"""
|
|
# Protect iLO Network Interface changes.
|
|
instances = [
|
|
inst for inst in self.monolith.iter() if inst.patches and "Managers/1/EthernetInterfaces/1" not in inst.path
|
|
]
|
|
|
|
if not instances or len(instances) == 0:
|
|
raise NothingSelectedError()
|
|
|
|
for instance in instances:
|
|
if validate_headers(instance, verbose=self.verbose):
|
|
continue
|
|
|
|
currdict = dict()
|
|
oridict = instance.resp.dict
|
|
totpayload = dict()
|
|
# apply patches to represent current edits
|
|
for patches in instance.patches:
|
|
if not self._iloversion:
|
|
self._iloversion = self.getiloversion()
|
|
if self._iloversion < 5.130:
|
|
self._checkforetagchange(instance=instance)
|
|
fulldict = jsonpatch.apply_patch(oridict, patches)
|
|
for patch in patches:
|
|
currdict = copy.deepcopy(fulldict)
|
|
patchpath = patch["path"]
|
|
pobj = jsonpointer.JsonPointer(patchpath)
|
|
indpayloadcount = 0
|
|
for item in pobj.parts:
|
|
payload = pobj.walk(currdict, item)
|
|
indpayloadcount = indpayloadcount + 1
|
|
if isinstance(payload, list):
|
|
break
|
|
else:
|
|
if not isinstance(payload, dict):
|
|
break
|
|
currdict = copy.deepcopy(payload)
|
|
indices = pobj.parts[:indpayloadcount]
|
|
createdict = lambda x, y: {x: y}
|
|
while len(indices):
|
|
payload = createdict(indices.pop(), payload)
|
|
merge_dict(totpayload, payload)
|
|
currdict = copy.deepcopy(totpayload)
|
|
if "PersistentBootConfigOrder" in currdict and type(currdict) is dict:
|
|
refdict = list(set(currdict["PersistentBootConfigOrder"]))
|
|
items = list(currdict["PersistentBootConfigOrder"])
|
|
items_to_keep = items[: int(len(refdict))]
|
|
currdict = {"PersistentBootConfigOrder": items_to_keep}
|
|
if currdict:
|
|
yield instance.resp.request.path
|
|
|
|
put_path = instance.resp.request.path
|
|
etag = self.monolith.paths[put_path].etag
|
|
headers = dict([("If-Match", etag)]) if self._iloversion > 5.130 else None
|
|
for key, value in currdict.items():
|
|
if key == "OldAdminPassword" and (value != ""):
|
|
if (value == "null") or (value == "none") or (value == "None"):
|
|
currdict["OldAdminPassword"] = ""
|
|
else:
|
|
hash = hashlib.sha256(value.encode()).hexdigest().upper()
|
|
headers = dict([("X-HPRESTFULAPI-AuthToken", hash)])
|
|
elif key == "AdminPassword" and (
|
|
(value is None) or (value == "none") or (value == "None") or (value == "null")
|
|
):
|
|
currdict["AdminPassword"] = ""
|
|
try:
|
|
self.patch_handler(
|
|
put_path,
|
|
currdict,
|
|
optionalpassword=self.current_client.bios_password,
|
|
headers=headers,
|
|
)
|
|
except IloResponseError:
|
|
yield True # Failure
|
|
else:
|
|
yield False # Success
|
|
|
|
def patch_handler(
|
|
self,
|
|
put_path,
|
|
body,
|
|
headers=None,
|
|
silent=False,
|
|
service=False,
|
|
optionalpassword=None,
|
|
noauth=False,
|
|
):
|
|
"""Performs the client HTTP PATCH operation with monolith and response handling support.
|
|
Response handling will output to logger or string depending on showmessages app argument.
|
|
|
|
:param put_path: The REST path to perform the patch operation on.
|
|
:type put_path: str
|
|
:param body: the body to perform the operation with.
|
|
:type body: dict
|
|
:param headers: Any additional headers to be added to the request.
|
|
:type headers: dict
|
|
:param optionalpassword: The bios password if it is required for the operation.
|
|
:type optionalpassword: str
|
|
:param silent: If False response will be parsed based on service flag and output to a log or
|
|
stdout. If True response will not be parsed and no message output or error
|
|
messages raised from the response handler.
|
|
:type silent: bool
|
|
:param service: When handling the response, if True registries will be gathered and a full,
|
|
response will be output if False they will not and response handler will
|
|
instead return a generic message.
|
|
:type service: bool
|
|
:returns: A :class:`redfish.rest.containers.RestResponse` object containing response data
|
|
"""
|
|
(put_path, body) = self._checkpostpatch(body=body, path=put_path, patch=True)
|
|
|
|
if optionalpassword:
|
|
self.current_client.bios_password = optionalpassword
|
|
|
|
results = self.current_client.patch(put_path, body=body, headers=headers, noauth=noauth)
|
|
|
|
if results and getattr(results, "status", None) and results.status == 401:
|
|
raise SessionExpired()
|
|
|
|
self._modifiedpath(results, replace=True)
|
|
|
|
# if results and getattr(results, "status", None) and results.status == 412:
|
|
if results and hasattr(results, "status") and results.status == 412:
|
|
self._updatemono(path=put_path, path_refresh=True)
|
|
|
|
if not silent and hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose
|
|
)
|
|
|
|
return results
|
|
|
|
def get_handler(
|
|
self,
|
|
get_path,
|
|
sessionid=None,
|
|
silent=False,
|
|
uncache=False,
|
|
headers=None,
|
|
service=False,
|
|
username=None,
|
|
password=None,
|
|
base_url=None,
|
|
stderr_flag=False,
|
|
noauth=False,
|
|
):
|
|
"""Performs the client HTTP GET operation with monolith and response handling support.
|
|
Response handling will output to logger or string depending on showmessages app argument.
|
|
|
|
:param put_path: The REST path to perform the get operation on.
|
|
:type put_path: str
|
|
:param uncache: flag to not store the data downloaded into monolith.
|
|
:type uncache: bool
|
|
:param headers: Any additional headers to be added to the request.
|
|
:type headers: dict
|
|
:param silent: If False response will be parsed based on service flag and output to a log or
|
|
stdout. If True response will not be parsed and no message output or error
|
|
messages raised from the response handler.
|
|
:type silent: bool
|
|
:param service: When handling the response, if True registries will be gathered and a full,
|
|
response will be output if False they will not and response handler will
|
|
instead return a generic message.
|
|
:type service: bool
|
|
:returns: A :class:`redfish.rest.containers.RestResponse` object
|
|
"""
|
|
try:
|
|
results = self.current_client.get(get_path, headers=headers, noauth=noauth)
|
|
except UndefinedClientError:
|
|
if sessionid:
|
|
self.redfishinst = RestClient(
|
|
sessionid=sessionid,
|
|
username=username,
|
|
password=password,
|
|
base_url=base_url,
|
|
noauth=noauth,
|
|
)
|
|
if not headers:
|
|
headers = dict()
|
|
headers["X-Auth-Token"] = sessionid
|
|
results = self.current_client.get(get_path, headers=headers)
|
|
|
|
if results and getattr(results, "status", None) and results.status == 404:
|
|
if not silent:
|
|
if hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose, stderr_flag=stderr_flag
|
|
)
|
|
else:
|
|
print_handler("[" + str(results.status) + "]" + " The operation completed successfully.\n")
|
|
return results
|
|
|
|
if results and results.status == 200 and sessionid:
|
|
if not silent:
|
|
if hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose, stderr_flag=stderr_flag
|
|
)
|
|
else:
|
|
print_handler("[" + str(results.status) + "]" + " The operation completed successfully.\n")
|
|
return results
|
|
|
|
if not uncache and results.status == 200 and not sessionid:
|
|
if self.monolith:
|
|
self.monolith.update_member(resp=results, path=get_path, init=False)
|
|
|
|
if results and getattr(results, "status", None) and results.status == 401:
|
|
raise SessionExpired()
|
|
|
|
if not silent:
|
|
if hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose, stderr_flag=stderr_flag
|
|
)
|
|
else:
|
|
print_handler("[" + str(results.status) + "]" + " The operation completed successfully.\n")
|
|
|
|
return results
|
|
|
|
def post_handler(self, put_path, body, headers=None, silent=False, service=False, noauth=False):
|
|
"""Performs the client HTTP POST operation with monolith and response handling support.
|
|
Response handling will output to logger or string depending on showmessages app argument.
|
|
|
|
:param put_path: The REST path to perform the post operation on.
|
|
:type put_path: str
|
|
:param body: the body to perform the operation with.
|
|
:type body: dict
|
|
:param headers: Any additional headers to be added to the request.
|
|
:type headers: dict
|
|
:param silent: If False response will be parsed based on service flag and output to a log or
|
|
stdout. If True response will not be parsed and no message output or error
|
|
messages raised from the response handler.
|
|
:type silent: bool
|
|
:param service: When handling the response, if True registries will be gathered and a full,
|
|
response will be output if False they will not and response handler will
|
|
instead return a generic message.
|
|
:type service: bool
|
|
:returns: A :class:`redfish.rest.containers.RestResponse` object containing response data
|
|
"""
|
|
(put_path, body) = self._checkpostpatch(body=body, path=put_path)
|
|
|
|
results = self.current_client.post(put_path, body=body, headers=headers, noauth=noauth)
|
|
|
|
if results and getattr(results, "status", None) and results.status == 401:
|
|
raise SessionExpired()
|
|
|
|
self._modifiedpath(results)
|
|
|
|
if results.status == 400 and results.dict is None:
|
|
return results
|
|
|
|
if not silent and hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose
|
|
)
|
|
|
|
return results
|
|
|
|
def put_handler(
|
|
self, put_path, body, headers=None, silent=False, optionalpassword=None, service=False, noauth=False
|
|
):
|
|
"""Performs the client HTTP PUT operation with monolith and response handling support.
|
|
Response handling will output to logger or string depending on showmessages app argument.
|
|
|
|
:param put_path: The REST path to perform the put operation on.
|
|
:type put_path: str
|
|
:param body: the body to perform the operation with.
|
|
:type body: dict
|
|
:param headers: Any additional headers to be added to the request.
|
|
:type headers: dict
|
|
:param optionalpassword: The bios password if it is required for the operation.
|
|
:type optionalpassword: str
|
|
:param silent: If False response will be parsed based on service flag and output to a log or
|
|
stdout. If True response will not be parsed and no message output or error
|
|
messages raised from the response handler.
|
|
:type silent: bool
|
|
:param service: When handling the response, if True registries will be gathered and a full,
|
|
response will be output if False they will not and response handler will
|
|
instead return a generic message.
|
|
:type service: bool
|
|
:returns: A :class:`redfish.rest.containers.RestResponse` object containing response data
|
|
"""
|
|
if optionalpassword:
|
|
self.current_client.bios_password = optionalpassword
|
|
results = self.current_client.put(put_path, body=body, headers=headers, noauth=noauth)
|
|
if results and getattr(results, "status", None) and results.status == 401:
|
|
raise SessionExpired()
|
|
|
|
self._modifiedpath(results, replace=True)
|
|
|
|
if not silent and hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose
|
|
)
|
|
|
|
return results
|
|
|
|
def delete_handler(self, put_path, headers=None, silent=False, service=False):
|
|
"""Performs the client HTTP DELETE operation with monolith and response handling support.
|
|
Response handling will output to logger or string depending on showmessages app argument.
|
|
|
|
:param put_path: The REST path to perform the delete operation on.
|
|
:type put_path: str
|
|
:param headers: Any additional headers to be added to the request.
|
|
:type headers: dict
|
|
:param silent: If False response will be parsed based on service flag and output to a log or
|
|
stdout. If True response will not be parsed and no message output or error
|
|
messages raised from the response handler.
|
|
:type silent: bool
|
|
:param service: When handling the response, if True registries will be gathered and a full,
|
|
response will be output if False they will not and response handler will
|
|
instead return a generic message.
|
|
:type service: bool
|
|
:returns: A :class:`redfish.rest.containers.RestResponse` object containing response data
|
|
"""
|
|
results = self.current_client.delete(put_path, headers=headers)
|
|
|
|
if results and getattr(results, "status", None) and results.status == 401:
|
|
raise SessionExpired()
|
|
self._modifiedpath(results, delete=True)
|
|
|
|
if not silent and hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose
|
|
)
|
|
|
|
return results
|
|
|
|
def head_handler(self, put_path, silent=False, service=False):
|
|
"""Performs the client HTTP HEAD operation with monolith and response handling support.
|
|
Response handling will output to logger or string depending on showmessages app argument.
|
|
|
|
:param put_path: The REST path to perform the head operation on.
|
|
:type put_path: str
|
|
:param silent: If False response will be parsed based on service flag and output to a log or
|
|
stdout. If True response will not be parsed and no message output or error
|
|
messages raised from the response handler.
|
|
:type silent: bool
|
|
:param service: When handling the response, if True registries will be gathered and a full,
|
|
response will be output if False they will not and response handler will
|
|
instead return a generic message.
|
|
:type service: bool
|
|
:returns: A :class:`redfish.rest.containers.RestResponse` object containing response data
|
|
"""
|
|
results = self.current_client.head(put_path)
|
|
|
|
if results and getattr(results, "status", None) and results.status == 401:
|
|
raise SessionExpired()
|
|
|
|
if not silent and hasattr(self.typepath.defs, "messageregistrytype"):
|
|
ResponseHandler(self.validationmanager, self.typepath.defs.messageregistrytype).output_resp(
|
|
results, dl_reg=service, verbosity=self.verbose
|
|
)
|
|
|
|
return results
|
|
|
|
def removereadonlyprops(self, currdict, emptyraise=False, removeunique=True, specify_props=None):
|
|
"""Remove read only properties from a dictionary. Requires schemas to be available.
|
|
|
|
:param currdict: The dictionary to remove read only properties from.
|
|
:type currdict: dictionary
|
|
:param emptyraise: Flag to raise an empty error for handling and failure to parse.
|
|
:type emptyraise: boolean
|
|
:type removeunique: Flag to remove system unique values as well as read only.
|
|
:type removeunique: boolean
|
|
:parm specify_props: Optionally set list of properties to be removed instead of the default.
|
|
:type specify_props: list
|
|
"""
|
|
try:
|
|
type_str = self.typepath.defs.typestring
|
|
currtype = currdict.get(type_str, None)
|
|
oridict = copy.deepcopy(currdict)
|
|
if specify_props:
|
|
templist = specify_props
|
|
else:
|
|
templist = [
|
|
"Modified",
|
|
"Type",
|
|
"Description",
|
|
"Status",
|
|
"links",
|
|
"SettingsResult",
|
|
"Attributes",
|
|
"@odata.context",
|
|
"@odata.type",
|
|
"@odata.id",
|
|
"@odata.etag",
|
|
"Links",
|
|
"Actions",
|
|
"AvailableActions",
|
|
"BiosVersion",
|
|
"AddressOrigin",
|
|
]
|
|
# Attributes removed and readded later as a validation workaround
|
|
currdict = iterateandclear(currdict, templist)
|
|
iloversion = self.getiloversion()
|
|
if not iloversion:
|
|
return currdict
|
|
self.validationmanager.validatedict(
|
|
currdict,
|
|
currtype=currtype,
|
|
monolith=self.monolith,
|
|
unique=removeunique,
|
|
searchtype=None,
|
|
)
|
|
if oridict.get("Attributes", None):
|
|
currdict["Attributes"] = oridict["Attributes"]
|
|
return currdict
|
|
except:
|
|
if emptyraise is True:
|
|
raise EmptyRaiseForEAFP()
|
|
elif emptyraise == "pass":
|
|
pass
|
|
else:
|
|
raise
|
|
|
|
def getidbytype(self, tpe):
|
|
"""Return a list of URIs that correspond to the supplied type string.
|
|
|
|
:param tpe: type string to search for.
|
|
:type tpe: string.
|
|
"""
|
|
urls = list()
|
|
val = next(self.monolith.gettypename(tpe), None)
|
|
urls.extend(self.monolith.typesadded[val] if val else [])
|
|
return urls
|
|
|
|
def getcollectionmembers(self, path, fullresp=False):
|
|
"""Returns collection/item lists of the provided path.
|
|
|
|
:param path: path to return.
|
|
:type path: string.
|
|
:param fullresp: Return full json data instead of only members.
|
|
:type path: bool.
|
|
:returns: list of collection members
|
|
"""
|
|
if (
|
|
self.typepath.defs.isgen10
|
|
and hasattr(self.typepath, "gencompany")
|
|
and self.typepath.gencompany
|
|
and "?$expand=." not in path
|
|
):
|
|
path += "?$expand=." if path.endswith("/") else "/?$expand=."
|
|
|
|
members = self.get_handler(path, service=True, silent=True)
|
|
if members and not fullresp:
|
|
try:
|
|
members = members.dict["Members"] if self.typepath.defs.isgen10 else members.dict["Current"]
|
|
except KeyError:
|
|
members = members
|
|
elif fullresp:
|
|
members = [members.dict]
|
|
|
|
return members
|
|
|
|
def getbiosfamilyandversion(self):
|
|
"""Function that returns the current BIOS version information."""
|
|
self._updatemono(currtype="ComputerSystem.", crawl=False)
|
|
|
|
try:
|
|
for inst in self.monolith.iter("ComputerSystem."):
|
|
if "Current" in inst.resp.obj["Bios"]:
|
|
oemjson = inst.resp.obj["Bios"]["Current"]
|
|
parts = oemjson["VersionString"].split(" ")
|
|
return parts[0], parts[1][1:]
|
|
else:
|
|
parts = inst.resp.obj["BiosVersion"].split(" ")
|
|
return parts[0], parts[1][1:]
|
|
except Exception:
|
|
pass
|
|
|
|
return None, None
|
|
|
|
def getiloversion(self, skipschemas=False):
|
|
"""Function that returns the current iLO version.
|
|
|
|
:param skipschemas: flag to determine whether to skip schema download. If False, this will
|
|
also verify if schemas are available.
|
|
:type skipschemas: bool
|
|
:returns: returns current iLO version
|
|
"""
|
|
iloversion = self._iloversion = self._iloversion if self._iloversion else self.typepath.iloversion
|
|
|
|
if not iloversion and hasattr(self.redfishinst, "iloversion"):
|
|
iloversion = self._iloversion = self.typepath.iloversion = self.redfishinst.iloversion
|
|
|
|
if (
|
|
hasattr(self.typepath, "gencompany")
|
|
and self.typepath.gencompany
|
|
and not self._iloversion
|
|
and not self.typepath.noschemas
|
|
):
|
|
self.monolith.load(self.typepath.defs.managerpath, crawl=False)
|
|
results = next(iter(self.getprops("Manager.", ["FirmwareVersion", "Firmware"])))
|
|
|
|
def quickdrill(_dict, key):
|
|
"""function to find key in nested dictionary"""
|
|
return _dict[key]
|
|
|
|
while isinstance(results, dict):
|
|
results = quickdrill(results, next(iter(results.keys())))
|
|
iloversionlist = results.replace("v", "").replace(".", "").split(" ")
|
|
try:
|
|
iloversion = float(".".join(iloversionlist[1:3]))
|
|
except:
|
|
iloversion = float(iloversionlist[0])
|
|
|
|
model = self.getprops("Manager.", ["Model"])
|
|
if model:
|
|
if next(iter(model))["Model"] == "iLO CM":
|
|
# Assume iLO 4 types in Moonshot
|
|
iloversion = None
|
|
|
|
self._iloversion = iloversion
|
|
elif (
|
|
hasattr(self.typepath, "gencompany") and not self.typepath.gencompany
|
|
): # Assume schemas are available somewhere in non-hpe redfish
|
|
self._iloversion = iloversion = 4.210
|
|
|
|
conf = None if not skipschemas else True
|
|
if not skipschemas:
|
|
if iloversion and iloversion >= 4.210:
|
|
conf = self._verifyschemasdownloaded(self.monolith)
|
|
elif iloversion and iloversion < 4.210:
|
|
warning_handler("Please upgrade to iLO 4 version 2.1 or above for schema support.")
|
|
else:
|
|
warning_handler("Schema support unavailable on the currently logged in system.")
|
|
|
|
return iloversion if iloversion and iloversion >= 4.210 and conf else None
|
|
|
|
def get_selection(self, selector=None, setenable=False, path_refresh=False):
|
|
"""Gathers instances and optionally the attributeregistry based on selector.
|
|
|
|
:param selector: The type selection for the get operation.
|
|
:type selector: str.
|
|
:param setenable: Flag to determine if registry should also be returned.
|
|
:type setenable: boolean.
|
|
:param path_refresh: Flag to reload the selected instances.
|
|
:type path_refresh: boolean.
|
|
:returns: returns a list of selected items
|
|
"""
|
|
instances = self._getinstances(selector=selector, path_refresh=path_refresh)
|
|
if setenable:
|
|
attributeregistryfound = getattributeregistry(instances=instances)
|
|
instances = skipnonsettingsinst(instances=instances)
|
|
return instances, attributeregistryfound
|
|
|
|
return instances
|
|
|
|
def create_save_header(self):
|
|
"""Adds save file headers to show what server the data came from.
|
|
|
|
:param selector: The type selection for the get save operation.
|
|
:type selector: str.
|
|
:param selectignore: Return the save header even if there isn't a selection to add it to.
|
|
:type selectignore: boolean
|
|
:returns: returns an header ordered dictionary
|
|
"""
|
|
instances = OrderedDict()
|
|
monolith = self.monolith
|
|
|
|
self._updatemono(currtype="ComputerSystem.", crawl=False)
|
|
self._updatemono(currtype=self.typepath.defs.biostype, crawl=False)
|
|
self._updatemono(currtype="Manager.", crawl=False)
|
|
|
|
instances["Comments"] = OrderedDict()
|
|
for instance in monolith.iter("ComputerSystem."):
|
|
if instance.resp.obj.get("Manufacturer"):
|
|
instances["Comments"]["Manufacturer"] = instance.resp.obj["Manufacturer"]
|
|
if instance.resp.obj.get("Model"):
|
|
instances["Comments"]["Model"] = instance.resp.obj["Model"]
|
|
try:
|
|
if instance.resp.obj["Oem"][self.typepath.defs.oemhp]["Bios"]["Current"]:
|
|
oemjson = instance.resp.obj["Oem"][self.typepath.defs.oemhp]["Bios"]["Current"]
|
|
instances["Comments"]["BIOSFamily"] = oemjson["Family"]
|
|
instances["Comments"]["BIOSDate"] = oemjson["Date"]
|
|
except KeyError:
|
|
pass
|
|
for instance in monolith.iter(self.typepath.defs.biostype):
|
|
try:
|
|
if getattr(instance.resp.obj, "Attributes", False):
|
|
if instance.resp.obj["Attributes"].get("SerialNumber"):
|
|
instances["Comments"]["SerialNumber"] = instance.resp.obj["Attributes"]["SerialNumber"]
|
|
if instance.resp.obj.get("SerialNumber"):
|
|
instances["Comments"]["SerialNumber"] = instance.resp.obj["SerialNumber"]
|
|
except KeyError:
|
|
pass
|
|
for instance in monolith.iter("Manager."):
|
|
if instance.resp.obj.get("FirmwareVersion"):
|
|
instances["Comments"]["iLOVersion"] = instance.resp.obj["FirmwareVersion"]
|
|
|
|
return instances
|
|
|
|
def download_path(self, paths, crawl=True, path_refresh=False):
|
|
"""Loads paths into the monolith.
|
|
|
|
:param paths: list of paths to download
|
|
:type paths: list
|
|
:param path_refresh: Flag to reload the paths or not.
|
|
:type path_refresh: bool.
|
|
:param crawl: Flag to determine if load should traverse found links.
|
|
:type crawl: boolean.
|
|
"""
|
|
if not paths:
|
|
return
|
|
try:
|
|
list(
|
|
map(
|
|
lambda x: self.monolith.load(
|
|
path=x,
|
|
init=False,
|
|
path_refresh=path_refresh,
|
|
crawl=crawl,
|
|
includelogs=True,
|
|
),
|
|
paths,
|
|
)
|
|
)
|
|
except Exception as excp:
|
|
try:
|
|
if excp.errno == 10053:
|
|
raise SessionExpired()
|
|
except:
|
|
raise excp
|
|
else:
|
|
raise excp
|
|
|
|
def get_model(self, currdict, attributeregistry, latestschema=None, newarg=None, proppath=None):
|
|
"""Returns a model and possibly a bios model for the current instance's schema/registry.
|
|
This model can be used to read schema data and validate patches.
|
|
|
|
:param currdict: The dictionary to gather the schema model from.
|
|
:type currdict: dict
|
|
:param attributeregistry: The current systems attribute registry. If not gathering a bios
|
|
registry this can be set to None.
|
|
:type attributeregistry: dict
|
|
:param latestschema: Flag to determine if we should drop the schema version when we try to
|
|
match schema information. If True, the version will be dropped.
|
|
:type latestschema: bool
|
|
:param newargs: List of multi level properties to be gathered.
|
|
:type newargs: list
|
|
:param proppath: The path of the schema you want to validate (from Location header).
|
|
:type proppath: str
|
|
:returns: model and bios model
|
|
"""
|
|
type_str = self.typepath.defs.typestring
|
|
bsmodel = None
|
|
valobj = self.validationmanager
|
|
model = valobj.get_registry_model(
|
|
currtype=currdict[type_str],
|
|
newarg=newarg,
|
|
latestschema=latestschema,
|
|
proppath=proppath,
|
|
)
|
|
if not attributeregistry and model:
|
|
return model, bsmodel
|
|
if not model and not attributeregistry:
|
|
LOGGER.warning("Unable to locate registry/schema for %s\n", currdict[type_str])
|
|
return None, None
|
|
attrval = currdict.get("AttributeRegistry", None)
|
|
attrval = list(attributeregistry.values())[0] if not attrval and attributeregistry else attrval
|
|
bsmodel = valobj.get_registry_model(
|
|
currtype=attrval if attrval else currdict[type_str],
|
|
newarg=newarg,
|
|
latestschema=latestschema,
|
|
searchtype=self.typepath.defs.attributeregtype,
|
|
)
|
|
return model, bsmodel
|
|
|
|
def _build_monolith(self, path=None, includelogs=False, skipbuild=False, json_out=False):
|
|
"""Run through the RIS tree to build monolith
|
|
|
|
:param path: path to initiate login to.
|
|
:type path: str.
|
|
:param includelogs: flag to determine id logs should be downloaded.
|
|
:type includelogs: boolean.
|
|
:param skipbuild: if true, skip build of monolith (initialize empty)
|
|
:type skipbuild: True
|
|
"""
|
|
self.monolith = RisMonolith(self.current_client, self.typepath)
|
|
if not skipbuild:
|
|
self.monolith.load(path=path, includelogs=includelogs, init=True, json_out=json_out)
|
|
else:
|
|
self.monolith.update_member(
|
|
resp=self.current_client.root,
|
|
path=self.current_client.default_prefix,
|
|
init=False,
|
|
)
|
|
|
|
def _modifiedpath(self, results, delete=False, replace=False):
|
|
"""Check the path and set the modified flag
|
|
|
|
:param delete: Flag to delete the path in the results
|
|
:type delete: bool
|
|
:param replace: Flag to replace the path from the results
|
|
:type replace: bool
|
|
:param results: Response for the path
|
|
:type results: RestResponse
|
|
"""
|
|
if not results or results.status not in (200, 201):
|
|
return
|
|
path = results.path
|
|
path = path.split("/Actions")[0] if "Actions" in path else path
|
|
path = path + "/" if self.typepath.defs.isgen10 and path[-1] != "/" else path
|
|
if not replace and path in self.monolith.paths:
|
|
self.monolith.paths[path].modified = True
|
|
_ = self.monolith.markmodified(path)
|
|
if delete and path in self.monolith.paths:
|
|
self.monolith.removepath(path)
|
|
if replace and path in self.monolith.paths:
|
|
self.monolith.paths[path].modified = True
|
|
self.monolith.paths[path].patches = []
|
|
|
|
def _checkforchange(self, paths, crawl=True):
|
|
"""Check if the given paths have been modified and updates monolith if it has
|
|
|
|
:param paths: paths to be checked
|
|
:type paths: list
|
|
"""
|
|
(pathtoetag, _) = self._gettypeswithetag()
|
|
mono = self.monolith
|
|
self.download_path(list(paths), crawl=crawl, path_refresh=True)
|
|
etags = [None if path not in mono.paths else mono.paths[path].etag for path in paths]
|
|
sametag = [
|
|
path
|
|
for ind, path in enumerate(paths)
|
|
if path in pathtoetag and path in self.monolith.paths and pathtoetag[path] != etags[ind]
|
|
]
|
|
for path in sametag:
|
|
self.monolith.paths[path].patches = []
|
|
if sametag:
|
|
LOGGER.warning(
|
|
"The data in the following paths have been updated. " "Recheck the changes made to . %s",
|
|
",".join([str(path) for path in sametag]),
|
|
)
|
|
|
|
def _updatemono(self, currtype=None, path=None, crawl=False, path_refresh=False):
|
|
"""Check if type/path exists in current monolith
|
|
|
|
:param entrytype: the found entry type.
|
|
:type entrytype: str.
|
|
:param currtype: the current entry type.
|
|
:type currtype: str.
|
|
:param crawl: flag to determine if load should traverse found links.
|
|
:type crawl: boolean.
|
|
"""
|
|
monolith = self.monolith
|
|
currtype = None if currtype == '"*"' else currtype
|
|
paths = set()
|
|
if currtype:
|
|
for path, resp in monolith.paths.items():
|
|
if currtype and currtype.lower() not in resp.maj_type.lower():
|
|
continue
|
|
if path_refresh or not resp:
|
|
paths.add(path)
|
|
if resp:
|
|
try:
|
|
if not resp.dict:
|
|
raise AttributeError
|
|
except AttributeError:
|
|
paths.add(path)
|
|
if resp.modified:
|
|
paths.add(path)
|
|
paths.update(monolith.checkmodified(path) if path in monolith.ctree else set())
|
|
elif path:
|
|
if monolith.paths and not list(monolith.paths)[0][-1] == "/":
|
|
path = path[:-1] if path[-1] == "/" else path
|
|
if path_refresh or not monolith.path(path):
|
|
paths.add(path)
|
|
if path in monolith.paths and monolith.paths[path].modified:
|
|
paths.add(path)
|
|
paths.update(monolith.checkmodified(path) if path in monolith.ctree else set())
|
|
if paths:
|
|
self._checkforchange(list(paths), crawl=crawl)
|
|
|
|
def _verifyschemasdownloaded(self, monolith):
|
|
"""Function to verify that the schema has been downloaded
|
|
|
|
:param monolith: full data model retrieved from server.
|
|
:type monolith: dict.
|
|
"""
|
|
|
|
schemaid = self.typepath.schemapath
|
|
regid = self.typepath.regpath
|
|
|
|
if not (schemaid and regid):
|
|
warning_handler("Missing Schemas or registries.")
|
|
return None
|
|
|
|
schemacoll = next(monolith.gettypename(self.typepath.defs.schemafilecollectiontype), None)
|
|
if not schemacoll or any(
|
|
paths.lower() == schemaid and monolith.paths[paths] for paths in monolith.typesadded[schemacoll]
|
|
):
|
|
self.download_path([schemaid], crawl=False)
|
|
schemacoll = next(monolith.gettypename(self.typepath.defs.schemafilecollectiontype), None)
|
|
|
|
regcoll = next(monolith.gettypename(self.typepath.defs.regfilecollectiontype), None)
|
|
if not regcoll or any(
|
|
paths.lower() == regid and monolith.paths[paths] for paths in monolith.typesadded[regcoll]
|
|
):
|
|
self.download_path([regid], crawl=False)
|
|
regcoll = next(monolith.gettypename(self.typepath.defs.regfilecollectiontype), None)
|
|
|
|
return any(
|
|
paths.lower() in (schemaid.lower(), regid.lower()) and monolith.paths[paths] for paths in monolith.paths
|
|
)
|
|
|
|
def _validatechanges(
|
|
self,
|
|
instance=None,
|
|
attributeregistry=None,
|
|
latestschema=None,
|
|
proppath=None,
|
|
newdict=None,
|
|
oridict=None,
|
|
unique=False,
|
|
):
|
|
"""Validate the changes that are requested by the user.
|
|
|
|
:param newdict: dictionary with only the properties that have changed
|
|
:type newdict: dict.
|
|
:param oridict: selection dictionary with current state.
|
|
:type oridict: dict.
|
|
:param unique: flag to determine override for unique properties.
|
|
:type unique: str.
|
|
:param iloversion: current iLO version.
|
|
:type iloversion: float.
|
|
:param instance: current selection instance.
|
|
:type instance: RisMonolithMemberv100.
|
|
:param attrreg: Registry entry of the given attribute.
|
|
:type attrreg: RepoRegistryEntry.
|
|
"""
|
|
entrymono = self.monolith
|
|
currtype = oridict[self.typepath.defs.typestring]
|
|
validation_manager = self.validationmanager
|
|
errors, warnings = validation_manager.validatedict(
|
|
newdict,
|
|
currtype=attributeregistry[instance.maj_type] if attributeregistry else currtype,
|
|
monolith=entrymono,
|
|
unique=unique,
|
|
searchtype=self.typepath.defs.attributeregtype if attributeregistry else None,
|
|
latestschema=latestschema,
|
|
proppath=proppath,
|
|
)
|
|
|
|
validation_errors = errors
|
|
for warninngs in warnings:
|
|
warning_handler(warninngs, override=True)
|
|
if validation_errors and len(validation_errors) > 0:
|
|
raise ValidationError(validation_errors)
|
|
checkallowablevalues(newdict=newdict, oridict=oridict)
|
|
|
|
def _getinstances(self, selector=None, path_refresh=False, crawl=False):
|
|
"""Main function to get instances of particular type and reload
|
|
|
|
:param selector: the type selection for the get operation.
|
|
:type selector: str.
|
|
:param setenable: flag to determine if registry should also be returned.
|
|
:type setenable: boolean.
|
|
:param setenable: flag to determine if registry should also be returned.
|
|
:type setenable: boolean.
|
|
:param path_refresh: flag to reload the selected instances.
|
|
:type path_refresh: boolean.
|
|
:returns: returns a list of selected items
|
|
"""
|
|
instances = list()
|
|
selector = self.selector if not selector else selector
|
|
if selector:
|
|
selector = ".".join(selector.split("#")[-1].split(".")[:2])
|
|
if self.monolith:
|
|
self._updatemono(currtype=selector, crawl=crawl, path_refresh=path_refresh)
|
|
if not selector:
|
|
return instances
|
|
selector = None if selector == '"*"' else selector
|
|
if self.monolith:
|
|
if self.redfishinst.is_redfish:
|
|
instances = [
|
|
inst
|
|
for inst in self.monolith.iter(selector)
|
|
if inst.maj_type not in ["object", "string"] and "redfish" in inst.path
|
|
]
|
|
else:
|
|
instances = [
|
|
inst
|
|
for inst in self.monolith.iter(selector)
|
|
if inst.maj_type not in ["object", "string"] and "rest" in inst.path
|
|
]
|
|
|
|
_ = [setattr(inst, "patches", []) for inst in instances if path_refresh]
|
|
return instances
|
|
|
|
def _checkpostpatch(self, body=None, path=None, patch=False):
|
|
"""Make the post file compatible with the system generation
|
|
|
|
:param body: contents to be checked
|
|
:type body: str.
|
|
:param path: The URL location to check
|
|
:type path: str.
|
|
:param service: flag to determine if minimum calls should be done.
|
|
:type service: boolean.
|
|
:param url: originating url.
|
|
:type url: str.
|
|
:param sessionid: session id to be used instead of iLO credentials.
|
|
:type sessionid: str.
|
|
:param headers: additional headers to be added to the request.
|
|
:type headers: str.
|
|
:param iloresponse: flag to return the iLO response.
|
|
:type iloresponse: str.
|
|
:param silent: flag to determine if no output should be done.
|
|
:type silent: boolean.
|
|
:param patch: flag to determine if a patch is being made
|
|
:type patch: boolean.
|
|
:returns: modified body and path parameter for target and action respectively
|
|
"""
|
|
try:
|
|
if self.typepath.defs.flagforrest:
|
|
if "Target" not in body and not patch:
|
|
if "/Oem/Hp" in path:
|
|
body["Target"] = self.typepath.defs.oempath
|
|
|
|
if path.startswith("/redfish/v1"):
|
|
path = path.replace("/redfish", "/rest", 1)
|
|
|
|
if "/Actions/" in path:
|
|
ind = path.find("/Actions/")
|
|
path = path[:ind]
|
|
|
|
if path.endswith("/"):
|
|
path = path[:-1]
|
|
elif path.startswith("/rest/") and self.typepath.defs.isgen9:
|
|
results = self.get_handler(put_path=path, service=True, silent=True)
|
|
if results and results.status == 200:
|
|
if results.dict:
|
|
if "Target" in body:
|
|
actions = results.dict["Oem"][self.typepath.defs.oemhp]["Actions"]
|
|
elif "Actions" in body:
|
|
actions = results.dict["Actions"]
|
|
else:
|
|
return path, body
|
|
|
|
allkeys = list(actions.keys())
|
|
targetkey = [x for x in allkeys if x.endswith(body["Action"])]
|
|
|
|
if targetkey[0].startswith("#"):
|
|
targetkey[0] = targetkey[0][1:]
|
|
|
|
path = path.replace("/rest", "/redfish", 1)
|
|
path = path + "/Actions"
|
|
|
|
if "Target" in body:
|
|
path = path + self.typepath.defs.oempath
|
|
del body["Target"]
|
|
|
|
if targetkey:
|
|
path = path + "/" + targetkey[0] + "/"
|
|
|
|
return path, body
|
|
except Exception as excp:
|
|
raise excp
|
|
|
|
def _checkforetagchange(self, instance=None):
|
|
"""Function to check the status of the etag
|
|
|
|
:param instance: retrieved instance to check etag for change.
|
|
:type instance: dict.
|
|
"""
|
|
if instance:
|
|
path = instance.path
|
|
(oldtag, _) = self._gettypeswithetag()
|
|
self._updatemono(path=path, path_refresh=True)
|
|
(newtag, _) = self._gettypeswithetag()
|
|
if (oldtag[path] != newtag[path]) and self.typepath.defs.hpilodatetimetype not in instance.maj_type:
|
|
warning_handler(
|
|
"The property you are trying to change "
|
|
"has been updated. Please check entry again "
|
|
" before manipulating it.\n"
|
|
)
|
|
raise ValueChangedError()
|
|
|
|
def _gettypeswithetag(self):
|
|
"""Gathers etags of all paths in monolith and their type associations"""
|
|
instancepath = dict()
|
|
instances = dict()
|
|
|
|
for inst in self.monolith.iter():
|
|
instancepath[inst.path] = inst.maj_type
|
|
instances[inst.path] = inst.etag
|
|
|
|
return [instances, instancepath]
|
|
|
|
# VNIC Functions
|
|
def generate_save_token(self, app_obj):
|
|
return app_obj.generate_and_save_apptoken()
|
|
|
|
def delete_token(self, app_obj):
|
|
return app_obj.remove_apptoken()
|
|
|
|
def token_exists(self, app_obj):
|
|
return app_obj.apptoken_exists()
|
|
|
|
def reactivate_token(self, app_obj):
|
|
return app_obj.reactivate_apptoken()
|
|
|
|
def vnic_login(
|
|
self,
|
|
app_obj=None,
|
|
path=None,
|
|
skipbuild=False,
|
|
includelogs=False,
|
|
json_out=False,
|
|
base_url=None,
|
|
username=None,
|
|
password=None,
|
|
log_dir=None,
|
|
login_otp=None,
|
|
is_redfish=False,
|
|
proxy=None,
|
|
user_ca_cert_data=None,
|
|
biospassword=None,
|
|
sessionid=None,
|
|
):
|
|
# Calling libhpsrv function to get session token
|
|
session_location = ""
|
|
if sessionid:
|
|
session_key = sessionid
|
|
else:
|
|
session_key, session_location = app_obj.vlogin()
|
|
|
|
if not session_key or not session_location:
|
|
raise Exception("Empty Session Id or Session Location was returned.\n")
|
|
|
|
response = self.login(
|
|
username=username,
|
|
password=password,
|
|
sessionid=session_key,
|
|
base_url=base_url,
|
|
path=path,
|
|
skipbuild=skipbuild,
|
|
includelogs=includelogs,
|
|
biospassword=biospassword,
|
|
is_redfish=is_redfish,
|
|
proxy=proxy,
|
|
user_ca_cert_data=user_ca_cert_data,
|
|
json_out=json_out,
|
|
login_otp=login_otp,
|
|
log_dir=log_dir,
|
|
session_location=session_location,
|
|
)
|
|
return response
|
|
|
|
def vexists(self, app_obj):
|
|
# app_obj = Appaccount()
|
|
return app_obj.vnic_exists()
|
|
|
|
def getilover_beforelogin(self, app_obj):
|
|
# app_obj = Appaccount(appname=appname, appid=appid, username=username, password=password)
|
|
ilo_ver, securitystate = app_obj.version_beforelogin()
|
|
return ilo_ver, securitystate
|
|
|
|
def GetIPAddress(self):
|
|
app_obj = AppAccount()
|
|
IPAddr = app_obj.GetIPAddress()
|
|
if "https://" not in IPAddr:
|
|
IPAddr = "https://" + IPAddr
|
|
return IPAddr
|
|
|
|
def ListAppIds(self, app_obj):
|
|
appid_list_info = app_obj.CompareAppIds()
|
|
return appid_list_info
|
|
|
|
def ExpandAppId(self, app_obj, appid):
|
|
expand_app_id = app_obj.ExpandAppId(appid)
|
|
return expand_app_id
|