Add all missing attributes of EthernetSwitch in RSD 2.1

Change-Id: Id6c6a5987f2f842e68ffb8019000831df2c035fa
This commit is contained in:
Lin Yang 2019-03-22 11:20:52 -07:00
parent c53072527b
commit 4acbd7a395
34 changed files with 1931 additions and 1781 deletions

View File

@ -0,0 +1,48 @@
# Copyright 2019 Intel, 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.
from sushy.resources import base
from rsd_lib import base as rsd_lib_base
class IPv6AddressCollectionField(rsd_lib_base.ReferenceableMemberField):
address = base.Field("Address")
"""This is the IPv6 Address."""
prefix_length = base.Field("PrefixLength")
"""This is the IPv6 Address Prefix Length."""
address_origin = base.Field("AddressOrigin")
"""This indicates how the address was determined."""
address_state = base.Field("AddressState")
"""The current state of this address as defined in RFC 4862."""
class IPv4AddressCollectionField(rsd_lib_base.ReferenceableMemberField):
address = base.Field("Address")
"""This is the IPv4 Address."""
subnet_mask = base.Field("SubnetMask")
"""This is the IPv4 Subnet mask."""
address_origin = base.Field("AddressOrigin")
"""This indicates how the address was determined."""
gateway = base.Field("Gateway")
"""This is the IPv4 gateway for this address."""

View File

@ -1,4 +1,4 @@
# Copyright 2018 Intel, Inc.
# Copyright 2019 Intel, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

View File

@ -1,4 +1,4 @@
# Copyright 2018 Intel, Inc.
# Copyright 2019 Intel, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may

View File

@ -1,85 +0,0 @@
# Copyright 2018 99cloud, 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.
from sushy.resources import base
from sushy import utils
from rsd_lib.resources.v2_1.ethernet_switch import acl_rule
class LinksField(base.CompositeField):
bound_ports = base.Field('BoundPorts',
adapter=utils.get_members_identities)
"""return a EthernetSwitchPort collection"""
class ACL(base.ResourceBase):
identity = base.Field('Id', required=True)
"""The acl identity string"""
name = base.Field('Name')
"""The acl name"""
description = base.Field('Description')
"""The acl description"""
oem = base.Field('Oem')
"""The acl oem info"""
links = LinksField('Links')
"""The acl links"""
def __init__(self, connector, identity, redfish_version=None):
"""A class representing an ACL
:param connector: A Connector instance
:param identity: The identity of the ACL resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(ACL, self).__init__(connector, identity, redfish_version)
def _get_acl_rule_collection_path(self):
"""Helper function to find the RuleCollection path"""
return utils.get_sub_resource_path_by(self, 'Rules')
@property
@utils.cache_it
def rules(self):
"""Property to provide reference to `RuleCollection` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return acl_rule.ACLRuleCollection(
self._conn, self._get_acl_rule_collection_path(),
redfish_version=self.redfish_version)
class ACLCollection(base.ResourceCollectionBase):
@property
def _resource_type(self):
return ACL
def __init__(self, connector, path, redfish_version=None):
"""A class representing an ACL
:param connector: A Connector instance
:param path: The canonical path to the ACL collection resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(ACLCollection, self).__init__(connector, path, redfish_version)

View File

@ -1,136 +0,0 @@
# Copyright 2018 99cloud, 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.
import logging
from jsonschema import validate
from sushy.resources import base
from sushy import utils
from rsd_lib.resources.v2_1.ethernet_switch import schemas as acl_rule_schema
from rsd_lib import utils as rsd_lib_utils
LOG = logging.getLogger(__name__)
class IPConditionTypeField(base.CompositeField):
ipv4_address = base.Field('IPv4Address')
mask = base.Field('Mask')
class MACConditionTypeField(base.CompositeField):
mac_address = base.Field('Address')
mask = base.Field('Mask')
class VlanIdConditionTypeField(base.CompositeField):
id = base.Field('Id', adapter=rsd_lib_utils.num_or_none)
mask = base.Field('Mask', adapter=rsd_lib_utils.num_or_none)
class PortConditionTypeField(base.CompositeField):
port = base.Field('Port', adapter=rsd_lib_utils.num_or_none)
mask = base.Field('Mask', adapter=rsd_lib_utils.num_or_none)
class ConditionTypeField(base.CompositeField):
ip_source = IPConditionTypeField('IPSource')
ip_destination = IPConditionTypeField('IPDestination')
mac_source = MACConditionTypeField('MACSource')
mac_destination = MACConditionTypeField('MACDestination')
vlan_id = VlanIdConditionTypeField('VLANId')
l4_source_port = PortConditionTypeField('L4SourcePort')
l4_destination_port = PortConditionTypeField('L4DestinationPort')
l4_protocol = base.Field('L4Protocol', adapter=rsd_lib_utils.num_or_none)
class ACLRule(base.ResourceBase):
identity = base.Field('Id')
"""The acl rule identity string"""
name = base.Field('Name')
"""The acl rule name"""
description = base.Field('Description')
"""The acl rule description"""
rule_id = base.Field('RuleId', adapter=rsd_lib_utils.num_or_none)
"""The acl rule id"""
action = base.Field('Action')
"""The acl rule action"""
forward_mirror_interface = base.Field(
'ForwardMirrorInterface',
adapter=rsd_lib_utils.get_resource_identity)
"""The acl rule forward mirror interface"""
mirror_port_region = base.Field('MirrorPortRegion',
adapter=utils.get_members_identities)
"""The acl rule mirror port region"""
mirror_type = base.Field('MirrorType')
"""The acl rule mirror type"""
condition = ConditionTypeField('Condition')
"""The acl rule condition field"""
oem = base.Field('Oem')
"""The ac rule oem field"""
links = base.Field('Links')
"""The acl rule links field"""
def __init__(self, connector, identity, redfish_version=None):
"""A class representing an ACL Rule
:param connector: A connector instance
:param identity: The identity of the ACL Rule resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(ACLRule, self).__init__(connector, identity, redfish_version)
class ACLRuleCollection(base.ResourceCollectionBase):
@property
def _resource_type(self):
return ACLRule
def __init__(self, connector, path, redfish_version=None):
"""A class representing an ACL Rule Collection
:param connector: A Connector instance
:param path: The canonical path to the ACL Rule collection resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(ACLRuleCollection, self).__init__(connector,
path,
redfish_version)
def add_acl_rule(self, acl_rule_req):
"""Add a acl rule
:param acl_rule: JSON for acl_rule
:returns: The location of the acl rule
"""
target_uri = self._path
validate(acl_rule_req, acl_rule_schema.acl_rule_req_schema)
resp = self._conn.post(target_uri, data=acl_rule_req)
acl_rule_url = resp.headers['Location']
LOG.info("ACL Rule add at %s", acl_rule_url)
return acl_rule_url[acl_rule_url.find(self._path):]

View File

@ -18,130 +18,101 @@ import logging
from sushy.resources import base
from sushy import utils
from rsd_lib import common as rsd_lib_common
from rsd_lib.resources.v2_1.ethernet_switch import acl
from rsd_lib.resources.v2_1.ethernet_switch import port
from rsd_lib import base as rsd_lib_base
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_acl
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_port
from rsd_lib import utils as rsd_lib_utils
LOG = logging.getLogger(__name__)
class LinksField(base.CompositeField):
chassis = base.Field('Chassis', default=(),
adapter=rsd_lib_utils.get_resource_identity)
chassis = base.Field(
"Chassis", default=(), adapter=rsd_lib_utils.get_resource_identity
)
"""Link to chassis of this ethernet switch"""
managed_by = base.Field('ManagedBy', default=(),
adapter=utils.get_members_identities)
managed_by = base.Field(
"ManagedBy", default=(), adapter=utils.get_members_identities
)
"""Link to manager of this ethernet switch"""
class EthernetSwitch(base.ResourceBase):
identity = base.Field('Id', required=True)
"""The ethernet switch identity string"""
class EthernetSwitch(rsd_lib_base.ResourceBase):
switch_id = base.Field('SwitchId')
"""The ethernet switch id"""
switch_id = base.Field("SwitchId")
"""Unique switch Id (within drawer) used to identify in switch hierarchy
discovery.
"""
name = base.Field('Name')
"""The ethernet switch name"""
manufacturer = base.Field("Manufacturer")
"""Switch manufacturer name"""
description = base.Field('Description')
"""The ethernet switch description"""
model = base.Field("Model")
"""Switch model"""
manufacturer = base.Field('Manufacturer')
"""The ethernet switch manufacturer"""
manufacturing_date = base.Field("ManufacturingDate")
"""Manufacturing date"""
model = base.Field('Model')
"""The ethernet switch model"""
serial_number = base.Field("SerialNumber")
"""Switch serial numberSS"""
manufacturing_date = base.Field('ManufacturingDate')
"""The ethernet switch manufacturing date"""
part_number = base.Field("PartNumber")
"""Switch part number"""
serial_number = base.Field('SerialNumber')
"""The ethernet switch serial number"""
firmware_name = base.Field("FirmwareName")
"""Switch firmware name"""
part_number = base.Field('PartNumber')
"""The ethernet switch port number"""
firmware_version = base.Field("FirmwareVersion")
"""Switch firmware version"""
firmware_name = base.Field('FirmwareName')
"""The ethernet switch fireware name"""
role = base.Field("Role")
""""""
firmware_version = base.Field('FirmwareVersion')
"""The ethernet switch firmware version"""
max_acl_number = base.Field(
"MaxACLNumber", adapter=rsd_lib_utils.num_or_none
)
"""Role of switch"""
role = base.Field('Role')
"""The ethernet switch role"""
status = rsd_lib_base.StatusField("Status")
"""This indicates the known state of the resource, such as if it is
enabled.
"""
max_acl_number = base.Field('MaxACLNumber')
"""The ethernet switch max acl number"""
status = rsd_lib_common.StatusField('Status')
"""The ethernet switch status"""
links = LinksField('Links')
"""The links to ethernet switch"""
def __init__(self, conncetor, identity, redfish_version=None):
"""A class representing a EthernetSwitch
:param connector: A Connector instance
:param identity: The identity of the EthernetSwitch resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(EthernetSwitch, self).__init__(conncetor,
identity,
redfish_version)
def _get_port_collection_path(self):
"""Helper function to find the PortCollection path"""
return utils.get_sub_resource_path_by(self, 'Ports')
links = LinksField("Links")
@property
@utils.cache_it
def ports(self):
"""Property to provide reference to `PortCollection` instance
"""Property to provide reference to `EthernetSwitchPortCollection` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return port.PortCollection(
self._conn, self._get_port_collection_path(),
redfish_version=self.redfish_version)
def _get_acl_collection_path(self):
"""Helper function to find the ACLCollection path"""
return utils.get_sub_resource_path_by(self, 'ACLs')
return ethernet_switch_port.EthernetSwitchPortCollection(
self._conn,
utils.get_sub_resource_path_by(self, "Ports"),
redfish_version=self.redfish_version,
)
@property
@utils.cache_it
def acls(self):
"""Property to provide reference to `ACLCollection` instance
"""Property to provide reference to `EthernetSwitchACLCollection` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return acl.ACLCollection(
self._conn, self._get_acl_collection_path(),
redfish_version=self.redfish_version)
return ethernet_switch_acl.EthernetSwitchACLCollection(
self._conn,
utils.get_sub_resource_path_by(self, "ACLs"),
redfish_version=self.redfish_version,
)
class EthernetSwitchCollection(base.ResourceCollectionBase):
class EthernetSwitchCollection(rsd_lib_base.ResourceCollectionBase):
@property
def _resource_type(self):
return EthernetSwitch
def __init__(self, connector, path, redfish_version=None):
"""A class representing a EthernetSwitch Collection
:param connector: A Connector instance
:param path: The canonical path to the EthernetSwitch collection
resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(EthernetSwitchCollection, self).__init__(connector,
path,
redfish_version)

View File

@ -0,0 +1,58 @@
# Copyright 2018 99cloud, 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.
from sushy.resources import base
from sushy import utils
from rsd_lib import base as rsd_lib_base
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_acl_rule
class LinksField(base.CompositeField):
bound_ports = base.Field(
"BoundPorts", adapter=utils.get_members_identities
)
class EthernetSwitchACL(rsd_lib_base.ResourceBase):
"""EthernetSwitchACL resource class
A Ethernet Switch ACL represents Access Control List for switch.
"""
links = LinksField("Links")
# TODO(linyang): Add Action Field
@property
@utils.cache_it
def rules(self):
"""Property to provide reference to `EthernetSwitchACLRuleCollection` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return ethernet_switch_acl_rule.EthernetSwitchACLRuleCollection(
self._conn,
utils.get_sub_resource_path_by(self, "Rules"),
redfish_version=self.redfish_version,
)
class EthernetSwitchACLCollection(rsd_lib_base.ResourceCollectionBase):
@property
def _resource_type(self):
return EthernetSwitchACL

