diff --git a/tests/unit_test/network/test_ut_network_fabricpc.py b/tests/unit_test/network/test_ut_network_fabricpc.py new file mode 100644 index 0000000..f54cd3e --- /dev/null +++ b/tests/unit_test/network/test_ut_network_fabricpc.py @@ -0,0 +1,170 @@ +import pytest +from ucsm_apis.network import fabricpc +from ucsmsdk.ucshandle import UcsHandle + +handle = UcsHandle("10.10.10.10", "username", "password") + + +def test_fabric_pc_create_success(mocker): + mock_login = mocker.patch('ucsmsdk.ucshandle.UcsHandle.login', + autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch('ucsmsdk.ucshandle.UcsHandle.query_dn', + autospec=True) + mock_query_dn.return_value = "fabric/lan/A" + mock_commit = mocker.patch('ucsmsdk.ucshandle.UcsHandle.commit', + autospec=True) + mock_commit.return_value = None + mock_fabric_pc = mocker.patch('ucsm_apis.network.' + 'fabricpc.FabricEthLanPc') + mock_fabric_pc_port = mocker.patch('ucsm_apis.network.' + 'fabricpc.FabricEthLanPcEp') + mock_add_mo = mocker.patch('ucsmsdk.ucshandle.UcsHandle.add_mo', + autospec=True) + mock_add_mo.return_value = None + + fabricpc.fabric_pc_create(handle, fabric="A", + pc_id="13", ports=["1/1", "1/2"]) + + mock_query_dn.assert_called_with(handle, "fabric/lan/A") + mock_fabric_pc.assert_called_with(admin_state="enabled", name=None, + parent_mo_or_dn="fabric/lan/A", + port_id="13") + assert mock_fabric_pc_port.call_count == 2 + + +def test_fabric_pc_create_fail_org_not_exist(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch.object(UcsHandle, 'query_dn', + autospec=True) + mock_query_dn.return_value = None + + with pytest.raises(fabricpc.UcsOperationError): + fabricpc.fabric_pc_create(handle, fabric="A", + pc_id="13") + + +def test_fabric_pc_get_success(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch.object(UcsHandle, 'query_dn', + autospec=True) + mock_query_dn.return_value = "fabric/lan/A/pc-13" + + fabricpc.fabric_pc_get(handle, fabric="A", + pc_id="13") + + mock_query_dn.assert_called_with(handle, "fabric/lan/A/pc-13") + + +def test_fabric_pc_get_fail_pc_does_not_exist(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch.object(UcsHandle, 'query_dn', + autospec=True) + mock_query_dn.return_value = None + + with pytest.raises(fabricpc.UcsOperationError): + fabricpc.fabric_pc_get(handle, fabric="A", + pc_id="1000") + + +def test_fabric_pc_exists_success(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch.object(UcsHandle, 'query_dn', + autospec=True) + mock_query_dn.return_value = "fabric/lan/A/pc-13" + mock_check_prop = mocker.patch.object(fabricpc.FabricEthLanPc, + 'check_prop_match', autospec=True) + mock_check_prop.return_value = True + mock_get = mocker.patch.object(fabricpc, 'fabric_pc_get', autospec=True) + + fabricpc.fabric_pc_exists(handle, fabric="A", + pc_id="13") + + mock_get.assert_called_with(handle=handle, caller='fabric_pc_exists', + fabric="A", pc_id="13") + + +def test_fabric_pc_exists_fail_org_does_not_exist(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_get = mocker.patch.object(fabricpc, 'fabric_pc_get', autospec=True) + mock_get.side_effect = fabricpc.UcsOperationError("query_dn", + "port channel" + "does not exist") + + result = fabricpc.fabric_pc_exists(handle, fabric="A", + pc_id="1000") + + assert result == (False, None) + + +def test_fabric_pc_modify_success(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_get = mocker.patch.object(fabricpc, 'fabric_pc_get', autospec=True) + mo_mock = mocker.Mock() + mo_mock.set_prop_multiple.return_value = True + mock_get.return_value = mo_mock + mock_set_mo = mocker.patch.object(UcsHandle, 'set_mo', autospec=True) + mock_set_mo.return_value = None + mock_commit = mocker.patch.object(UcsHandle, 'commit', autospec=True) + mock_commit.return_value = None + + fabricpc.fabric_pc_modify(handle, fabric="A", + pc_id="13", + descr="test") + + mock_get.assert_called_with(handle=handle, fabric="A", + pc_id="13", + caller="fabric_pc_modify") + mo_mock.set_prop_multiple.assert_called_with(descr="test") + + +def test_fabric_pc_modify_failure_pool_nonexistent(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_get = mocker.patch.object(fabricpc, 'fabric_pc_get', autospec=True) + mock_get.side_effect = fabricpc.UcsOperationError("query_dn", + "port channel" + "does not exist") + + with pytest.raises(fabricpc.UcsOperationError): + fabricpc.fabric_pc_modify(handle, fabric="A", + pc_id="1000", + descr="no aqui") + + +def test_fabric_pc_delete_success(mocker): + mock_login = mocker.patch('ucsmsdk.ucshandle.UcsHandle.login', + autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch('ucsmsdk.ucshandle.UcsHandle.query_dn', + autospec=True) + mock_query_dn.return_value = "fabric/lan/A" + mock_commit = mocker.patch('ucsmsdk.ucshandle.' + 'UcsHandle.commit', + autospec=True) + mock_commit.return_value = None + mock_remove_mo = mocker.patch.object(UcsHandle, 'remove_mo', autospec=True) + + fabricpc.fabric_pc_delete(handle, fabric="A", + pc_id="13") + + mock_query_dn.assert_called_with(handle, "fabric/lan/A/pc-13") + assert mock_remove_mo.call_count == 1 + + +def test_fabric_pc_delete_failure_org_nonexistent(mocker): + mock_login = mocker.patch.object(UcsHandle, 'login', autospec=True) + mock_login.return_value = True + mock_query_dn = mocker.patch.object(UcsHandle, 'query_dn', + autospec=True) + mock_query_dn.return_value = None + + with pytest.raises(fabricpc.UcsOperationError): + fabricpc.fabric_pc_delete(handle, fabric="A", + pc_id="1000") diff --git a/ucsm_apis/network/fabricpc.py b/ucsm_apis/network/fabricpc.py new file mode 100644 index 0000000..0ac520e --- /dev/null +++ b/ucsm_apis/network/fabricpc.py @@ -0,0 +1,216 @@ +# Copyright 2017 Cisco Systems, Inc. +# +# 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. + +""" +This module performs operations related to Fabric Port Channels. +""" + +from ucsmsdk.mometa.fabric.FabricEthLanPc import FabricEthLanPc +from ucsmsdk.mometa.fabric.FabricEthLanPcEp import FabricEthLanPcEp +from ucsmsdk.ucsexception import UcsOperationError + + +def fabric_pc_create(handle, fabric=None, + name=None, pc_id=None, + ports=None, + **kwargs): + + """ + Creates fabric portchannel + + Args: + handle (UCSHandle) + fabric (string): Fabric A or B + name (string): Portchannel name + pc_id (string): portchannel id number + ports (list): List of port members in the portchannel + **kwargs: Any additional key-value pair of managed object(MO)'s + property and value, which are not part of regular args. + This should be used for future version compatibility. + + Returns: + FabricEthLanPc: managed object + + Raises: + UcsOperationError: if Org_dn is not present + + Example: + fabric_pc_create(handle, + fabric="A", + name="vpc-nexus-a" + pc_id="10", + ports=["1/1", "1/2"] + """ + + org_dn = "fabric/lan/" + fabric + obj = handle.query_dn(org_dn) + if not obj: + raise UcsOperationError("fabric_pc_create", "Org {} \ + does not exist".format(org_dn)) + + mo = FabricEthLanPc(parent_mo_or_dn=obj, admin_state="enabled", + name=name, port_id=pc_id,) + mo.set_prop_multiple(**kwargs) + + port_list = [p.split("/") for p in ports] + mo_list = [] + for port in port_list: + mop = FabricEthLanPcEp(parent_mo_or_dn=mo, admin_state="enabled", + auto_negotiate="yes", + eth_link_profile_name="default", + slot_id=port[0], port_id=port[1]) + mo_list.append(mop) + + handle.add_mo(mo, modify_present=True) + handle.commit() + return mo + + +def fabric_pc_get(handle, fabric=None, + pc_id=None, + caller="fabric_pc_get"): + + """ + Gets fabric port channel + + Args: + handle (UCSHandle) + fabric (string): Fabric A or B + pc_id (string): portchannel id number + caller (string): caller method name + + Returns: + FabricEthLanPc: managed object + + Raises: + UcsOperationError: if FabricEthLanPc is not present + + Example: + fabric_pc_get(handle, + fabric="A", + pc_id="1") + """ + + dn = "fabric/lan/" + fabric + "/pc-" + pc_id + mo = handle.query_dn(dn) + if not mo: + raise UcsOperationError(caller, "Fabric port {} \ + does not exist".format(dn)) + return mo + + +def fabric_pc_exists(handle, fabric=None, pc_id=None, + **kwargs): + + """ + checks if fabric port channel exists + + Args: + handle (UCSHandle) + fabric (string): Fabric A or B + pc_id (string): portchannel id number + **kwargs: key-value pair of managed object(MO) property and value, Use + 'print(ucscoreutils.get_meta_info().config_props)' + to get all configurable properties of class + + Returns: + (True/False, FabricEthLanPc MO/None) + + Raises: + None + + Example: + fabric_pc_exists:(handle, + fabric="A", + pc_id="13") + """ + + try: + mo = fabric_pc_get(handle=handle, fabric=fabric, + pc_id=pc_id, + caller="fabric_pc_exists") + except UcsOperationError: + return (False, None) + mo_exists = mo.check_prop_match(**kwargs) + return (mo_exists, mo if mo_exists else None) + + +def fabric_pc_modify(handle, fabric=None, + pc_id=None, **kwargs): + + """ + modifies fabric port channel + + Args: + handle (UCSHandle) + fabric (string): Fabric A or B + pc_id (string): port channel id number + **kwargs: key-value pair of managed object(MO) property and value, Use + 'print(ucscoreutils.get_meta_info().config_props)' + to get all configurable properties of class + + Returns: + FabricEthLanPc: managed object + + Raises: + UcsOperationError: if FabricEthLanPc is not present + + Example: + fabric_pc_modify(handle, + fabric="A", + pc_id) + """ + + mo = fabric_pc_get(handle=handle, fabric=fabric, + pc_id=pc_id, + caller="fabric_pc_modify") + mo.set_prop_multiple(**kwargs) + handle.set_mo(mo) + handle.commit() + return mo + + +def fabric_pc_delete(handle, fabric=None, + pc_id=None, + **kwargs): + + """ + Deletes fabric port channel + + Args: + handle (UCSHandle) + fabric (string): Fabric A or B + pc_id (string): port channel id number + **kwargs: Any additional key-value pair of managed object(MO)'s + property and value, which are not part of regular args. + This should be used for future version compatibility. + + Returns: + None + + Raises: + UcsOperationError: if if FabricEthLanPc is not present + + Example: + fabric_pc_delete(handle, + fabric="A", + pc_id="13") + """ + + org_dn = "fabric/lan/" + fabric + "/pc-" + pc_id + mo = handle.query_dn(org_dn) + if not mo: + raise UcsOperationError("fabric_pc_delete", "Org {} \ + does not exist".format(org_dn)) + handle.remove_mo(mo) + handle.commit()