# -*- coding: utf-8 -*-
###
# (C) Copyright [2021] Hewlett Packard Enterprise Development LP
#
# 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.
###
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
from future import standard_library
standard_library.install_aliases()
from hpeOneView.exceptions import HPEOneViewResourceNotFound
from hpeOneView.resources.resource import (Resource, ResourcePatchMixin, merge_resources,
ensure_resource_client, unavailable_method)
[docs]class LogicalInterconnects(ResourcePatchMixin, Resource):
"""
Logical Interconnects API client.
"""
URI = '/rest/logical-interconnects'
FIRMWARE_PATH = "/firmware"
SNMP_CONFIGURATION_PATH = "/snmp-configuration"
PORT_MONITOR_PATH = "/port-monitor"
FORWARDING_INFORMATION_PATH = "/forwarding-information-base"
QOS_AGGREGATED_CONFIGURATION = "/qos-aggregated-configuration"
SETTINGS_DEFAULT_VALUES = {
'200': {"type": "InterconnectSettingsV3"},
'300': {"type": "InterconnectSettingsV201"},
'500': {"type": "InterconnectSettingsV201"},
}
SETTINGS_ETHERNET_DEFAULT_VALUES = {
'200': {"type": "EthernetInterconnectSettingsV3"},
'300': {"type": "EthernetInterconnectSettingsV201"},
'500': {"type": "EthernetInterconnectSettingsV201"},
'600': {"type": "EthernetInterconnectSettingsV4"},
'800': {"type": "EthernetInterconnectSettingsV4"},
'1000': {'type': "EthernetInterconnectSettingsV5"},
'1200': {'type': "EthernetInterconnectSettingsV6"},
'1400': {'type': "EthernetInterconnectSettingsV7"},
'1600': {'type': "EthernetInterconnectSettingsV7"},
'1800': {'type': "EthernetInterconnectSettingsV7"}
}
SETTINGS_TELEMETRY_CONFIG_DEFAULT_VALUES = {
'200': {"type": "telemetry-configuration"},
'300': {"type": "telemetry-configuration"},
'500': {"type": "telemetry-configuration"},
'600': {"type": "telemetry-configuration"},
'800': {"type": "telemetry-configuration"},
'1000': {"type": "telemetry-configuration"},
'1200': {"type": "telemetry-configuration"},
'1400': {"type": "telemetry-configuration"},
'1600': {"type": "telemetry-configuration"},
'1800': {"type": "telemetry-configuration"}
}
def __init__(self, connection, data=None):
super(LogicalInterconnects, self).__init__(connection, data)
[docs] def create(self):
"""Create method is not available for this resource"""
unavailable_method()
[docs] def update(self):
"""Update method is not available for this resource"""
unavailable_method()
[docs] def delete(self):
"""Delete method is not available for this resource"""
unavailable_method()
[docs] def get_all(self, start=0, count=-1, sort=''):
"""
Gets a list of logical interconnects based on optional sorting and filtering and is constrained by start
and count parameters.
Args:
start:
The first item to return, using 0-based indexing.
If not specified, the default is 0 - start with the first available item.
count:
The number of resources to return. A count of -1 requests all items.
The actual number of items in the response might differ from the requested
count if the sum of start and count exceeds the total number of items.
sort:
The sort order of the returned data set. By default, the sort order is based
on create time with the oldest entry first.
Returns:
list: A list of logical interconnects.
"""
return self._helper.get_all(start, count, sort=sort)
[docs] def get_by_name(self, name):
"""
Gets a logical interconnect by name.
Args:
name: Name of the logical interconnect.
Returns:
dict: Logical Interconnect.
"""
logical_interconnects = self.get_all()
result = [x for x in logical_interconnects if x['name'] == name]
resource = result[0] if result else None
if resource:
resource = self.new(self._connection, resource)
return resource
@ensure_resource_client
def update_compliance(self, timeout=-1):
"""
Returns logical interconnects to a consistent state. The current logical interconnect state is
compared to the associated logical interconnect group.
Any differences identified are corrected, bringing the logical interconnect back to a consistent
state. Changes are asynchronously applied to all managed interconnects. Note that if the changes detected
involve differences in the interconnect map between the logical interconnect group and the logical interconnect,
the process of bringing the logical interconnect back to a consistent state might involve automatically removing
existing interconnects from management and/or adding new interconnects for management.
Args:
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}/compliance".format(self.data["uri"])
return self._helper.update(None, uri, timeout=timeout)
@ensure_resource_client
def update_ethernet_settings(self, configuration, force=False, timeout=-1):
"""
Updates the Ethernet interconnect settings for the logical interconnect.
Args:
configuration: Ethernet interconnect settings.
force: If set to true, the operation completes despite any problems with network connectivity or errors
on the resource itself. The default is false.
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}/ethernetSettings".format(self.data["uri"])
return self._helper.update(configuration, uri=uri, force=force, timeout=timeout)
@ensure_resource_client
def update_internal_networks(self, network_uri_list, force=False, timeout=-1):
"""
Updates internal networks on the logical interconnect.
Args:
network_uri_list: List of Ethernet network uris.
force: If set to true, the operation completes despite any problems with network connectivity or errors
on the resource itself. The default is false.
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}/internalNetworks".format(self.data["uri"])
return self._helper.update(network_uri_list, uri=uri, force=force, timeout=timeout)
@ensure_resource_client
def get_internal_vlans(self):
"""
Gets the internal VLAN IDs for the provisioned networks on a logical interconnect.
Returns:
dict: Collection of URIs
"""
uri = "{}/internalVlans".format(self.data["uri"])
response = self._helper.do_get(uri)
return self._helper.get_members(response)
@ensure_resource_client
def update_settings(self, settings, force=False, timeout=-1):
"""
Updates interconnect settings on the logical interconnect. Changes to interconnect settings are asynchronously
applied to all managed interconnects.
(This method is not available from API version 600 onwards)
Args:
settings: Interconnect settings
force: If set to true, the operation completes despite any problems with network connectivity or errors
on the resource itself. The default is false.
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect
"""
data = settings.copy()
if 'ethernetSettings' in data:
ethernet_default_values = self._get_default_values(self.SETTINGS_ETHERNET_DEFAULT_VALUES)
data['ethernetSettings'] = merge_resources(data['ethernetSettings'],
ethernet_default_values)
uri = "{}/settings".format(self.data["uri"])
default_values = self._get_default_values(self.SETTINGS_DEFAULT_VALUES)
data = self._helper.update_resource_fields(data, default_values)
return self._helper.update(data, uri=uri, force=force, timeout=timeout)
@ensure_resource_client
def update_configuration(self, timeout=-1):
"""
Asynchronously applies or re-applies the logical interconnect configuration to all managed interconnects.
Args:
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}/configuration".format(self.data["uri"])
return self._helper.update(None, uri=uri, timeout=timeout)
@ensure_resource_client
def get_snmp_configuration(self):
"""
Gets the SNMP configuration for a logical interconnect.
Returns:
dict: SNMP configuration.
"""
uri = "{}{}".format(self.data["uri"], self.SNMP_CONFIGURATION_PATH)
return self._helper.do_get(uri)
@ensure_resource_client
def update_snmp_configuration(self, configuration, timeout=-1):
"""
Updates the SNMP configuration of a logical interconnect. Changes to the SNMP configuration are asynchronously
applied to all managed interconnects.
Args:
configuration: snmp configuration.
Returns:
dict: The Logical Interconnect.
"""
data = configuration.copy()
if 'type' not in data:
data['type'] = 'snmp-configuration'
uri = "{}{}".format(self.data["uri"], self.SNMP_CONFIGURATION_PATH)
return self._helper.update(data, uri=uri, timeout=timeout)
@ensure_resource_client
def get_unassigned_ports(self):
"""
Gets the collection ports from the member interconnects
which are eligible for assignment to an anlyzer port
Returns:
dict: Collection of ports
"""
uri = "{}/unassignedPortsForPortMonitor".format(self.data["uri"])
response = self._helper.do_get(uri)
return self._helper.get_members(response)
@ensure_resource_client
def get_unassigned_uplink_ports(self):
"""
Gets a collection of uplink ports from the member interconnects which are eligible for assignment to an
analyzer port. To be eligible, a port must be a valid uplink, must not be a member of an existing uplink set,
and must not currently be used for stacking.
Returns:
dict: Collection of uplink ports.
"""
uri = "{}/unassignedUplinkPortsForPortMonitor".format(self.data["uri"])
response = self._helper.do_get(uri)
return self._helper.get_members(response)
@ensure_resource_client
def get_port_monitor(self):
"""
Gets the port monitor configuration of a logical interconnect.
Returns:
dict: The Logical Interconnect.
"""
uri = "{}{}".format(self.data["uri"], self.PORT_MONITOR_PATH)
return self._helper.do_get(uri)
@ensure_resource_client
def update_port_monitor(self, resource, timeout=-1):
"""
Updates the port monitor configuration of a logical interconnect.
Args:
resource: Port monitor configuration.
Returns:
dict: Port monitor configuration.
"""
data = resource.copy()
if 'type' not in data:
data['type'] = 'port-monitor'
uri = "{}{}".format(self.data["uri"], self.PORT_MONITOR_PATH)
return self._helper.update(data, uri=uri, timeout=timeout)
@ensure_resource_client
def get_firmware(self):
"""
Gets the installed firmware for a logical interconnect.
Returns:
dict: LIFirmware.
"""
firmware_uri = self._helper.build_subresource_uri(self.data["uri"], subresource_path=self.FIRMWARE_PATH)
return self._helper.do_get(firmware_uri)
@ensure_resource_client
def install_firmware(self, firmware_information):
"""
Installs firmware to a logical interconnect. The three operations that are supported for the firmware
update are Stage (uploads firmware to the interconnect), Activate (installs firmware on the interconnect),
and Update (which does a Stage and Activate in a sequential manner).
Args:
firmware_information: Options to install firmware to a logical interconnect.
Returns:
dict
"""
firmware_uri = self._helper.build_subresource_uri(self.data["uri"], subresource_path=self.FIRMWARE_PATH)
return self._helper.update(firmware_information, firmware_uri)
@ensure_resource_client
def get_forwarding_information_base(self, filter=''):
"""
Gets the forwarding information base data for a logical interconnect. A maximum of 100 entries is returned.
Optional filtering criteria might be specified.
Args:
filter (list or str):
Filtering criteria may be specified using supported attributes: interconnectUri, macAddress,
internalVlan, externalVlan, and supported relation = (Equals). macAddress is 12 hexadecimal digits with
a colon between each pair of digits (upper case or lower case).
The default is no filter; all resources are returned.
Returns:
list: A set of interconnect MAC address entries.
"""
uri = "{}{}".format(self.data["uri"], self.FORWARDING_INFORMATION_PATH)
return self._helper.get_collection(uri, filter=filter)
@ensure_resource_client
def create_forwarding_information_base(self, timeout=-1):
"""
Generates the forwarding information base dump file for a logical interconnect.
Args:
timeout:
Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation in
OneView, just stops waiting for its completion.
Returns: Interconnect Forwarding Information Base DataInfo.
"""
uri = "{}{}".format(self.data["uri"], self.FORWARDING_INFORMATION_PATH)
return self._helper.do_post(uri, None, timeout, None)
@ensure_resource_client
def get_qos_aggregated_configuration(self):
"""
Gets the QoS aggregated configuration for the logical interconnect.
Returns:
dict: QoS Configuration.
"""
uri = "{}{}".format(self.data["uri"], self.QOS_AGGREGATED_CONFIGURATION)
return self._helper.do_get(uri)
@ensure_resource_client
def update_qos_aggregated_configuration(self, qos_configuration, timeout=-1):
"""
Updates the QoS aggregated configuration for the logical interconnect.
Args:
qos_configuration:
QOS configuration.
timeout:
Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation in
OneView, just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}{}".format(self.data["uri"], self.QOS_AGGREGATED_CONFIGURATION)
return self._helper.update(qos_configuration, uri=uri, timeout=timeout)
def _get_telemetry_configuration_uri(self):
telemetry_conf = self.data.get("telemetryConfiguration", {})
if not telemetry_conf.get("uri"):
raise HPEOneViewResourceNotFound("Telemetry configuration uri is not available")
return telemetry_conf["uri"]
@ensure_resource_client
def get_telemetry_configuration(self):
"""
Gets the telemetry configuration of a logical interconnect.
Returns:
dict: Telemetry configuration.
"""
telemetry_conf_uri = self._get_telemetry_configuration_uri()
return self._helper.do_get(telemetry_conf_uri)
@ensure_resource_client
def update_telemetry_configurations(self, configuration, timeout=-1):
"""
Updates the telemetry configuration of a logical interconnect. Changes to the telemetry configuration are
asynchronously applied to all managed interconnects.
Args:
configuration:
The telemetry configuration for the logical interconnect.
timeout:
Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation in
OneView, just stops waiting for its completion.
Returns:
dict: The Logical Interconnect.
"""
telemetry_conf_uri = self._get_telemetry_configuration_uri()
default_values = self._get_default_values(self.SETTINGS_TELEMETRY_CONFIG_DEFAULT_VALUES)
configuration = self._helper.update_resource_fields(configuration, default_values)
return self._helper.update(configuration, uri=telemetry_conf_uri, timeout=timeout)
@ensure_resource_client
def get_ethernet_settings(self):
"""
Gets the Ethernet interconnect settings for the Logical Interconnect.
Returns:
dict: Ethernet Interconnect Settings
"""
uri = "{}/ethernetSettings".format(self.data["uri"])
return self._helper.do_get(uri)
@ensure_resource_client
def get_igmp_settings(self):
"""
Gets the Igmp settings for the Logical Interconnect.
Returns:
dict: Ethernet Igmp Settings
"""
uri = "{}/igmpSettings".format(self.data["uri"])
return self._helper.do_get(uri)
@ensure_resource_client
def update_igmp_settings(self, configuration, force=False, timeout=-1):
"""
Updates the Igmp interconnect settings for the logical interconnect.
Args:
configuration: Igmp interconnect settings.
force: If set to true, the operation completes despite any problems with network connectivity or errors
on the resource itself. The default is false.
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}/igmpSettings".format(self.data["uri"])
return self._helper.update(configuration, uri=uri, force=force, timeout=timeout)
@ensure_resource_client
def bulk_inconsistency_validate(self, validation_request, timeout=-1):
"""
Validates the bulk update from group operation and gets the consolidated inconsistency report
Args:
timeout:
Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation in
OneView, just stops waiting for its completion.
Returns: Consolidated inconsistency report.
"""
uri = "{}/bulk-inconsistency-validation".format(self.URI)
return self._helper.do_post(uri, validation_request, timeout, None)
@ensure_resource_client
def update_port_flap_settings(self, configuration, force=False, timeout=-1):
"""
Updates the port flap interconnect settings for the logical interconnect.
Args:
configuration: Port flap settings.
force: If set to true, the operation completes despite any problems with network connectivity or errors
on the resource itself. The default is false.
timeout: Timeout in seconds. Wait for task completion by default. The timeout does not abort the operation
in OneView; it just stops waiting for its completion.
Returns:
dict: Logical Interconnect.
"""
uri = "{}/portFlapSettings".format(self.data["uri"])
return self._helper.update(configuration, uri=uri, force=force, timeout=timeout)