View File

@ -0,0 +1,146 @@
# Copyright 2018 99cloud, 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.
import logging
from jsonschema import validate
from sushy.resources import base
from sushy import utils
from rsd_lib import base as rsd_lib_base
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_port
from rsd_lib.resources.v2_1.ethernet_switch import schemas as acl_rule_schema
from rsd_lib import utils as rsd_lib_utils
LOG = logging.getLogger(__name__)
class PortConditionTypeField(base.CompositeField):
port = base.Field("Port", adapter=rsd_lib_utils.num_or_none)
mask = base.Field("Mask", adapter=rsd_lib_utils.num_or_none)
class VlanIdConditionTypeField(base.CompositeField):
identity = base.Field("Id", adapter=rsd_lib_utils.num_or_none)
mask = base.Field("Mask", adapter=rsd_lib_utils.num_or_none)
class MACConditionTypeField(base.CompositeField):
mac_address = base.Field("MACAddress")
mask = base.Field("Mask")
class IPConditionTypeField(base.CompositeField):
ipv4_address = base.Field("IPv4Address")
mask = base.Field("Mask")
class ConditionTypeField(base.CompositeField):
ip_source = IPConditionTypeField("IPSource")
ip_destination = IPConditionTypeField("IPDestination")
mac_source = MACConditionTypeField("MACSource")
mac_destination = MACConditionTypeField("MACDestination")
vlan_id = VlanIdConditionTypeField("VLANId")
l4_source_port = PortConditionTypeField("L4SourcePort")
l4_destination_port = PortConditionTypeField("L4DestinationPort")
l4_protocol = base.Field("L4Protocol", adapter=rsd_lib_utils.num_or_none)
class EthernetSwitchACLRule(rsd_lib_base.ResourceBase):
"""EthernetSwitchACLRule resource class
A Ethernet Switch ACL Rule represents Access Control List rule for
switch.
"""
rule_id = base.Field("RuleId", adapter=rsd_lib_utils.num_or_none)
"""This is ACL rule ID which determine rule priority."""
action = base.Field("Action")
"""Action that will be executed when rule condition will be met.s"""
mirror_type = base.Field("MirrorType")
"""Type of mirroring that should be use for Mirror action."""
condition = ConditionTypeField("Condition")
"""Property contain set of conditions that should be met to trigger Rule
action.
"""
@property
@utils.cache_it
def forward_mirror_interface(self):
"""Property to provide reference to `EthernetSwitchPort` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return ethernet_switch_port.EthernetSwitchPort(
self._conn,
utils.get_sub_resource_path_by(self, "ForwardMirrorInterface"),
redfish_version=self.redfish_version,
)
@property
@utils.cache_it
def mirror_port_region(self):
"""Property to provide a list of `EthernetSwitchPort` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return [
ethernet_switch_port.EthernetSwitchPort(
self._conn, path, redfish_version=self.redfish_version
)
for path in rsd_lib_utils.get_sub_resource_path_list_by(
self, "MirrorPortRegion"
)
]
class EthernetSwitchACLRuleCollection(rsd_lib_base.ResourceCollectionBase):
@property
def _resource_type(self):
return EthernetSwitchACLRule
def add_acl_rule(self, acl_rule_req):
"""Add a acl rule
:param acl_rule: JSON for acl_rule
:returns: The location of the acl rule
"""
target_uri = self._path
validate(acl_rule_req, acl_rule_schema.acl_rule_req_schema)
resp = self._conn.post(target_uri, data=acl_rule_req)
acl_rule_url = resp.headers["Location"]
LOG.info("ACL Rule add at %s", acl_rule_url)
return acl_rule_url[acl_rule_url.find(self._path):]

View File

@ -0,0 +1,175 @@
# Copyright 2018 99cloud, 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.
import logging
from sushy.resources import base
from sushy import utils
from rsd_lib import base as rsd_lib_base
from rsd_lib.resources.v2_1.common import ip_addresses
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_static_mac
from rsd_lib.resources.v2_1.ethernet_switch import vlan_network_interface
from rsd_lib import utils as rsd_lib_utils
LOG = logging.getLogger(__name__)
class NeighborInfoField(base.CompositeField):
switch_id = base.Field("SwitchId")
port_id = base.Field("PortId")
cable_id = base.Field("CableId")
class LinksIntelRackScaleField(base.CompositeField):
neighbor_interface = base.Field(
"NeighborInterface", adapter=rsd_lib_utils.get_resource_identity
)
class LinksOemField(base.CompositeField):
intel_rackscale = LinksIntelRackScaleField("Intel_RackScale")
"""Intel Rack Scale Design specific properties."""
class LinksField(base.CompositeField):
primary_vlan = base.Field(
"PrimaryVLAN", adapter=rsd_lib_utils.get_resource_identity
)
switch = base.Field("Switch", adapter=rsd_lib_utils.get_resource_identity)
member_of_port = base.Field(
"MemberOfPort", adapter=rsd_lib_utils.get_resource_identity
)
port_members = base.Field(
"PortMembers", adapter=utils.get_members_identities
)
active_acls = base.Field(
"ActiveACLs", adapter=utils.get_members_identities
)
oem = LinksOemField("Oem")
"""Oem specific properties."""
class EthernetSwitchPort(rsd_lib_base.ResourceBase):
port_id = base.Field("PortId")
"""Switch port unique identifier."""
link_type = base.Field("LinkType")
"""Type of port link"""
operational_state = base.Field("OperationalState")
"""Port link operational state"""
administrative_state = base.Field("AdministrativeState")
"""Port link state forced by user."""
link_speed_mbps = base.Field(
"LinkSpeedMbps", adapter=rsd_lib_utils.num_or_none
)
"""Port speed"""
neighbor_info = NeighborInfoField("NeighborInfo")
"""For Upstream port type this property provide information about neighbor
switch (and switch port if available) connected to this port
"""
neighbor_mac = base.Field("NeighborMAC")
"""For Downstream port type this property provide MAC address of NIC
connected to this port.
"""
frame_size = base.Field("FrameSize", adapter=rsd_lib_utils.num_or_none)
"""MAC frame size in bytes"""
autosense = base.Field("Autosense", adapter=bool)
"""Indicates if the speed and duplex is automatically configured by the NIC
"""
full_duplex = base.Field("FullDuplex", adapter=bool)
"""Indicates if port is in Full Duplex mode or not"""
mac_address = base.Field("MACAddress")
"""MAC address of port."""
ipv4_addresses = ip_addresses.IPv4AddressCollectionField("IPv4Addresses")
"""Array of following IPv4 address"""
ipv6_addresses = ip_addresses.IPv6AddressCollectionField("IPv6Addresses")
"""Array of following IPv6 address"""
port_class = base.Field("PortClass")
"""Port class"""
port_mode = base.Field("PortMode")
"""Port working mode. The value shall correspond to the port class
(especially to the logical port definition).
"""
port_type = base.Field("PortType")
"""PortType"""
status = rsd_lib_base.StatusField("Status")
"""This indicates the known state of the resource, such as if it is
enabled.
"""
links = LinksField("Links")
@property
@utils.cache_it
def vlans(self):
"""Property to provide reference to `VLanNetworkInterfaceCollection`
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return vlan_network_interface.VLanNetworkInterfaceCollection(
self._conn,
utils.get_sub_resource_path_by(self, "VLANs"),
redfish_version=self.redfish_version,
)
@property
@utils.cache_it
def static_macs(self):
"""Property to provide reference to `EthernetSwitchStaticMACCollection`
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return ethernet_switch_static_mac.EthernetSwitchStaticMACCollection(
self._conn,
utils.get_sub_resource_path_by(self, "StaticMACs"),
redfish_version=self.redfish_version,
)
class EthernetSwitchPortCollection(rsd_lib_base.ResourceCollectionBase):
@property
def _resource_type(self):
return EthernetSwitchPort

View File

@ -0,0 +1,38 @@
# Copyright 2018 99cloud, 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.
from sushy.resources import base
from rsd_lib import base as rsd_lib_base
from rsd_lib import utils as rsd_lib_utils
class EthernetSwitchStaticMAC(rsd_lib_base.ResourceBase):
"""EthernetSwitchStaticMAC resource class
A Ethernet Switch ACL represents Access Control List for switch.
"""
vlan_id = base.Field("VLANId", adapter=rsd_lib_utils.num_or_none)
"""The static mac vlan id"""
mac_address = base.Field("MACAddress")
"""The static mac address"""
class EthernetSwitchStaticMACCollection(rsd_lib_base.ResourceCollectionBase):
@property
def _resource_type(self):
return EthernetSwitchStaticMAC

View File

@ -1,199 +0,0 @@
# Copyright 2018 99cloud, 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.
import logging
from sushy.resources import base
from sushy import utils
from rsd_lib import common as rsd_lib_common
from rsd_lib.resources.v2_1.ethernet_switch import static_mac
from rsd_lib.resources.v2_1.ethernet_switch import vlan
from rsd_lib import utils as rsd_lib_utils
LOG = logging.getLogger(__name__)
class NeighborInfoField(base.CompositeField):
switch_id = base.Field('SwitchId')
port_id = base.Field('PortId')
cable_id = base.Field('CableId')
class IPv4AddressesField(base.ListField):
address = base.Field('Address')
"""The port ipv4 address"""
subnet_mask = base.Field('SubnetMask')
"""The port ipv4 address subnet mask"""
address_origin = base.Field('AddressOrigin')
"""The port ipv4 address origin"""
gateway = base.Field('Gateway')
"""The port ipv4 address gateway"""
class IPv6AddressesField(base.ListField):
address = base.Field('Address')
"""The port ipv6 address"""
prefix_length = base.Field(
'PrefixLength', adapter=rsd_lib_utils.num_or_none)
"""The port ipv6 address prefix length"""
address_origin = base.Field('AddressOrigin')
"""The port ipv6 address origin"""
address_state = base.Field('AddressState')
"""The port ipv6 address gateway"""
class LinksField(base.CompositeField):
primary_vlan = base.Field('PrimaryVLAN',
adapter=rsd_lib_utils.get_resource_identity)
switch = base.Field('Switch', adapter=rsd_lib_utils.get_resource_identity)
member_of_port = base.Field('MemberOfPort',
adapter=rsd_lib_utils.get_resource_identity)
port_members = base.Field('PortMembers')
active_acls = base.Field('ActiveACLs',
adapter=utils.get_members_identities)
class Port(base.ResourceBase):
identity = base.Field('Id', required=True)
"""The port identity string"""
name = base.Field('Name')
"""The port name"""
description = base.Field('Description')
"""The port description"""
port_id = base.Field('PortId')
"""The port id"""
status = rsd_lib_common.StatusField('Status')
"""The port status"""
link_type = base.Field('LinkType')
"""The port link type"""
operational_state = base.Field('OperationalState')
"""The port operational state"""
administrative_state = base.Field('AdministrativeState')
"""The port administrative state"""
link_speed_mbps = base.Field(
'LinkSpeedMbps', adapter=rsd_lib_utils.num_or_none)
"""The port link speed(mbps)"""
neighbor_info = NeighborInfoField('NeighborInfo')
"""The port neighbor info"""
neighbor_mac = base.Field('NeighborMAC')
"""The port neighbor mac"""
frame_size = base.Field(
'FrameSize', adapter=rsd_lib_utils.num_or_none)
"""The port frame size"""
autosense = base.Field('Autosense', adapter=bool)
"""The boolean indicate the autosense is enabled or not"""
full_duplex = base.Field('FullDuplex', adapter=bool)
"""The boolean indicate the full duplex is enabled or not"""
mac_address = base.Field('MACAddress')
"""The port mac address"""
ipv4_addresses = IPv4AddressesField('IPv4Addresses')
"""The port ipv4 link info"""
ipv6_addresses = IPv6AddressesField('IPv6Addresses')
"""The port ipv6 link info"""
port_class = base.Field('PortClass')
"""The port class"""
port_mode = base.Field('PortMode')
"""The port mode"""
port_type = base.Field('PortType')
"""The port type"""
links = LinksField('Links')
"""The port links"""
def __init__(self, connector, identity, redfish_version=None):
"""A class representing an Port
:param connector: A Connector instance
:param identity: The identity of the Port resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(Port, self).__init__(connector, identity, redfish_version)
def _get_static_mac_collection_path(self):
"""Helper function to find the StaticMACCollection path"""
return utils.get_sub_resource_path_by(self, 'StaticMACs')
@property
@utils.cache_it
def static_macs(self):
"""Property to provide reference to `StaticMACollection` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return static_mac.StaticMACCollection(
self._conn, self._get_static_mac_collection_path(),
redfish_version=self.redfish_version)
def _get_vlan_collection_path(self):
"""Helper function to find the VLANCollection path"""
return utils.get_sub_resource_path_by(self, 'VLANs')
@property
@utils.cache_it
def vlans(self):
"""Property to provide reference to `VLANCollection` instance
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return vlan.VLANCollection(
self._conn, self._get_vlan_collection_path(),
redfish_version=self.redfish_version)
class PortCollection(base.ResourceCollectionBase):
@property
def _resource_type(self):
return Port
def __init__(self, connector, path, redfish_version=None):
"""A class representing an Port
:param connector: A Connector instance
:param path: The canonical path to the Port collection resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(PortCollection, self).__init__(connector, path, redfish_version)

View File

@ -14,148 +14,129 @@
# under the License.
vlan_network_interface_req_schema = {
'type': 'object',
'properties': {
'VLANId': {'type': 'number'},
'VLANEnable': {'type': 'boolean'},
'Oem': {
'type': 'object',
'properties': {
'Intel_RackScale': {
'type': 'object',
'properties': {
'Tagged': {'type': 'boolean'}
},
'required': ['Tagged']
"type": "object",
"properties": {
"VLANId": {"type": "number"},
"VLANEnable": {"type": "boolean"},
"Oem": {
"type": "object",
"properties": {
"Intel_RackScale": {
"type": "object",
"properties": {"Tagged": {"type": "boolean"}},
"required": ["Tagged"],
}
},
'required': ['Intel_RackScale']
}
"required": ["Intel_RackScale"],
},
},
'required': [
'VLANId',
'VLANEnable',
'Oem'
],
'additionalProperties': False
"required": ["VLANId", "VLANEnable", "Oem"],
"additionalProperties": False,
}
acl_rule_req_schema = {
'type': 'object',
'oneOf': [
"type": "object",
"oneOf": [
{
'properties': {
'Action': {'enum': ['Forward']}
},
'required': ['ForwardMirrorInterface']
"properties": {"Action": {"enum": ["Forward"]}},
"required": ["ForwardMirrorInterface"],
},
{
'properties': {
'Action': {'enum': ['Mirror']}
},
'required': ['ForwardMirrorInterface',
'MirrorPortRegion', 'MirrorType']
"properties": {"Action": {"enum": ["Mirror"]}},
"required": [
"ForwardMirrorInterface",
"MirrorPortRegion",
"MirrorType",
],
},
{
'properties': {
'Action': {'enum': ['Permit', 'Deny']}
}
}
{"properties": {"Action": {"enum": ["Permit", "Deny"]}}},
],
'properties': {
'RuleId': {'type': 'number'},
'Action': {
'type': 'string',
'enum': ['Permit', 'Deny', 'Forward', 'Mirror']
"properties": {
"RuleId": {"type": "number"},
"Action": {
"type": "string",
"enum": ["Permit", "Deny", "Forward", "Mirror"],
},
'ForwardMirrorInterface': {
'type': 'object',
'properties': {
'@odata.id': {
'type': 'string'
}
"ForwardMirrorInterface": {
"type": "object",
"properties": {"@odata.id": {"type": "string"}},
"required": ["@odata.id"],
},
"MirrorPortRegion": {
"type": "array",
"items": {
"type": "object",
"properties": {"@odata.id": {"type": "string"}},
"required": ["@odata.id"],
},
'required': ['@odata.id']
},
'MirrorPortRegion': {
'type': 'array',
'items': {
'type': 'object',
'properties': {
'@odata.id': {
'type': 'string'
}
},
'required': ['@odata.id']
}
"MirrorType": {
"type": "string",
"enum": ["Egress", "Ingress", "Bidirectional", "Redirect"],
},
'MirrorType': {
'type': 'string',
'enum': ['Egress', 'Ingress', 'Bidirectional', 'Redirect']
},
'Condition': {
'type': 'object',
'properties': {
'IPSource': {
'type': 'object',
'properties': {
'IPv4Addresses': {'type': 'string'},
'Mask': {'type': ['string', 'null']}
"Condition": {
"type": "object",
"properties": {
"IPSource": {
"type": "object",
"properties": {
"IPv4Addresses": {"type": "string"},
"Mask": {"type": ["string", "null"]},
},
'required': ['IPv4Address']
"required": ["IPv4Address"],
},
'IPDestination': {
'type': 'object',
'properties': {
'IPv4Address': {'type': 'string'},
'Mask': {'type': ['string', 'null']}
"IPDestination": {
"type": "object",
"properties": {
"IPv4Address": {"type": "string"},
"Mask": {"type": ["string", "null"]},
},
'required': ['IPv4Address']
"required": ["IPv4Address"],
},
'MACSource': {
'type': 'object',
'properties': {
'MACAddress': {'type': 'string'},
'Mask': {'type': ['string', 'null']}
"MACSource": {
"type": "object",
"properties": {
"MACAddress": {"type": "string"},
"Mask": {"type": ["string", "null"]},
},
'required': ['MACAddress']
"required": ["MACAddress"],
},
'MACDestination': {
'type': 'object',
'properties': {
'MACAddress': {'type': 'string'},
'Mask': {'type': ['string', 'null']}
"MACDestination": {
"type": "object",
"properties": {
"MACAddress": {"type": "string"},
"Mask": {"type": ["string", "null"]},
},
'required': ['MACAddress']
"required": ["MACAddress"],
},
'VLANId': {
'type': 'object',
'properties': {
'Id': {'type': 'number'},
'Mask': {'type': ['number', 'null']}
"VLANId": {
"type": "object",
"properties": {
"Id": {"type": "number"},
"Mask": {"type": ["number", "null"]},
},
'required': ['Id']
"required": ["Id"],
},
'L4SourcePort': {
'type': 'object',
'properties': {
'Port': {'type': 'number'},
'Mask': {'type': ['number', 'null']}
"L4SourcePort": {
"type": "object",
"properties": {
"Port": {"type": "number"},
"Mask": {"type": ["number", "null"]},
},
'required': ['Port']
"required": ["Port"],
},
'L4DestinationPort': {
'type': 'object',
'properties': {
'Port': {'type': 'number'},
'Mask': {'type': ['number', 'null']}
"L4DestinationPort": {
"type": "object",
"properties": {
"Port": {"type": "number"},
"Mask": {"type": ["number", "null"]},
},
'required': ['Port']
"required": ["Port"],
},
'L4Protocol': {'type': ['number', 'null']}
}
"L4Protocol": {"type": ["number", "null"]},
},
},
},
'required': ['Action', 'Condition'],
'additionalProperties': False
"required": ["Action", "Condition"],
"additionalProperties": False,
}

View File

@ -1,68 +0,0 @@
# Copyright 2018 99cloud, 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.
from sushy.resources import base
from rsd_lib import utils as rsd_lib_utils
class StaticMAC(base.ResourceBase):
identity = base.Field('Id', required=True)
"""The static mac identity string"""
name = base.Field('Name')
"""The static mac name"""
description = base.Field('Description')
"""The static mac description"""
mac_address = base.Field('MACAddress')
"""The static mac address"""
vlan_id = base.Field('VLANId', adapter=rsd_lib_utils.num_or_none)
"""The static mac vlan id"""
oem = base.Field('Oem')
"""The static mac oem info"""
def __init__(self, connector, identity, redfish_version=None):
"""A class representing an StaticMAC
:param connector: A Connector instance
:param identity: The identity of the StaticMAC resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(StaticMAC, self).__init__(connector, identity, redfish_version)
class StaticMACCollection(base.ResourceCollectionBase):
@property
def _resource_type(self):
return StaticMAC
def __init__(self, connector, path, redfish_version=None):
"""A class representing an StaticMAC
:param connector: A Connector instance
:param identity: The identity of the StaticMAC Collection resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(StaticMACCollection, self).__init__(connector,
path,
redfish_version)

View File

@ -1,88 +0,0 @@
# Copyright 2018 99cloud, 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.
from jsonschema import validate
import logging
from sushy.resources import base
from rsd_lib.resources.v2_1.ethernet_switch import schemas as \
ethernet_switch_schemas
from rsd_lib import utils as rsd_lib_utils
LOG = logging.getLogger(__name__)
class VLAN(base.ResourceBase):
identity = base.Field('Id', required=True)
"""The vlan network interface identity"""
name = base.Field('Name')
"""The vlan network interface name"""
description = base.Field('Description')
"""The vlan network interface description"""
vlan_enable = base.Field('VLANEnable', adapter=bool)
"""The boolean indicate this vlan network interface is enabled or not"""
vlan_id = base.Field('VLANId', adapter=rsd_lib_utils.num_or_none)
"""The vlan network interface id"""
oem = base.Field('Oem')
"""The vlan network interface oem info"""
def __init__(self, connector, identity, redfish_version=None):
"""A class representing an VLAN network interface
:param connector: A Connector instance
:param identity: The identity of the VLAN network interface resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(VLAN, self).__init__(connector, identity, redfish_version)
class VLANCollection(base.ResourceCollectionBase):
@property
def _resource_type(self):
return VLAN
def __init__(self, connector, path, redfish_version=None):
"""A class representing an VLAN network interface collection
:param connector: A Connector instance
:param path: The canonical path to the VLAN network interface
collection resource
:param redfish_version: The version of RedFish. Used to construct
the object according to schema of the given version.
"""
super(VLANCollection, self).__init__(connector, path, redfish_version)
def add_vlan(self, vlan_network_interface_req):
"""Add a vlan to port
:param vlan_network_interface_req: JSON for vlan network interface
:returns: The location of the vlan network interface
"""
target_uri = self._path
validate(vlan_network_interface_req,
ethernet_switch_schemas.vlan_network_interface_req_schema)
resp = self._conn.post(target_uri, data=vlan_network_interface_req)
LOG.info("VLAN add at %s", resp.headers['Location'])
vlan_network_interface_url = resp.headers['Location']
return vlan_network_interface_url[vlan_network_interface_url.
find(self._path):]

View File

@ -0,0 +1,84 @@
# Copyright 2018 99cloud, 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.
from jsonschema import validate
import logging
from sushy.resources import base
from rsd_lib import base as rsd_lib_base
from rsd_lib.resources.v2_1.ethernet_switch import schemas \
as ethernet_switch_schemas
LOG = logging.getLogger(__name__)
class IntelRackScaleField(base.CompositeField):
status = rsd_lib_base.StatusField("Status")
"""This indicates the known state of the resource, such as if it is
enabled.
"""
tagged = base.Field("Tagged", adapter=bool)
"""This indicates if VLAN is tagged (as defined in IEEE 802.1Q)."""
class OemField(base.CompositeField):
intel_rackscale = IntelRackScaleField("Intel_RackScale")
"""Intel Rack Scale Design specific properties."""
class VLanNetworkInterface(rsd_lib_base.ResourceBase):
"""VLanNetworkInterface resource class
This resource contains information for a Virtual LAN (VLAN) network
instance available on a manager, system or other device.
"""
vlan_enable = base.Field("VLANEnable", adapter=bool)
"""This indicates if this VLAN is enabled."""
vlan_id = base.Field("VLANId")
"""This indicates the VLAN identifier for this VLAN."""
oem = OemField("Oem")
"""Oem specific properties."""
class VLanNetworkInterfaceCollection(rsd_lib_base.ResourceCollectionBase):
@property
def _resource_type(self):
return VLanNetworkInterface
def add_vlan(self, vlan_network_interface_req):
"""Add a vlan to port
:param vlan_network_interface_req: JSON for vlan network interface
:returns: The location of the vlan network interface
"""
target_uri = self._path
validate(
vlan_network_interface_req,
ethernet_switch_schemas.vlan_network_interface_req_schema,
)
resp = self._conn.post(target_uri, data=vlan_network_interface_req)
LOG.info("VLAN add at %s", resp.headers["Location"])
vlan_network_interface_url = resp.headers["Location"]
return vlan_network_interface_url[
vlan_network_interface_url.find(self._path):
]

View File

@ -17,7 +17,7 @@ from sushy.resources import base
from sushy import utils
from rsd_lib import common as rsd_lib_common
from rsd_lib.resources.v2_1.ethernet_switch import vlan
from rsd_lib.resources.v2_1.ethernet_switch import vlan_network_interface
from rsd_lib import utils as rsd_lib_utils
@ -174,7 +174,7 @@ class EthernetInterface(base.ResourceBase):
It is calculated once when it is queried for the first time. On
refresh, this property is reset.
"""
return vlan.VLANCollection(
return vlan_network_interface.VLanNetworkInterfaceCollection(
self._conn, self._get_vlan_collection_path(),
redfish_version=self.redfish_version)

View File

@ -33,7 +33,7 @@ class EthernetSwitch(v2_1_ethernet_switch.EthernetSwitch):
refresh, this property is reset.
"""
return port.PortCollection(
self._conn, self._get_port_collection_path(),
self._conn, utils.get_sub_resource_path_by(self, 'Metrics'),
redfish_version=self.redfish_version)
def _get_metrics_path(self):

View File

@ -16,11 +16,12 @@
from sushy.resources import base
from sushy import utils
from rsd_lib.resources.v2_1.ethernet_switch import port as v2_1_port
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_port \
as v2_1_port
from rsd_lib.resources.v2_2.ethernet_switch import port_metrics
class Port(v2_1_port.Port):
class Port(v2_1_port.EthernetSwitchPort):
def _get_metrics_path(self):
"""Helper function to find the Port metrics path"""

View File

@ -26,7 +26,7 @@
},
"IPDestination": null,
"MACSource": {
"Address": "00:11:22:33:44:55",
"MACAddress": "00:11:22:33:44:55",
"Mask": null
},
"MACDestination": null,

View File

@ -1,150 +0,0 @@
# Copyright 2018 99cloud, 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.
import json
import mock
import testtools
from sushy import exceptions
from rsd_lib.resources.v2_1.ethernet_switch import acl
from rsd_lib.resources.v2_1.ethernet_switch import acl_rule
class ACLTestCase(testtools.TestCase):
def setUp(self):
super(ACLTestCase, self).setUp()
self.conn = mock.Mock()
with open(
'rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_acl.'
'json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_inst = acl.ACL(
self.conn, '/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1',
redfish_version='1.0.2'
)
def test__parse_attributes(self):
self.acl_inst._parse_attributes()
self.assertEqual('1.0.2', self.acl_inst.redfish_version)
self.assertEqual('ACL1', self.acl_inst.identity)
self.assertEqual('Ethernet Switch Access Control List',
self.acl_inst.name)
self.assertEqual('Switch ACL', self.acl_inst.description)
self.assertEqual({}, self.acl_inst.oem)
self.assertEqual(('/redfish/v1/EthernetSwitches/Switch1/Ports/sw0p1',),
self.acl_inst.links.bound_ports)
def test__get_acl_rule_collection_path(self):
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules',
self.acl_inst._get_acl_rule_collection_path())
def test__get_acl_rule_collection_path_missing_attr(self):
self.acl_inst._json.pop('Rules')
self.assertRaisesRegex(
exceptions.MissingAttributeError, 'attribute Rules',
self.acl_inst._get_acl_rule_collection_path)
def test_acl_rule(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_acl.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_acl_rules = self.acl_inst.rules
# | THEN |
self.assertIsInstance(actual_acl_rules,
acl_rule.ACLRuleCollection)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
# tests for same object on invoking subsequently
self.assertIs(actual_acl_rules,
self.acl_inst.rules)
self.conn.get.return_value.json.assert_not_called()
def test_acl_rule_on_refresh(self):
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'acl_rule_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.acl_inst.rules,
acl_rule.ACLRuleCollection)
# On refreshing the acl_rule instance...
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_acl.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_inst.invalidate()
self.acl_inst.refresh(force=False)
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'acl_rule_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.acl_inst.rules,
acl_rule.ACLRuleCollection)
class ACLCollectionTestCase(testtools.TestCase):
def setUp(self):
super(ACLCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_acl_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_col = acl.ACLCollection(
self.conn, '/redfish/v1/EthernetSwitches/ACLs',
redfish_version='1.0.2')
def test__parse_attributes(self):
self.acl_col._parse_attributes()
self.assertEqual('1.0.2', self.acl_col.redfish_version)
self.assertEqual(
'Ethernet Switch Access Control List Collection',
self.acl_col.name)
self.assertEqual(
('/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1',),
self.acl_col.members_identities)
@mock.patch.object(acl, 'ACL', autospec=True)
def test_get_member(self, mock_acl):
self.acl_col.get_member(
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1')
mock_acl.assert_called_once_with(
self.acl_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1',
redfish_version=self.acl_col.redfish_version)
@mock.patch.object(acl, 'ACL', autospec=True)
def test_get_members(self, mock_acl):
members = self.acl_col.get_members()
mock_acl.assert_called_with(
self.acl_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1',
redfish_version=self.acl_col.redfish_version)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))

View File

@ -1,257 +0,0 @@
# Copyright 2018 99cloud, 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.
import json
import jsonschema
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import acl_rule
from rsd_lib.tests.unit.fakes import request_fakes
class ACLRuleTestCase(testtools.TestCase):
def setUp(self):
super(ACLRuleTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/acl_rule.json',
'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_rule_inst = acl_rule.ACLRule(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules/Rule1',
redfish_version='1.0.2')
def test__parse_attributes(self):
self.acl_rule_inst._parse_attributes()
self.assertEqual('1.0.2', self.acl_rule_inst.redfish_version)
self.assertEqual('Rule1', self.acl_rule_inst.identity)
self.assertEqual('Example Rule', self.acl_rule_inst.name)
self.assertEqual('User defined rule for ACL',
self.acl_rule_inst.description)
self.assertEqual(1, self.acl_rule_inst.rule_id)
self.assertEqual('Mirror', self.acl_rule_inst.action)
self.assertEqual('/redfish/v1/EthernetSwitches/Switch1/Ports/Port9',
self.acl_rule_inst.forward_mirror_interface)
self.assertEqual(('/redfish/v1/EthernetSwitches/Switch1/Ports/Port1',
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port2',),
self.acl_rule_inst.mirror_port_region)
self.assertEqual('Bidirectional',
self.acl_rule_inst.mirror_type)
self.assertEqual('192.168.1.0',
self.acl_rule_inst.condition.ip_source.ipv4_address)
self.assertEqual('0.0.0.255',
self.acl_rule_inst.condition.ip_source.mask)
self.assertEqual(None, self.acl_rule_inst.condition.ip_destination)
self.assertEqual('00:11:22:33:44:55',
self.acl_rule_inst.condition.mac_source.mac_address)
self.assertEqual(None, self.acl_rule_inst.condition.mac_source.mask)
self.assertEqual(1088, self.acl_rule_inst.condition.vlan_id.id)
self.assertEqual(4095,
self.acl_rule_inst.condition.vlan_id.mask)
self.assertEqual(22,
self.acl_rule_inst.condition.l4_source_port.port)
self.assertEqual(255,
self.acl_rule_inst.condition.l4_source_port.mask)
self.assertEqual(None,
self.acl_rule_inst.condition.l4_destination_port)
class ACLRuleCollectionTestCase(testtools.TestCase):
def setUp(self):
super(ACLRuleCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'acl_rule_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.conn.post.return_value = \
request_fakes.fake_request_post(
None,
headers={"Location": "https://localhost:8443/redfish/v1/"
"EthernetSwitches/Switch1/ACLs/ACL1/"
"Rules/Rule1"})
self.acl_rule_col = acl_rule.ACLRuleCollection(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules',
redfish_version='1.0.2')
def test__parse_attributes(self):
self.acl_rule_col._parse_attributes()
self.assertEqual('1.0.2', self.acl_rule_col.redfish_version)
self.assertEqual('Ethernet Switch Access Control '
'List Rules Collection',
self.acl_rule_col.name)
self.assertEqual(('/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/'
'Rules/Rule1',),
self.acl_rule_col.members_identities)
@mock.patch.object(acl_rule, 'ACLRule', autospec=True)
def test_get_member(self, mock_acl_rule):
self.acl_rule_col.get_member(
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules/Rule1')
mock_acl_rule.assert_called_once_with(
self.acl_rule_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules/Rule1',
redfish_version=self.acl_rule_col.redfish_version)
@mock.patch.object(acl_rule, 'ACLRule', autospec=True)
def test_get_members(self, mock_acl_rule):
members = self.acl_rule_col.get_members()
calls = [
mock.call(self.acl_rule_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/'
'Rules/Rule1',
redfish_version=self.acl_rule_col.redfish_version)
]
mock_acl_rule.assert_has_calls(calls)
self.assertEqual(mock_acl_rule.call_count, 1)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))
def test_add_acl_rule_reqs(self):
reqs = {
'RuleId': 1,
'Action': 'Mirror',
'ForwardMirrorInterface': {
'@odata.id': '/redfish/v1/EthernetSwitches/Switch1/Ports/Port9'
},
'MirrorPortRegion': [
{
'@odata.id': '/redfish/v1/EthernetSwitches/Switch1/Ports/'
'Port1'
}
],
'MirrorType': 'Bidirectional',
'Condition': {
'IPSource': {
'IPv4Address': '192.168.8.0',
'Mask': '0.0.0.255'
},
'IPDestination': {
'IPv4Address': '192.168.1.0'
},
'MACSource': {
'MACAddress': '00:11:22:33:44:55',
},
'MACDestination': {
'MACAddress': '55:44:33:22:11:00'
},
'VLANid': {
'Id': 1088,
'Mask': 4095
},
'L4SourcePort': {
'Port': 22,
'Mask': 255
},
'L4DestinationPort': {
'Port': 22,
'Mask': 255
},
'L4Protocol': 1
}
}
result = self.acl_rule_col.add_acl_rule(reqs)
self.acl_rule_col._conn.post.assert_called_once_with(
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules',
data=reqs)
self.assertEqual(result,
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/'
'Rules/Rule1')
def test_add_acl_rule_invalid_reqs(self):
reqs = {
'RuleId': 1,
'Action': 'Mirror',
'ForwardMirrorInterface': {
'@odata.id': '/redfish/v1/EthernetSwitches/Switch1/Ports/Port9'
},
'MirrorPortRegion': [
{
'@odata.id': '/redfish/v1/EthernetSwitches/Switch1/Ports/'
'Port1'
}
],
'MirrorType': 'Bidirectional',
'Condition': {
'IPSource': {
'IPv4Address': '192.168.8.0',
'Mask': '0.0.0.255'
},
'IPDestination': {
'IPv4Address': '192.168.1.0'
},
'MACSource': {
'MACAddress': '00:11:22:33:44:55',
},
'MACDestination': {
'MACAddress': '55:44:33:22:11:00'
},
'VLANid': {
'Id': 1088,
'Mask': 4095
},
'L4SourcePort': {
'Port': 22,
'Mask': 255
},
'L4DestinationPort': {
'Port': 22,
'Mask': 255
},
'L4Protocol': 1
}
}
# Missing field
acl_rule_req = reqs.copy()
acl_rule_req.pop('Action')
self.assertRaises(jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req)
# Wrong format
acl_rule_req = reqs.copy()
acl_rule_req.update({'RuleId': 'WrongFormat'})
self.assertRaises(jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req)
# Wrong additional fields
acl_rule_req = reqs.copy()
acl_rule_req['Additional'] = 'AdditionalField'
self.assertRaises(jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req)
# Wrong enum
acl_rule_req = reqs.copy()
acl_rule_req['MirrorType'] = 'WrongEnum'
self.assertRaises(jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req)
# Wrong dependency
acl_rule_req = reqs.copy()
acl_rule_req.pop('ForwardMirrorInterface')
self.assertRaises(jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req)

View File

@ -16,200 +16,218 @@
import json
import mock
from sushy import exceptions
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import acl
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch
from rsd_lib.resources.v2_1.ethernet_switch import port
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_acl
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_port
class EthernetSwtichTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwtichTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch.json',
'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch.json", "r"
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.ethernet_switch_inst = ethernet_switch.EthernetSwitch(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1',
redfish_version='1.0.2')
"/redfish/v1/EthernetSwitches/Switch1",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.ethernet_switch_inst._parse_attributes()
self.assertEqual('1.0.2', self.ethernet_switch_inst.redfish_version)
self.assertEqual('Switch1', self.ethernet_switch_inst.identity)
self.assertEqual('Switch1', self.ethernet_switch_inst.name)
self.assertEqual('description-as-string',
self.ethernet_switch_inst.description)
self.assertEqual('Quanta', self.ethernet_switch_inst.manufacturer)
self.assertEqual('ly8_rangley', self.ethernet_switch_inst.model)
self.assertEqual('02/21/2015 00:00:00',
self.ethernet_switch_inst.manufacturing_date)
self.assertEqual('2M220100SL', self.ethernet_switch_inst.serial_number)
self.assertEqual('1LY8UZZ0007', self.ethernet_switch_inst.part_number)
self.assertEqual('ONIE', self.ethernet_switch_inst.firmware_name)
self.assertEqual('1.1', self.ethernet_switch_inst.firmware_version)
self.assertEqual('TOR', self.ethernet_switch_inst.role)
self.assertEqual('Enabled', self.ethernet_switch_inst.status.state)
self.assertEqual('OK', self.ethernet_switch_inst.status.health)
self.assertEqual('/redfish/v1/Chassis/FabricModule1',
self.ethernet_switch_inst.links.chassis)
self.assertEqual(('/redfish/v1/Managers/Manager1',),
self.ethernet_switch_inst.links.managed_by)
def test__get_port_collection_path(self):
expected = '/redfish/v1/EthernetSwitches/Switch1/Ports'
result = self.ethernet_switch_inst._get_port_collection_path()
self.assertEqual(expected, result)
def test__get_port_collection_path_missing_attr(self):
self.ethernet_switch_inst._json.pop('Ports')
self.assertRaisesRegex(
exceptions.MissingAttributeError, 'attribute Ports',
self.ethernet_switch_inst._get_port_collection_path)
self.assertEqual("1.0.2", self.ethernet_switch_inst.redfish_version)
self.assertEqual("Switch1", self.ethernet_switch_inst.identity)
self.assertEqual("Switch1", self.ethernet_switch_inst.name)
self.assertEqual(
"description-as-string", self.ethernet_switch_inst.description
)
self.assertEqual("Quanta", self.ethernet_switch_inst.manufacturer)
self.assertEqual("ly8_rangley", self.ethernet_switch_inst.model)
self.assertEqual(
"02/21/2015 00:00:00", self.ethernet_switch_inst.manufacturing_date
)
self.assertEqual("2M220100SL", self.ethernet_switch_inst.serial_number)
self.assertEqual("1LY8UZZ0007", self.ethernet_switch_inst.part_number)
self.assertEqual("ONIE", self.ethernet_switch_inst.firmware_name)
self.assertEqual("1.1", self.ethernet_switch_inst.firmware_version)
self.assertEqual("TOR", self.ethernet_switch_inst.role)
self.assertEqual("Enabled", self.ethernet_switch_inst.status.state)
self.assertEqual("OK", self.ethernet_switch_inst.status.health)
self.assertEqual(
"/redfish/v1/Chassis/FabricModule1",
self.ethernet_switch_inst.links.chassis,
)
self.assertEqual(
("/redfish/v1/Managers/Manager1",),
self.ethernet_switch_inst.links.managed_by,
)
def test_ports(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_ports = self.ethernet_switch_inst.ports
# | THEN |
self.assertIsInstance(actual_ports,
port.PortCollection)
self.assertIsInstance(
actual_ports, ethernet_switch_port.EthernetSwitchPortCollection
)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
# tests for same object on invoking subsequently
self.assertIs(actual_ports,
self.ethernet_switch_inst.ports)
self.assertIs(actual_ports, self.ethernet_switch_inst.ports)
self.conn.get.return_value.json.assert_not_called()
def test_ports_on_refresh(self):
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.ethernet_switch_inst.ports,
port.PortCollection)
self.assertIsInstance(
self.ethernet_switch_inst.ports,
ethernet_switch_port.EthernetSwitchPortCollection,
)
# On refreshing the port instance...
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/" "ethernet_switch.json", "r"
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.ethernet_switch_inst.invalidate()
self.ethernet_switch_inst.refresh(force=False)
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.ethernet_switch_inst.ports,
port.PortCollection)
self.assertIsInstance(
self.ethernet_switch_inst.ports,
ethernet_switch_port.EthernetSwitchPortCollection,
)
def test__get_acl_collection_path(self):
self.assertEqual('/redfish/v1/EthernetSwitches/Switch1/ACLs',
self.ethernet_switch_inst._get_acl_collection_path())
def test__get_acl_collection_path_missing_attr(self):
self.ethernet_switch_inst._json.pop('ACLs')
with self.assertRaisesRegex(
exceptions.MissingAttributeError, 'attribute ACLs'):
self.ethernet_switch_inst._get_acl_collection_path()
def test_acl(self):
def test_acls(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_acl_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_acls_col = self.ethernet_switch_inst.acls
# | THEN |
self.assertIsInstance(actual_acls_col,
acl.ACLCollection)
self.assertIsInstance(
actual_acls_col, ethernet_switch_acl.EthernetSwitchACLCollection
)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
self.assertIs(actual_acls_col,
self.ethernet_switch_inst.acls)
self.assertIs(actual_acls_col, self.ethernet_switch_inst.acls)
self.conn.get.return_value.json.assert_not_called()
def test_acls_on_refresh(self):
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_acl_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.ethernet_switch_inst.acls,
acl.ACLCollection)
self.assertIsInstance(
self.ethernet_switch_inst.acls,
ethernet_switch_acl.EthernetSwitchACLCollection,
)
# On refreshing...
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/" "ethernet_switch.json", "r"
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.ethernet_switch_inst.invalidate()
self.ethernet_switch_inst.refresh(force=False)
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_acl_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.ethernet_switch_inst.acls,
acl.ACLCollection)
self.assertIsInstance(
self.ethernet_switch_inst.acls,
ethernet_switch_acl.EthernetSwitchACLCollection,
)
class EthernetSwitchCollectionTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_collection.json', 'r') as f:
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.ethernet_switch_col = ethernet_switch.EthernetSwitchCollection(
self.conn,
'redfish/v1/EthernetSwitches',
redfish_version='1.0.2')
self.conn, "redfish/v1/EthernetSwitches", redfish_version="1.0.2"
)
def test__parse_attributes(self):
self.ethernet_switch_col._parse_attributes()
self.assertEqual('1.0.2', self.ethernet_switch_col.redfish_version)
self.assertEqual('Ethernet Switches Collection',
self.ethernet_switch_col.name)
self.assertEqual(('/redfish/v1/EthernetSwitches/Switch1',),
self.ethernet_switch_col.members_identities)
self.assertEqual("1.0.2", self.ethernet_switch_col.redfish_version)
self.assertEqual(
"Ethernet Switches Collection", self.ethernet_switch_col.name
)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1",),
self.ethernet_switch_col.members_identities,
)
@mock.patch.object(ethernet_switch, 'EthernetSwitch', autospec=True)
@mock.patch.object(ethernet_switch, "EthernetSwitch", autospec=True)
def test_get_member(self, mock_ethernet_switch):
self.ethernet_switch_col.get_member(
'/redfish/v1/EthernetSwitches/Switch1')
"/redfish/v1/EthernetSwitches/Switch1"
)
mock_ethernet_switch.assert_called_once_with(
self.ethernet_switch_col._conn,
'/redfish/v1/EthernetSwitches/Switch1',
redfish_version=self.ethernet_switch_col.redfish_version
"/redfish/v1/EthernetSwitches/Switch1",
redfish_version=self.ethernet_switch_col.redfish_version,
)
@mock.patch.object(ethernet_switch, 'EthernetSwitch', autospec=True)
@mock.patch.object(ethernet_switch, "EthernetSwitch", autospec=True)
def test_get_members(self, mock_ethernet_switch):
members = self.ethernet_switch_col.get_members()
self.assertEqual(mock_ethernet_switch.call_count, 1)

View File

@ -0,0 +1,163 @@
# Copyright 2018 99cloud, 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.
import json
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_acl
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_acl_rule
class EthernetSwitchACLTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchACLTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_acl." "json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_inst = ethernet_switch_acl.EthernetSwitchACL(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.acl_inst._parse_attributes()
self.assertEqual("1.0.2", self.acl_inst.redfish_version)
self.assertEqual("ACL1", self.acl_inst.identity)
self.assertEqual(
"Ethernet Switch Access Control List", self.acl_inst.name
)
self.assertEqual("Switch ACL", self.acl_inst.description)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1/Ports/sw0p1",),
self.acl_inst.links.bound_ports,
)
def test_acl_rule(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/" "ethernet_switch_acl.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_acl_rules = self.acl_inst.rules
# | THEN |
self.assertIsInstance(
actual_acl_rules,
ethernet_switch_acl_rule.EthernetSwitchACLRuleCollection,
)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
# tests for same object on invoking subsequently
self.assertIs(actual_acl_rules, self.acl_inst.rules)
self.conn.get.return_value.json.assert_not_called()
def test_acl_rule_on_refresh(self):
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_rule_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.acl_inst.rules,
ethernet_switch_acl_rule.EthernetSwitchACLRuleCollection,
)
# On refreshing the acl_rule instance...
with open(
"rsd_lib/tests/unit/json_samples/v2_1/" "ethernet_switch_acl.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_inst.invalidate()
self.acl_inst.refresh(force=False)
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_rule_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.acl_inst.rules,
ethernet_switch_acl_rule.EthernetSwitchACLRuleCollection,
)
class EthernetSwitchACLCollectionTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchACLCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_col = ethernet_switch_acl.EthernetSwitchACLCollection(
self.conn,
"/redfish/v1/EthernetSwitches/ACLs",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.acl_col._parse_attributes()
self.assertEqual("1.0.2", self.acl_col.redfish_version)
self.assertEqual(
"Ethernet Switch Access Control List Collection", self.acl_col.name
)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1",),
self.acl_col.members_identities,
)
@mock.patch.object(ethernet_switch_acl, "EthernetSwitchACL", autospec=True)
def test_get_member(self, mock_acl):
self.acl_col.get_member(
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1"
)
mock_acl.assert_called_once_with(
self.acl_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1",
redfish_version=self.acl_col.redfish_version,
)
@mock.patch.object(ethernet_switch_acl, "EthernetSwitchACL", autospec=True)
def test_get_members(self, mock_acl):
members = self.acl_col.get_members()
mock_acl.assert_called_with(
self.acl_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1",
redfish_version=self.acl_col.redfish_version,
)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))

View File

@ -0,0 +1,374 @@
# Copyright 2018 99cloud, 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.
import json
import jsonschema
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_acl_rule
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_port
from rsd_lib.tests.unit.fakes import request_fakes
class EthernetSwitchACLRuleTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchACLRuleTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_rule.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_rule_inst = ethernet_switch_acl_rule.EthernetSwitchACLRule(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules/Rule1",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.acl_rule_inst._parse_attributes()
self.assertEqual("1.0.2", self.acl_rule_inst.redfish_version)
self.assertEqual("Rule1", self.acl_rule_inst.identity)
self.assertEqual("Example Rule", self.acl_rule_inst.name)
self.assertEqual(
"User defined rule for ACL", self.acl_rule_inst.description
)
self.assertEqual(1, self.acl_rule_inst.rule_id)
self.assertEqual("Mirror", self.acl_rule_inst.action)
self.assertEqual("Bidirectional", self.acl_rule_inst.mirror_type)
self.assertEqual(
"192.168.1.0", self.acl_rule_inst.condition.ip_source.ipv4_address
)
self.assertEqual(
"0.0.0.255", self.acl_rule_inst.condition.ip_source.mask
)
self.assertEqual(None, self.acl_rule_inst.condition.ip_destination)
self.assertEqual(
"00:11:22:33:44:55",
self.acl_rule_inst.condition.mac_source.mac_address,
)
self.assertEqual(None, self.acl_rule_inst.condition.mac_source.mask)
self.assertEqual(1088, self.acl_rule_inst.condition.vlan_id.identity)
self.assertEqual(4095, self.acl_rule_inst.condition.vlan_id.mask)
self.assertEqual(22, self.acl_rule_inst.condition.l4_source_port.port)
self.assertEqual(255, self.acl_rule_inst.condition.l4_source_port.mask)
self.assertEqual(
None, self.acl_rule_inst.condition.l4_destination_port
)
def test_forward_mirror_interface(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_forward_mirror_interface = (
self.acl_rule_inst.forward_mirror_interface
)
# | THEN |
self.assertIsInstance(
actual_forward_mirror_interface,
ethernet_switch_port.EthernetSwitchPort,
)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
# tests for same object on invoking subsequently
self.assertIs(
actual_forward_mirror_interface,
self.acl_rule_inst.forward_mirror_interface,
)
self.conn.get.return_value.json.assert_not_called()
def test_forward_mirror_interface_on_refresh(self):
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.acl_rule_inst.forward_mirror_interface,
ethernet_switch_port.EthernetSwitchPort,
)
# On refreshing the port instance...
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch.json", "r"
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_rule_inst.invalidate()
self.acl_rule_inst.refresh(force=False)
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.acl_rule_inst.forward_mirror_interface,
ethernet_switch_port.EthernetSwitchPort,
)
def test_mirror_port_region(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_mirror_port_region = (
self.acl_rule_inst.mirror_port_region
)
# | THEN |
self.assertIsInstance(actual_mirror_port_region, list)
self.conn.get.return_value.json.assert_called_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
# tests for same object on invoking subsequently
self.assertIs(
actual_mirror_port_region,
self.acl_rule_inst.mirror_port_region,
)
self.conn.get.return_value.json.assert_not_called()
def test_mirror_port_region_on_refresh(self):
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.acl_rule_inst.mirror_port_region, list)
# On refreshing the port instance...
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch.json", "r"
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.acl_rule_inst.invalidate()
self.acl_rule_inst.refresh(force=False)
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.acl_rule_inst.mirror_port_region, list)
class EthernetSwitchACLRuleCollectionTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchACLRuleCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_acl_rule_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.conn.post.return_value = request_fakes.fake_request_post(
None,
headers={
"Location": "https://localhost:8443/redfish/v1/"
"EthernetSwitches/Switch1/ACLs/ACL1/"
"Rules/Rule1"
},
)
self.acl_rule_col = ethernet_switch_acl_rule.\
EthernetSwitchACLRuleCollection(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.acl_rule_col._parse_attributes()
self.assertEqual("1.0.2", self.acl_rule_col.redfish_version)
self.assertEqual(
"Ethernet Switch Access Control " "List Rules Collection",
self.acl_rule_col.name,
)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/" "Rules/Rule1",),
self.acl_rule_col.members_identities,
)
@mock.patch.object(
ethernet_switch_acl_rule, "EthernetSwitchACLRule", autospec=True
)
def test_get_member(self, mock_acl_rule):
self.acl_rule_col.get_member(
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules/Rule1"
)
mock_acl_rule.assert_called_once_with(
self.acl_rule_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules/Rule1",
redfish_version=self.acl_rule_col.redfish_version,
)
@mock.patch.object(
ethernet_switch_acl_rule, "EthernetSwitchACLRule", autospec=True
)
def test_get_members(self, mock_acl_rule):
members = self.acl_rule_col.get_members()
calls = [
mock.call(
self.acl_rule_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/"
"Rules/Rule1",
redfish_version=self.acl_rule_col.redfish_version,
)
]
mock_acl_rule.assert_has_calls(calls)
self.assertEqual(mock_acl_rule.call_count, 1)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))
def test_add_acl_rule_reqs(self):
reqs = {
"RuleId": 1,
"Action": "Mirror",
"ForwardMirrorInterface": {
"@odata.id": "/redfish/v1/EthernetSwitches/Switch1/Ports/Port9"
},
"MirrorPortRegion": [
{
"@odata.id": "/redfish/v1/EthernetSwitches/Switch1/Ports/"
"Port1"
}
],
"MirrorType": "Bidirectional",
"Condition": {
"IPSource": {
"IPv4Address": "192.168.8.0",
"Mask": "0.0.0.255",
},
"IPDestination": {"IPv4Address": "192.168.1.0"},
"MACSource": {"MACAddress": "00:11:22:33:44:55"},
"MACDestination": {"MACAddress": "55:44:33:22:11:00"},
"VLANid": {"Id": 1088, "Mask": 4095},
"L4SourcePort": {"Port": 22, "Mask": 255},
"L4DestinationPort": {"Port": 22, "Mask": 255},
"L4Protocol": 1,
},
}
result = self.acl_rule_col.add_acl_rule(reqs)
self.acl_rule_col._conn.post.assert_called_once_with(
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/Rules", data=reqs
)
self.assertEqual(
result,
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1/" "Rules/Rule1",
)
def test_add_acl_rule_invalid_reqs(self):
reqs = {
"RuleId": 1,
"Action": "Mirror",
"ForwardMirrorInterface": {
"@odata.id": "/redfish/v1/EthernetSwitches/Switch1/Ports/Port9"
},
"MirrorPortRegion": [
{
"@odata.id": "/redfish/v1/EthernetSwitches/Switch1/Ports/"
"Port1"
}
],
"MirrorType": "Bidirectional",
"Condition": {
"IPSource": {
"IPv4Address": "192.168.8.0",
"Mask": "0.0.0.255",
},
"IPDestination": {"IPv4Address": "192.168.1.0"},
"MACSource": {"MACAddress": "00:11:22:33:44:55"},
"MACDestination": {"MACAddress": "55:44:33:22:11:00"},
"VLANid": {"Id": 1088, "Mask": 4095},
"L4SourcePort": {"Port": 22, "Mask": 255},
"L4DestinationPort": {"Port": 22, "Mask": 255},
"L4Protocol": 1,
},
}
# Missing field
acl_rule_req = reqs.copy()
acl_rule_req.pop("Action")
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req,
)
# Wrong format
acl_rule_req = reqs.copy()
acl_rule_req.update({"RuleId": "WrongFormat"})
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req,
)
# Wrong additional fields
acl_rule_req = reqs.copy()
acl_rule_req["Additional"] = "AdditionalField"
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req,
)
# Wrong enum
acl_rule_req = reqs.copy()
acl_rule_req["MirrorType"] = "WrongEnum"
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req,
)
# Wrong dependency
acl_rule_req = reqs.copy()
acl_rule_req.pop("ForwardMirrorInterface")
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.acl_rule_col.add_acl_rule,
acl_rule_req,
)

View File

@ -0,0 +1,286 @@
# Copyright 2018 99cloud, 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.
import json
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_port
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_static_mac
from rsd_lib.resources.v2_1.ethernet_switch import vlan_network_interface
class EthernetSwitchPortTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchPortTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_inst = ethernet_switch_port.EthernetSwitchPort(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.port_inst._parse_attributes()
self.assertEqual("1.0.2", self.port_inst.redfish_version)
self.assertEqual("Port1", self.port_inst.identity)
self.assertEqual("Switch Port", self.port_inst.name)
self.assertEqual("description-as-string", self.port_inst.description)
self.assertEqual("sw0p10", self.port_inst.port_id)
self.assertEqual("Enabled", self.port_inst.status.state)
self.assertEqual("OK", self.port_inst.status.health)
self.assertEqual("OK", self.port_inst.status.health_rollup)
self.assertEqual("Ethernet", self.port_inst.link_type)
self.assertEqual("Up", self.port_inst.operational_state)
self.assertEqual("Up", self.port_inst.administrative_state)
self.assertEqual(10000, self.port_inst.link_speed_mbps)
self.assertEqual("sw2", self.port_inst.neighbor_info.switch_id)
self.assertEqual("11", self.port_inst.neighbor_info.port_id)
self.assertEqual(
"CustomerWritableThing", self.port_inst.neighbor_info.cable_id
)
self.assertEqual("00:11:22:33:44:55", self.port_inst.neighbor_mac)
self.assertEqual(1520, self.port_inst.frame_size)
self.assertEqual(True, self.port_inst.autosense)
self.assertEqual(True, self.port_inst.full_duplex)
self.assertEqual("2c:60:0c:72:e6:33", self.port_inst.mac_address)
self.assertEqual(
"192.168.0.10", self.port_inst.ipv4_addresses[0].address
)
self.assertEqual(
"255.255.252.0", self.port_inst.ipv4_addresses[0].subnet_mask
)
self.assertEqual(
"Static", self.port_inst.ipv4_addresses[0].address_origin
)
self.assertEqual(
"192.168.0.1", self.port_inst.ipv4_addresses[0].gateway
)
self.assertEqual(
"fe80::1ec1:deff:fe6f:1e24",
self.port_inst.ipv6_addresses[0].address,
)
self.assertEqual(64, self.port_inst.ipv6_addresses[0].prefix_length)
self.assertEqual(
"Static", self.port_inst.ipv6_addresses[0].address_origin
)
self.assertEqual(
"Preferred", self.port_inst.ipv6_addresses[0].address_state
)
self.assertEqual("Logical", self.port_inst.port_class)
self.assertEqual("LinkAggregationStatic", self.port_inst.port_mode)
self.assertEqual("Upstream", self.port_inst.port_type)
self.assertEqual(
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1",
self.port_inst.links.primary_vlan,
)
self.assertEqual(
"/redfish/v1/EthernetSwitches/Switch1", self.port_inst.links.switch
)
self.assertEqual(
"/redfish/v1/EthernetSwitches/Switch1/Ports/LAG1",
self.port_inst.links.member_of_port,
)
self.assertEqual((), self.port_inst.links.port_members)
self.assertEqual(
"/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1",
self.port_inst.links.active_acls[0],
)
self.assertEqual(
"/redfish/v1/Systems/1/EthernetInterfaces/3",
self.port_inst.links.oem.intel_rackscale.neighbor_interface,
)
def test_static_mac(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_static_mac_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_static_macs = self.port_inst.static_macs
# | THEN |
self.assertIsInstance(
actual_static_macs,
ethernet_switch_static_mac.EthernetSwitchStaticMACCollection,
)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
self.assertIs(actual_static_macs, self.port_inst.static_macs)
self.conn.get.return_value.json.assert_not_called()
def test_static_mac_on_refresh(self):
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_static_mac_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.port_inst.static_macs,
ethernet_switch_static_mac.EthernetSwitchStaticMACCollection,
)
# On refreshing...
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_inst.invalidate()
self.port_inst.refresh(force=False)
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_static_mac_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.port_inst.static_macs,
ethernet_switch_static_mac.EthernetSwitchStaticMACCollection,
)
def test_vlans(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_vlan_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_vlans = self.port_inst.vlans
# | THEN |
self.assertIsInstance(
actual_vlans, vlan_network_interface.VLanNetworkInterfaceCollection
)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
self.assertIs(actual_vlans, self.port_inst.vlans)
self.conn.get.return_value.json.assert_not_called()
def test_vlans_on_refresh(self):
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_vlan_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.port_inst.vlans,
vlan_network_interface.VLanNetworkInterfaceCollection,
)
# On refreshing...
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_inst.invalidate()
self.port_inst.refresh(force=False)
# | GIVEN |
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_vlan_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(
self.port_inst.vlans,
vlan_network_interface.VLanNetworkInterfaceCollection,
)
class EthernetSwitchPortCollectionTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchPortCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_col = ethernet_switch_port.EthernetSwitchPortCollection(
self.conn,
"/redfish/v1/EthernetSwitches/Ports",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.port_col._parse_attributes()
self.assertEqual("1.0.2", self.port_col.redfish_version)
self.assertEqual("Ethernet Switch Port Collection", self.port_col.name)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1/Ports/Port1",),
self.port_col.members_identities,
)
@mock.patch.object(
ethernet_switch_port, "EthernetSwitchPort", autospec=True
)
def test_get_member(self, mock_port):
self.port_col.get_member(
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1"
)
mock_port.assert_called_once_with(
self.port_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1",
redfish_version=self.port_col.redfish_version,
)
@mock.patch.object(
ethernet_switch_port, "EthernetSwitchPort", autospec=True
)
def test_get_members(self, mock_port):
members = self.port_col.get_members()
mock_port.assert_called_with(
self.port_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1",
redfish_version=self.port_col.redfish_version,
)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))

View File

@ -0,0 +1,109 @@
# Copyright 2018 99cloud, 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.
import json
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import ethernet_switch_static_mac
class EthernetSwitchStaticMACTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchStaticMACTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_static_mac.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.static_mac_inst = ethernet_switch_static_mac.\
EthernetSwitchStaticMAC(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/StaticMACs/1",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.static_mac_inst._parse_attributes()
self.assertEqual("1.0.2", self.static_mac_inst.redfish_version)
self.assertEqual("1", self.static_mac_inst.identity)
self.assertEqual("StaticMAC", self.static_mac_inst.name)
self.assertEqual(
"description-as-string", self.static_mac_inst.description
)
self.assertEqual("00:11:22:33:44:55", self.static_mac_inst.mac_address)
self.assertEqual(112, self.static_mac_inst.vlan_id)
class EthernetSwitchStaticMACCollectionTestCase(testtools.TestCase):
def setUp(self):
super(EthernetSwitchStaticMACCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_static_mac_collection.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.static_mac_col = ethernet_switch_static_mac.\
EthernetSwitchStaticMACCollection(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/"
"StaticMACs",
redfish_version="1.1.0",
)
def test__parse_attributes(self):
self.static_mac_col._parse_attributes()
self.assertEqual("1.1.0", self.static_mac_col.redfish_version)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs/1",),
self.static_mac_col.members_identities,
)
@mock.patch.object(
ethernet_switch_static_mac, "EthernetSwitchStaticMAC", autospec=True
)
def test_get_member(self, mock_static_mac):
self.static_mac_col.get_member(
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs/1"
)
mock_static_mac.assert_called_once_with(
self.static_mac_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs/1",
redfish_version=self.static_mac_col.redfish_version,
)
@mock.patch.object(
ethernet_switch_static_mac, "EthernetSwitchStaticMAC", autopspec=True
)
def test_get_members(self, mock_static_mac):
members = self.static_mac_col.get_members()
calls = [
mock.call(
self.static_mac_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/"
"StaticMACs/1",
redfish_version=self.static_mac_col.redfish_version,
)
]
mock_static_mac.assert_has_calls(calls)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))

View File

@ -1,248 +0,0 @@
# Copyright 2018 99cloud, 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.
import json
import mock
from sushy import exceptions
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import port
from rsd_lib.resources.v2_1.ethernet_switch import static_mac
from rsd_lib.resources.v2_1.ethernet_switch import vlan
class PortTestCase(testtools.TestCase):
def setUp(self):
super(PortTestCase, self).setUp()
self.conn = mock.Mock()
with open(
'rsd_lib/tests/unit/json_samples/v2_1/ethernet_switch_port.json',
'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_inst = port.Port(
self.conn, '/redfish/v1/EthernetSwitches/Switch1/Ports/Port1',
redfish_version='1.0.2'
)
def test__parse_attributes(self):
self.port_inst._parse_attributes()
self.assertEqual('1.0.2', self.port_inst.redfish_version)
self.assertEqual('Port1', self.port_inst.identity)
self.assertEqual('Switch Port', self.port_inst.name)
self.assertEqual('description-as-string', self.port_inst.description)
self.assertEqual('sw0p10', self.port_inst.port_id)
self.assertEqual('Enabled', self.port_inst.status.state)
self.assertEqual('OK', self.port_inst.status.health)
self.assertEqual('OK', self.port_inst.status.health_rollup)
self.assertEqual('Ethernet', self.port_inst.link_type)
self.assertEqual('Up', self.port_inst.operational_state)
self.assertEqual('Up', self.port_inst.administrative_state)
self.assertEqual(10000, self.port_inst.link_speed_mbps)
self.assertEqual('sw2', self.port_inst.neighbor_info.switch_id)
self.assertEqual('11', self.port_inst.neighbor_info.port_id)
self.assertEqual('CustomerWritableThing',
self.port_inst.neighbor_info.cable_id)
self.assertEqual('00:11:22:33:44:55',
self.port_inst.neighbor_mac)
self.assertEqual(1520, self.port_inst.frame_size)
self.assertEqual(True, self.port_inst.autosense)
self.assertEqual(True, self.port_inst.full_duplex)
self.assertEqual('2c:60:0c:72:e6:33', self.port_inst.mac_address)
self.assertEqual('192.168.0.10',
self.port_inst.ipv4_addresses[0].address)
self.assertEqual('255.255.252.0',
self.port_inst.ipv4_addresses[0].subnet_mask)
self.assertEqual('Static',
self.port_inst.ipv4_addresses[0].address_origin)
self.assertEqual('192.168.0.1',
self.port_inst.ipv4_addresses[0].gateway)
self.assertEqual('fe80::1ec1:deff:fe6f:1e24',
self.port_inst.ipv6_addresses[0].address)
self.assertEqual(64, self.port_inst.ipv6_addresses[0].prefix_length)
self.assertEqual('Static',
self.port_inst.ipv6_addresses[0].address_origin)
self.assertEqual('Preferred',
self.port_inst.ipv6_addresses[0].address_state)
self.assertEqual('Logical', self.port_inst.port_class)
self.assertEqual('LinkAggregationStatic', self.port_inst.port_mode)
self.assertEqual('Upstream', self.port_inst.port_type)
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1',
self.port_inst.links.primary_vlan)
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1',
self.port_inst.links.switch)
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1/Ports/LAG1',
self.port_inst.links.member_of_port)
self.assertEqual([], self.port_inst.links.port_members)
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1/ACLs/ACL1',
self.port_inst.links.active_acls[0])
def test__get_static_mac_collection_path(self):
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs',
self.port_inst._get_static_mac_collection_path())
def test__get_static_mac_collection_path_missing_attr(self):
self.port_inst._json.pop('StaticMACs')
self.assertRaisesRegex(
exceptions.MissingAttributeError, 'attribute StaticMAC',
self.port_inst._get_static_mac_collection_path)
def test_static_mac(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_static_mac_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_static_macs = self.port_inst.static_macs
# | THEN |
self.assertIsInstance(actual_static_macs,
static_mac.StaticMACCollection)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
self.assertIs(actual_static_macs, self.port_inst.static_macs)
self.conn.get.return_value.json.assert_not_called()
def test_static_mac_on_refresh(self):
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_static_mac_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.port_inst.static_macs,
static_mac.StaticMACCollection)
# On refreshing...
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_inst.invalidate()
self.port_inst.refresh(force=False)
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_static_mac_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.port_inst.static_macs,
static_mac.StaticMACCollection)
def test__get_vlan_collection_path(self):
self.assertEqual(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs',
self.port_inst._get_vlan_collection_path())
def test__get_vlan_collection_path_missing_attr(self):
self.port_inst._json.pop('VLANs')
self.assertRaisesRegex(
exceptions.MissingAttributeError, 'attribute VLAN',
self.port_inst._get_vlan_collection_path)
def test_vlan(self):
# | GIVEN |
self.conn.get.return_value.json.reset_mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_vlan_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN |
actual_vlans = self.port_inst.vlans
# | THEN |
self.assertIsInstance(actual_vlans,
vlan.VLANCollection)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
self.conn.get.return_value.json.reset_mock()
# | WHEN & THEN |
self.assertIs(actual_vlans, self.port_inst.vlans)
self.conn.get.return_value.json.assert_not_called()
def test_vlan_on_refresh(self):
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_vlan_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.port_inst.vlans,
vlan.VLANCollection)
# On refreshing...
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_inst.invalidate()
self.port_inst.refresh(force=False)
# | GIVEN |
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_vlan_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
# | WHEN & THEN |
self.assertIsInstance(self.port_inst.vlans, vlan.VLANCollection)
class PortCollectionTestCase(testtools.TestCase):
def setUp(self):
super(PortCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open(
'rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.port_col = port.PortCollection(
self.conn, '/redfish/v1/EthernetSwitches/Ports',
redfish_version='1.0.2')
def test__parse_attributes(self):
self.port_col._parse_attributes()
self.assertEqual('1.0.2', self.port_col.redfish_version)
self.assertEqual(
'Ethernet Switch Port Collection',
self.port_col.name)
self.assertEqual(
('/redfish/v1/EthernetSwitches/Switch1/Ports/Port1',),
self.port_col.members_identities)
@mock.patch.object(port, 'Port', autospec=True)
def test_get_member(self, mock_port):
self.port_col.get_member(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1')
mock_port.assert_called_once_with(
self.port_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1',
redfish_version=self.port_col.redfish_version)
@mock.patch.object(port, 'Port', autospec=True)
def test_get_members(self, mock_port):
members = self.port_col.get_members()
mock_port.assert_called_with(
self.port_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1',
redfish_version=self.port_col.redfish_version)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))

View File

@ -1,91 +0,0 @@
# Copyright 2018 99cloud, 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.
import json
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import static_mac
class StaticMACTestCase(testtools.TestCase):
def setUp(self):
super(StaticMACTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_static_mac.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.static_mac_inst = static_mac.StaticMAC(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/StaticMACs/1',
redfish_version='1.0.2')
def test__parse_attributes(self):
self.static_mac_inst._parse_attributes()
self.assertEqual('1.0.2', self.static_mac_inst.redfish_version)
self.assertEqual('1', self.static_mac_inst.identity)
self.assertEqual('StaticMAC', self.static_mac_inst.name)
self.assertEqual('description-as-string',
self.static_mac_inst.description)
self.assertEqual('00:11:22:33:44:55', self.static_mac_inst.mac_address)
self.assertEqual(112, self.static_mac_inst.vlan_id)
self.assertEqual({}, self.static_mac_inst.oem)
class StaticMACCollectionTestCase(testtools.TestCase):
def setUp(self):
super(StaticMACCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_static_mac_collection.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.static_mac_col = static_mac.StaticMACCollection(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs',
redfish_version='1.1.0')
def test__parse_attributes(self):
self.static_mac_col._parse_attributes()
self.assertEqual('1.1.0', self.static_mac_col.redfish_version)
self.assertEqual(
('/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs/1',),
self.static_mac_col.members_identities)
@mock.patch.object(static_mac, 'StaticMAC', autospec=True)
def test_get_member(self, mock_static_mac):
self.static_mac_col.get_member(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs/1')
mock_static_mac.assert_called_once_with(
self.static_mac_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/StaticMACs/1',
redfish_version=self.static_mac_col.redfish_version)
@mock.patch.object(static_mac, 'StaticMAC', autopspec=True)
def test_get_members(self, mock_static_mac):
members = self.static_mac_col.get_members()
calls = [
mock.call(self.static_mac_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/'
'StaticMACs/1',
redfish_version=self.static_mac_col.redfish_version)
]
mock_static_mac.assert_has_calls(calls)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))

View File

@ -1,146 +0,0 @@
# Copyright 2018 99cloud, 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.
import json
import jsonschema
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import vlan
from rsd_lib.tests.unit.fakes import request_fakes
class VLANTestCase(testtools.TestCase):
def setUp(self):
super(VLANTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_vlan.json', 'r') as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.vlan_inst = vlan.VLAN(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1',
redfish_version='1.0.2')
def test__parse_attributes(self):
self.vlan_inst._parse_attributes()
self.assertEqual('1.0.2', self.vlan_inst.redfish_version)
self.assertEqual('VLAN1', self.vlan_inst.identity)
self.assertEqual('VLAN Network Interface', self.vlan_inst.name)
self.assertEqual('System NIC 1 VLAN', self.vlan_inst.description)
self.assertEqual(True, self.vlan_inst.vlan_enable)
self.assertEqual(101, self.vlan_inst.vlan_id)
class VLANCollectionTestCase(testtools.TestCase):
def setUp(self):
super(VLANCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open('rsd_lib/tests/unit/json_samples/v2_1/'
'ethernet_switch_port_vlan_collection.json', 'r') as f:
self.conn.get.return_value = \
request_fakes.fake_request_get(json.loads(f.read()))
self.conn.post.return_value = \
request_fakes.fake_request_post(
None,
headers={"Location": "https://localhost:8443/redfish/v1/"
"EthernetSwitches/Switch1/"
"Ports/Port1/VLANs/VLAN1"})
self.vlan_col = vlan.VLANCollection(
self.conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs',
redfish_version='1.1.0')
def test__parse_attributes(self):
self.vlan_col._parse_attributes()
self.assertEqual('1.1.0', self.vlan_col.redfish_version)
self.assertEqual(
('/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1',),
self.vlan_col.members_identities)
@mock.patch.object(vlan, 'VLAN', autospec=True)
def test_get_member(self, mock_vlan):
self.vlan_col.get_member(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1')
mock_vlan.assert_called_once_with(
self.vlan_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1',
redfish_version=self.vlan_col.redfish_version)
@mock.patch.object(vlan, 'VLAN', autopspec=True)
def test_get_members(self, mock_vlan):
members = self.vlan_col.get_members()
calls = [
mock.call(self.vlan_col._conn,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/'
'VLANs/VLAN1',
redfish_version=self.vlan_col.redfish_version)
]
mock_vlan.assert_has_calls(calls)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))
def test_add_vlan_reqs(self):
reqs = {
'VLANId': 101,
'VLANEnable': True,
'Oem': {
'Intel_RackScale': {
'Tagged': False
}
}
}
result = self.vlan_col.add_vlan(reqs)
self.vlan_col._conn.post.assert_called_once_with(
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs',
data=reqs)
self.assertEqual(result,
'/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/'
'VLANs/VLAN1')
def test_add_vlan_invalid_reqs(self):
reqs = {
'VLANId': 101,
'VLANEnable': True,
'Oem': {
'Intel_RackScale': {
'Tagged': False
}
}
}
# Missing filed
vlan_network_interface_req = reqs.copy()
vlan_network_interface_req.pop('VLANId')
self.assertRaises(jsonschema.exceptions.ValidationError,
self.vlan_col.add_vlan,
vlan_network_interface_req)
# Wrong format
vlan_network_interface_req = reqs.copy()
vlan_network_interface_req.update({'VLANId': 'WrongFormat'})
self.assertRaises(jsonschema.exceptions.ValidationError,
self.vlan_col.add_vlan,
vlan_network_interface_req)
# Wrong additional fields
vlan_network_interface_req = reqs.copy()
vlan_network_interface_req['Additional'] = 'AdditionalField'
self.assertRaises(jsonschema.exceptions.ValidationError,
self.vlan_col.add_vlan,
vlan_network_interface_req)

View File

@ -0,0 +1,163 @@
# Copyright 2018 99cloud, 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.
import json
import jsonschema
import mock
import testtools
from rsd_lib.resources.v2_1.ethernet_switch import vlan_network_interface
from rsd_lib.tests.unit.fakes import request_fakes
class VLanNetworkInterfaceTestCase(testtools.TestCase):
def setUp(self):
super(VLanNetworkInterfaceTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_vlan.json",
"r",
) as f:
self.conn.get.return_value.json.return_value = json.loads(f.read())
self.vlan_inst = vlan_network_interface.VLanNetworkInterface(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1",
redfish_version="1.0.2",
)
def test__parse_attributes(self):
self.vlan_inst._parse_attributes()
self.assertEqual("1.0.2", self.vlan_inst.redfish_version)
self.assertEqual("VLAN1", self.vlan_inst.identity)
self.assertEqual("VLAN Network Interface", self.vlan_inst.name)
self.assertEqual("System NIC 1 VLAN", self.vlan_inst.description)
self.assertEqual(True, self.vlan_inst.vlan_enable)
self.assertEqual(101, self.vlan_inst.vlan_id)
class VLanNetworkInterfaceCollectionTestCase(testtools.TestCase):
def setUp(self):
super(VLanNetworkInterfaceCollectionTestCase, self).setUp()
self.conn = mock.Mock()
with open(
"rsd_lib/tests/unit/json_samples/v2_1/"
"ethernet_switch_port_vlan_collection.json",
"r",
) as f:
self.conn.get.return_value = request_fakes.fake_request_get(
json.loads(f.read())
)
self.conn.post.return_value = request_fakes.fake_request_post(
None,
headers={
"Location": "https://localhost:8443/redfish/v1/"
"EthernetSwitches/Switch1/"
"Ports/Port1/VLANs/VLAN1"
},
)
self.vlan_col = vlan_network_interface.VLanNetworkInterfaceCollection(
self.conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs",
redfish_version="1.1.0",
)
def test__parse_attributes(self):
self.vlan_col._parse_attributes()
self.assertEqual("1.1.0", self.vlan_col.redfish_version)
self.assertEqual(
("/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1",),
self.vlan_col.members_identities,
)
@mock.patch.object(
vlan_network_interface, "VLanNetworkInterface", autospec=True
)
def test_get_member(self, mock_vlan):
self.vlan_col.get_member(
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1"
)
mock_vlan.assert_called_once_with(
self.vlan_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs/VLAN1",
redfish_version=self.vlan_col.redfish_version,
)
@mock.patch.object(
vlan_network_interface, "VLanNetworkInterface", autopspec=True
)
def test_get_members(self, mock_vlan):
members = self.vlan_col.get_members()
calls = [
mock.call(
self.vlan_col._conn,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/"
"VLANs/VLAN1",
redfish_version=self.vlan_col.redfish_version,
)
]
mock_vlan.assert_has_calls(calls)
self.assertIsInstance(members, list)
self.assertEqual(1, len(members))
def test_add_vlan_reqs(self):
reqs = {
"VLANId": 101,
"VLANEnable": True,
"Oem": {"Intel_RackScale": {"Tagged": False}},
}
result = self.vlan_col.add_vlan(reqs)
self.vlan_col._conn.post.assert_called_once_with(
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/VLANs", data=reqs
)
self.assertEqual(
result,
"/redfish/v1/EthernetSwitches/Switch1/Ports/Port1/" "VLANs/VLAN1",
)
def test_add_vlan_invalid_reqs(self):
reqs = {
"VLANId": 101,
"VLANEnable": True,
"Oem": {"Intel_RackScale": {"Tagged": False}},
}
# Missing filed
vlan_network_interface_req = reqs.copy()
vlan_network_interface_req.pop("VLANId")
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.vlan_col.add_vlan,
vlan_network_interface_req,
)
# Wrong format
vlan_network_interface_req = reqs.copy()
vlan_network_interface_req.update({"VLANId": "WrongFormat"})
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.vlan_col.add_vlan,
vlan_network_interface_req,
)
# Wrong additional fields
vlan_network_interface_req = reqs.copy()
vlan_network_interface_req["Additional"] = "AdditionalField"
self.assertRaises(
jsonschema.exceptions.ValidationError,
self.vlan_col.add_vlan,
vlan_network_interface_req,
)

View File

@ -19,7 +19,7 @@ import testtools
from sushy import exceptions
from rsd_lib.resources.v2_1.ethernet_switch import vlan
from rsd_lib.resources.v2_1.ethernet_switch import vlan_network_interface
from rsd_lib.resources.v2_1.system import ethernet_interface
@ -122,8 +122,9 @@ class EthernetInterface(testtools.TestCase):
# | WHEN |
actual_vlans = self.ethernet_interface_inst.vlans
# | THEN |
self.assertIsInstance(actual_vlans,
vlan.VLANCollection)
self.assertIsInstance(
actual_vlans,
vlan_network_interface.VLanNetworkInterfaceCollection)
self.conn.get.return_value.json.assert_called_once_with()
# reset mock
@ -140,8 +141,9 @@ class EthernetInterface(testtools.TestCase):
# | WHEN & THEN |
actual_vlans = self.ethernet_interface_inst.vlans
self.assertIsInstance(actual_vlans,
vlan.VLANCollection)
self.assertIsInstance(
actual_vlans,
vlan_network_interface.VLanNetworkInterfaceCollection)
# On refreshing...
with open('rsd_lib/tests/unit/json_samples/v2_1/'
@ -158,8 +160,9 @@ class EthernetInterface(testtools.TestCase):
self.conn.get.return_value.json.return_value = json.loads(
f.read())
# | WHEN & THEN |
self.assertIsInstance(actual_vlans,
vlan.VLANCollection)
self.assertIsInstance(
actual_vlans,
vlan_network_interface.VLanNetworkInterfaceCollection)
class EthernetInterfaceCollectionTestCase(testtools.TestCase):