build.tests.unit.models.test_link_protection - Code Metrics - Inspection of "Create tags decorator to tests (#184)" - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1f710a...437722 )

by Humberto

created 2020-06-02 14:56 UTC

build.tests.unit.models.test_link_protection A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	551
Duplicated Lines	35.57 %

Importance

Changes

Metric	Value
eloc	437
dl	196
loc	551
rs	10
c	0
b	0
f	0
wmc	14

14 Methods

Rating	Name	Duplication	Size	Complexity
A	TestLinkProtection.test_deploy_to_case_1()	0	23	1
A	TestLinkProtection.test_is_using_backup_path()	0	21	1
A	TestLinkProtection.test_handle_link_up_case_2()	0	41	1
A	TestLinkProtection.test_is_using_primary_path()	0	21	1
A	TestLinkProtection.test_deploy_to_case_3()	0	25	1
B	TestLinkProtection.test_handle_link_down_case_3()	48	48	1
B	TestLinkProtection.test_handle_link_up_case_4()	0	58	1
B	TestLinkProtection.test_handle_link_down_case_1()	47	47	1
A	TestLinkProtection.test_deploy_to_case_2()	0	32	1
A	TestLinkProtection._mocked_requests_get_path_down()	0	4	1
B	TestLinkProtection.test_handle_link_down_case_2()	45	45	1
B	TestLinkProtection.test_handle_link_up_case_3()	0	54	1
B	TestLinkProtection.test_handle_link_down_case_4()	56	56	1
A	TestLinkProtection.test_handle_link_up_case_1()	0	39	1

How to fix Duplicated Code

"""Module to test the LinkProtection class."""
import sys
from unittest import TestCase
from unittest.mock import patch
from unittest.mock import Mock

from kytos.core.common import EntityStatus

# pylint: disable=wrong-import-position
sys.path.insert(0, '/var/lib/kytos/napps/..')
# pylint: enable=wrong-import-position
from napps.kytos.mef_eline.models import EVC, Path  # NOQA pycodestyle
from tests.helpers import MockResponse, get_link_mocked,\
    get_uni_mocked, get_controller_mock  # NOQA pycodestyle

DEPLOY_TO_PRIMARY_PATH = \
    'napps.kytos.mef_eline.models.LinkProtection.deploy_to_primary_path'


class TestLinkProtection(TestCase):  # pylint: disable=too-many-public-methods
    """Tests to validate LinkProtection class."""

    def test_is_using_backup_path(self):
        """Test test is using backup path."""
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5}),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6})
        ]

        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "backup_path": backup_path
        }

        evc = EVC(**attributes)
        self.assertFalse(evc.is_using_backup_path())
        evc.current_path = evc.backup_path
        self.assertTrue(evc.is_using_backup_path())

    def test_is_using_primary_path(self):
        """Test test is using primary path."""
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5}),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6})
        ]

        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path
        }

        evc = EVC(**attributes)
        self.assertFalse(evc.is_using_primary_path())
        evc.current_path = evc.primary_path
        self.assertTrue(evc.is_using_primary_path())

    @patch('napps.kytos.mef_eline.models.log')
    def test_deploy_to_case_1(self, log_mocked):
        """Test if the path is equal to current_path."""
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5}),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6})
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path
        }
        evc = EVC(**attributes)
        evc.current_path = evc.primary_path

        expected_deployed = evc.deploy_to('primary_path', evc.primary_path)
        expected_msg = 'primary_path is equal to current_path.'
        log_mocked.debug.assert_called_with(expected_msg)
        self.assertTrue(expected_deployed)

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
    @patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
    def test_deploy_to_case_2(self, install_uni_flows_mocked,
                              install_nni_flows_mocked,
                              deploy_mocked):
        """Test deploy with all links up."""
        deploy_mocked.return_value = True

        primary_path = [
                 get_link_mocked(status=EntityStatus.UP),
                 get_link_mocked(status=EntityStatus.UP)
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "enabled": True
        }
        evc = EVC(**attributes)

        # storehouse mock
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
        evc._storehouse.box.data = {}  # pylint: disable=protected-access

        deployed = evc.deploy_to('primary_path', evc.primary_path)
        install_uni_flows_mocked.assert_called_with(evc.primary_path)
        install_nni_flows_mocked.assert_called_with(evc.primary_path)
        self.assertTrue(deployed)

    # This method will be used by the mock to replace requests.get
    def _mocked_requests_get_path_down(self):
        # pylint: disable=no-self-use
        return MockResponse({'links': {'abc': {'active': False},
                                       'def': {'active': True}}}, 200)

    @patch('requests.get', side_effect=_mocked_requests_get_path_down)
    def test_deploy_to_case_3(self, requests_mocked):
        # pylint: disable=unused-argument
        """Test deploy with one link down."""
        link1 = get_link_mocked()
        link2 = get_link_mocked()
        link1.id = 'abc'
        link2.id = 'def'
        primary_path = [link1, link2]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "enabled": True
        }
        evc = EVC(**attributes)

        # storehouse initialization mock
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
        evc._storehouse.box.data = {}  # pylint: disable=protected-access

        deployed = evc.deploy_to('primary_path', evc.primary_path)
        self.assertFalse(deployed)

    @patch('napps.kytos.mef_eline.models.log')

    @patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
    @patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to_backup_path')
    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.Path.status')
    def test_handle_link_down_case_1(self, path_status_mocked,
                                     deploy_mocked, deploy_to_mocked,
                                     _send_flow_mods_mocked, log_mocked):
        """Test if deploy_to backup path is called."""
        deploy_mocked.return_value = True
        path_status_mocked.side_effect = [EntityStatus.DOWN, EntityStatus.UP]

        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True
        }
        evc = EVC(**attributes)

        evc.current_path = evc.primary_path
        current_handle_link_down = evc.handle_link_down()
        self.assertEqual(deploy_mocked.call_count, 0)
        deploy_to_mocked.assert_called_once()

        self.assertTrue(current_handle_link_down)
        msg = f"{evc} deployed after link down."
        log_mocked.debug.assert_called_once_with(msg)

    @patch('napps.kytos.mef_eline.models.log')

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch(DEPLOY_TO_PRIMARY_PATH)
    @patch('napps.kytos.mef_eline.models.Path.status')
    def test_handle_link_down_case_2(self, path_status_mocked,
                                     deploy_to_mocked, deploy_mocked,
                                     log_mocked):
        """Test if deploy_to backup path is called."""
        deploy_mocked.return_value = True
        deploy_to_mocked.return_value = True
        path_status_mocked.side_effect = [EntityStatus.UP, EntityStatus.DOWN]
        primary_path = [
                get_link_mocked(endpoint_a_port=7, endpoint_b_port=8,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True
        }

        evc = EVC(**attributes)
        evc.current_path = evc.backup_path
        current_handle_link_down = evc.handle_link_down()
        self.assertEqual(deploy_mocked.call_count, 0)
        deploy_to_mocked.assert_called_once()
        self.assertTrue(current_handle_link_down)
        msg = f"{evc} deployed after link down."
        log_mocked.debug.assert_called_once_with(msg)

    @patch('napps.kytos.mef_eline.models.log')

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch(DEPLOY_TO_PRIMARY_PATH)
    @patch('napps.kytos.mef_eline.models.DynamicPathManager.get_paths')
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
    def test_handle_link_down_case_3(self, get_paths_mocked,
                                     deploy_to_mocked, deploy_mocked,
                                     log_mocked):
        """Test if circuit without dynamic path is return failed."""
        deploy_mocked.return_value = False
        deploy_to_mocked.return_value = False
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True
        }

        evc = EVC(**attributes)
        evc.current_path = evc.backup_path
        current_handle_link_down = evc.handle_link_down()

        self.assertEqual(get_paths_mocked.call_count, 0)
        self.assertEqual(deploy_mocked.call_count, 0)
        self.assertEqual(deploy_to_mocked.call_count, 1)

        self.assertFalse(current_handle_link_down)
        msg = f'Failed to re-deploy {evc} after link down.'
        log_mocked.debug.assert_called_once_with(msg)

    @patch('napps.kytos.mef_eline.models.log')

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
    @patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
    @patch(DEPLOY_TO_PRIMARY_PATH)
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
    def test_handle_link_down_case_4(self, deploy_to_mocked,
                                     get_best_path_mocked,
                                     _send_flow_mods_mocked,
                                     deploy_mocked,
                                     log_mocked):
        """Test if circuit with dynamic path is return success."""
        deploy_mocked.return_value = True
        deploy_to_mocked.return_value = False
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True,
            "dynamic_backup_path": True
        }

        evc = EVC(**attributes)
        evc.current_path = evc.backup_path

        # storehouse mock
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
        evc._storehouse.box.data = {}  # pylint: disable=protected-access

        current_handle_link_down = evc.handle_link_down()

        self.assertEqual(get_best_path_mocked.call_count, 1)
        self.assertEqual(deploy_to_mocked.call_count, 1)

        self.assertTrue(current_handle_link_down)
        msg = f"{evc} deployed after link down."
        log_mocked.debug.assert_called_with(msg)

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    def test_handle_link_up_case_1(self, deploy_to_mocked, deploy_mocked):
        """Test if handle link up do nothing when is using primary path."""
        deploy_mocked.return_value = True
        deploy_to_mocked.return_value = True
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True,
            "dynamic_backup_path": True
        }

        evc = EVC(**attributes)
        evc.current_path = evc.primary_path
        current_handle_link_up = evc.handle_link_up(backup_path[0])
        self.assertEqual(deploy_mocked.call_count, 0)
        self.assertEqual(deploy_to_mocked.call_count, 0)
        self.assertTrue(current_handle_link_up)

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
    def test_handle_link_up_case_2(self, deploy_to_path_mocked, deploy_mocked):
        """Test if it is changing from backup_path to primary_path."""
        deploy_mocked.return_value = True
        deploy_to_path_mocked.return_value = True
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True,
            "dynamic_backup_path": True
        }

        evc = EVC(**attributes)
        evc.current_path = evc.backup_path
        current_handle_link_up = evc.handle_link_up(primary_path[0])
        self.assertEqual(deploy_mocked.call_count, 0)
        self.assertEqual(deploy_to_path_mocked.call_count, 1)
        deploy_to_path_mocked.assert_called_once_with(evc.primary_path)
        self.assertTrue(current_handle_link_up)

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
    @patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
    @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
    @patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
    def test_handle_link_up_case_3(self, _install_uni_flows_mocked,
                                   _install_nni_flows_mocked,
                                   get_best_path_mocked,
                                   deploy_to_path_mocked, deploy_mocked):
        """Test if it is deployed after the backup is up."""
        deploy_mocked.return_value = True
        deploy_to_path_mocked.return_value = True
        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]
        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True,
            "dynamic_backup_path": True
        }

        evc = EVC(**attributes)

        # storehouse initialization mock
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
        evc._storehouse.box.data = {}  # pylint: disable=protected-access

        evc.current_path = Path([])
        current_handle_link_up = evc.handle_link_up(backup_path[0])

        self.assertEqual(get_best_path_mocked.call_count, 0)
        self.assertEqual(deploy_mocked.call_count, 0)
        self.assertEqual(deploy_to_path_mocked.call_count, 1)
        deploy_to_path_mocked.assert_called_once_with(evc.backup_path)
        self.assertTrue(current_handle_link_up)

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
    @patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
    @patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
    @patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
    def test_handle_link_up_case_4(self, *args):
        """Test if not path is found a dynamic path is used."""
        (_install_uni_flows_mocked, _install_nni_flows_mocked,
         get_best_path_mocked, deploy_to_path_mocked) = args

        deploy_to_path_mocked.return_value = True

        primary_path = [
                get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.UP),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.DOWN),
        ]
        backup_path = [
                get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
                                metadata={"s_vlan": 5},
                                status=EntityStatus.DOWN),
                get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.DOWN),
        ]

        # Setup best_path mock
        best_path = Path()
        best_path.append(primary_path[0])
        get_best_path_mocked.return_value = best_path

        attributes = {
            "controller": get_controller_mock(),
            "name": "circuit_name",
            "uni_a": get_uni_mocked(is_valid=True),
            "uni_z": get_uni_mocked(is_valid=True),
            "primary_path": primary_path,
            "backup_path": backup_path,
            "enabled": True,
            "dynamic_backup_path": True
        }

        evc = EVC(**attributes)
        evc.current_path = Path([])

        # storehouse mock
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
        evc._storehouse.box.data = {}  # pylint: disable=protected-access

        current_handle_link_up = evc.handle_link_up(backup_path[0])

        self.assertEqual(get_best_path_mocked.call_count, 0)
        self.assertEqual(deploy_to_path_mocked.call_count, 1)
        deploy_to_path_mocked.assert_called_once_with()
        self.assertTrue(current_handle_link_up)


1		"""Module to test the LinkProtection class."""
2		import sys
3		from unittest import TestCase
4		from unittest.mock import patch
5		from unittest.mock import Mock
6
7		from kytos.core.common import EntityStatus
8
9		# pylint: disable=wrong-import-position
10		sys.path.insert(0, '/var/lib/kytos/napps/..')
11		# pylint: enable=wrong-import-position
12		from napps.kytos.mef_eline.models import EVC, Path # NOQA pycodestyle
13		from tests.helpers import MockResponse, get_link_mocked,\
14		get_uni_mocked, get_controller_mock # NOQA pycodestyle
15
16		DEPLOY_TO_PRIMARY_PATH = \
17		'napps.kytos.mef_eline.models.LinkProtection.deploy_to_primary_path'
18
19
20		class TestLinkProtection(TestCase): # pylint: disable=too-many-public-methods
21		"""Tests to validate LinkProtection class."""
22
23		def test_is_using_backup_path(self):
24		"""Test test is using backup path."""
25		backup_path = [
26		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
27		metadata={"s_vlan": 5}),
28		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
29		metadata={"s_vlan": 6})
30		]
31
32		attributes = {
33		"controller": get_controller_mock(),
34		"name": "circuit_name",
35		"uni_a": get_uni_mocked(is_valid=True),
36		"uni_z": get_uni_mocked(is_valid=True),
37		"backup_path": backup_path
38		}
39
40		evc = EVC(**attributes)
41		self.assertFalse(evc.is_using_backup_path())
42		evc.current_path = evc.backup_path
43		self.assertTrue(evc.is_using_backup_path())
44
45		def test_is_using_primary_path(self):
46		"""Test test is using primary path."""
47		primary_path = [
48		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
49		metadata={"s_vlan": 5}),
50		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
51		metadata={"s_vlan": 6})
52		]
53
54		attributes = {
55		"controller": get_controller_mock(),
56		"name": "circuit_name",
57		"uni_a": get_uni_mocked(is_valid=True),
58		"uni_z": get_uni_mocked(is_valid=True),
59		"primary_path": primary_path
60		}
61
62		evc = EVC(**attributes)
63		self.assertFalse(evc.is_using_primary_path())
64		evc.current_path = evc.primary_path
65		self.assertTrue(evc.is_using_primary_path())
66
67		@patch('napps.kytos.mef_eline.models.log')
68		def test_deploy_to_case_1(self, log_mocked):
69		"""Test if the path is equal to current_path."""
70		primary_path = [
71		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
72		metadata={"s_vlan": 5}),
73		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
74		metadata={"s_vlan": 6})
75		]
76		attributes = {
77		"controller": get_controller_mock(),
78		"name": "circuit_name",
79		"uni_a": get_uni_mocked(is_valid=True),
80		"uni_z": get_uni_mocked(is_valid=True),
81		"primary_path": primary_path
82		}
83		evc = EVC(**attributes)
84		evc.current_path = evc.primary_path
85
86		expected_deployed = evc.deploy_to('primary_path', evc.primary_path)
87		expected_msg = 'primary_path is equal to current_path.'
88		log_mocked.debug.assert_called_with(expected_msg)
89		self.assertTrue(expected_deployed)
90
91		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
92		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
93		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
94		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
95		def test_deploy_to_case_2(self, install_uni_flows_mocked,
96		install_nni_flows_mocked,
97		deploy_mocked):
98		"""Test deploy with all links up."""
99		deploy_mocked.return_value = True
100
101		primary_path = [
102		get_link_mocked(status=EntityStatus.UP),
103		get_link_mocked(status=EntityStatus.UP)
104		]
105		attributes = {
106		"controller": get_controller_mock(),
107		"name": "circuit_name",
108		"uni_a": get_uni_mocked(is_valid=True),
109		"uni_z": get_uni_mocked(is_valid=True),
110		"primary_path": primary_path,
111		"enabled": True
112		}
113		evc = EVC(**attributes)
114
115		# storehouse mock
116		evc._storehouse.box = Mock() # pylint: disable=protected-access
117		evc._storehouse.box.data = {} # pylint: disable=protected-access
118
119		deployed = evc.deploy_to('primary_path', evc.primary_path)
120		install_uni_flows_mocked.assert_called_with(evc.primary_path)
121		install_nni_flows_mocked.assert_called_with(evc.primary_path)
122		self.assertTrue(deployed)
123
124		# This method will be used by the mock to replace requests.get
125		def _mocked_requests_get_path_down(self):
126		# pylint: disable=no-self-use
127		return MockResponse({'links': {'abc': {'active': False},
128		'def': {'active': True}}}, 200)
129
130		@patch('requests.get', side_effect=_mocked_requests_get_path_down)
131		def test_deploy_to_case_3(self, requests_mocked):
132		# pylint: disable=unused-argument
133		"""Test deploy with one link down."""
134		link1 = get_link_mocked()
135		link2 = get_link_mocked()
136		link1.id = 'abc'
137		link2.id = 'def'
138		primary_path = [link1, link2]
139		attributes = {
140		"controller": get_controller_mock(),
141		"name": "circuit_name",
142		"uni_a": get_uni_mocked(is_valid=True),
143		"uni_z": get_uni_mocked(is_valid=True),
144		"primary_path": primary_path,
145		"enabled": True
146		}
147		evc = EVC(**attributes)
148
149		# storehouse initialization mock
150		evc._storehouse.box = Mock() # pylint: disable=protected-access
151		evc._storehouse.box.data = {} # pylint: disable=protected-access
152
153		deployed = evc.deploy_to('primary_path', evc.primary_path)
154		self.assertFalse(deployed)
155
156	View Code Duplication	@patch('napps.kytos.mef_eline.models.log')
		0 ignored issues – show Duplication introduced 2018-09-04 18:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
157		@patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
158		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to_backup_path')
159		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
160		@patch('napps.kytos.mef_eline.models.Path.status')
161		def test_handle_link_down_case_1(self, path_status_mocked,
162		deploy_mocked, deploy_to_mocked,
163		_send_flow_mods_mocked, log_mocked):
164		"""Test if deploy_to backup path is called."""
165		deploy_mocked.return_value = True
166		path_status_mocked.side_effect = [EntityStatus.DOWN, EntityStatus.UP]
167
168		primary_path = [
169		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
170		metadata={"s_vlan": 5},
171		status=EntityStatus.DOWN),
172		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
173		metadata={"s_vlan": 6},
174		status=EntityStatus.UP),
175		]
176		backup_path = [
177		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
178		metadata={"s_vlan": 5},
179		status=EntityStatus.UP),
180		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
181		metadata={"s_vlan": 6},
182		status=EntityStatus.UP),
183		]
184		attributes = {
185		"controller": get_controller_mock(),
186		"name": "circuit_name",
187		"uni_a": get_uni_mocked(is_valid=True),
188		"uni_z": get_uni_mocked(is_valid=True),
189		"primary_path": primary_path,
190		"backup_path": backup_path,
191		"enabled": True
192		}
193		evc = EVC(**attributes)
194
195		evc.current_path = evc.primary_path
196		current_handle_link_down = evc.handle_link_down()
197		self.assertEqual(deploy_mocked.call_count, 0)
198		deploy_to_mocked.assert_called_once()
199
200		self.assertTrue(current_handle_link_down)
201		msg = f"{evc} deployed after link down."
202		log_mocked.debug.assert_called_once_with(msg)
203
204	View Code Duplication	@patch('napps.kytos.mef_eline.models.log')
		0 ignored issues – show Duplication introduced 2018-08-30 17:36 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
205		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
206		@patch(DEPLOY_TO_PRIMARY_PATH)
207		@patch('napps.kytos.mef_eline.models.Path.status')
208		def test_handle_link_down_case_2(self, path_status_mocked,
209		deploy_to_mocked, deploy_mocked,
210		log_mocked):
211		"""Test if deploy_to backup path is called."""
212		deploy_mocked.return_value = True
213		deploy_to_mocked.return_value = True
214		path_status_mocked.side_effect = [EntityStatus.UP, EntityStatus.DOWN]
215		primary_path = [
216		get_link_mocked(endpoint_a_port=7, endpoint_b_port=8,
217		metadata={"s_vlan": 5},
218		status=EntityStatus.UP),
219		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
220		metadata={"s_vlan": 6},
221		status=EntityStatus.UP),
222		]
223		backup_path = [
224		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
225		metadata={"s_vlan": 5},
226		status=EntityStatus.DOWN),
227		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
228		metadata={"s_vlan": 6},
229		status=EntityStatus.UP),
230		]
231		attributes = {
232		"controller": get_controller_mock(),
233		"name": "circuit_name",
234		"uni_a": get_uni_mocked(is_valid=True),
235		"uni_z": get_uni_mocked(is_valid=True),
236		"primary_path": primary_path,
237		"backup_path": backup_path,
238		"enabled": True
239		}
240
241		evc = EVC(**attributes)
242		evc.current_path = evc.backup_path
243		current_handle_link_down = evc.handle_link_down()
244		self.assertEqual(deploy_mocked.call_count, 0)
245		deploy_to_mocked.assert_called_once()
246		self.assertTrue(current_handle_link_down)
247		msg = f"{evc} deployed after link down."
248		log_mocked.debug.assert_called_once_with(msg)
249
250	View Code Duplication	@patch('napps.kytos.mef_eline.models.log')
		0 ignored issues – show Duplication introduced 2018-09-04 18:23 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
251		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
252		@patch(DEPLOY_TO_PRIMARY_PATH)
253		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_paths')
254		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
255		def test_handle_link_down_case_3(self, get_paths_mocked,
256		deploy_to_mocked, deploy_mocked,
257		log_mocked):
258		"""Test if circuit without dynamic path is return failed."""
259		deploy_mocked.return_value = False
260		deploy_to_mocked.return_value = False
261		primary_path = [
262		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
263		metadata={"s_vlan": 5},
264		status=EntityStatus.DOWN),
265		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
266		metadata={"s_vlan": 6},
267		status=EntityStatus.UP),
268		]
269		backup_path = [
270		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
271		metadata={"s_vlan": 5},
272		status=EntityStatus.DOWN),
273		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
274		metadata={"s_vlan": 6},
275		status=EntityStatus.UP),
276		]
277		attributes = {
278		"controller": get_controller_mock(),
279		"name": "circuit_name",
280		"uni_a": get_uni_mocked(is_valid=True),
281		"uni_z": get_uni_mocked(is_valid=True),
282		"primary_path": primary_path,
283		"backup_path": backup_path,
284		"enabled": True
285		}
286
287		evc = EVC(**attributes)
288		evc.current_path = evc.backup_path
289		current_handle_link_down = evc.handle_link_down()
290
291		self.assertEqual(get_paths_mocked.call_count, 0)
292		self.assertEqual(deploy_mocked.call_count, 0)
293		self.assertEqual(deploy_to_mocked.call_count, 1)
294
295		self.assertFalse(current_handle_link_down)
296		msg = f'Failed to re-deploy {evc} after link down.'
297		log_mocked.debug.assert_called_once_with(msg)
298
299	View Code Duplication	@patch('napps.kytos.mef_eline.models.log')
		0 ignored issues – show Duplication introduced 2018-08-30 17:36 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
300		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
301		@patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
302		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
303		@patch(DEPLOY_TO_PRIMARY_PATH)
304		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
305		def test_handle_link_down_case_4(self, deploy_to_mocked,
306		get_best_path_mocked,
307		_send_flow_mods_mocked,
308		deploy_mocked,
309		log_mocked):
310		"""Test if circuit with dynamic path is return success."""
311		deploy_mocked.return_value = True
312		deploy_to_mocked.return_value = False
313		primary_path = [
314		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
315		metadata={"s_vlan": 5},
316		status=EntityStatus.DOWN),
317		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
318		metadata={"s_vlan": 6},
319		status=EntityStatus.UP),
320		]
321		backup_path = [
322		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
323		metadata={"s_vlan": 5},
324		status=EntityStatus.DOWN),
325		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
326		metadata={"s_vlan": 6},
327		status=EntityStatus.UP),
328		]
329		attributes = {
330		"controller": get_controller_mock(),
331		"name": "circuit_name",
332		"uni_a": get_uni_mocked(is_valid=True),
333		"uni_z": get_uni_mocked(is_valid=True),
334		"primary_path": primary_path,
335		"backup_path": backup_path,
336		"enabled": True,
337		"dynamic_backup_path": True
338		}
339
340		evc = EVC(**attributes)
341		evc.current_path = evc.backup_path
342
343		# storehouse mock
344		evc._storehouse.box = Mock() # pylint: disable=protected-access
345		evc._storehouse.box.data = {} # pylint: disable=protected-access
346
347		current_handle_link_down = evc.handle_link_down()
348
349		self.assertEqual(get_best_path_mocked.call_count, 1)
350		self.assertEqual(deploy_to_mocked.call_count, 1)
351
352		self.assertTrue(current_handle_link_down)
353		msg = f"{evc} deployed after link down."
354		log_mocked.debug.assert_called_with(msg)
355
356		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
357		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
358		def test_handle_link_up_case_1(self, deploy_to_mocked, deploy_mocked):
359		"""Test if handle link up do nothing when is using primary path."""
360		deploy_mocked.return_value = True
361		deploy_to_mocked.return_value = True
362		primary_path = [
363		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
364		metadata={"s_vlan": 5},
365		status=EntityStatus.UP),
366		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
367		metadata={"s_vlan": 6},
368		status=EntityStatus.UP),
369		]
370		backup_path = [
371		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
372		metadata={"s_vlan": 5},
373		status=EntityStatus.UP),
374		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
375		metadata={"s_vlan": 6},
376		status=EntityStatus.UP),
377		]
378		attributes = {
379		"controller": get_controller_mock(),
380		"name": "circuit_name",
381		"uni_a": get_uni_mocked(is_valid=True),
382		"uni_z": get_uni_mocked(is_valid=True),
383		"primary_path": primary_path,
384		"backup_path": backup_path,
385		"enabled": True,
386		"dynamic_backup_path": True
387		}
388
389		evc = EVC(**attributes)
390		evc.current_path = evc.primary_path
391		current_handle_link_up = evc.handle_link_up(backup_path[0])
392		self.assertEqual(deploy_mocked.call_count, 0)
393		self.assertEqual(deploy_to_mocked.call_count, 0)
394		self.assertTrue(current_handle_link_up)
395
396		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
397		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
398		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
399		def test_handle_link_up_case_2(self, deploy_to_path_mocked, deploy_mocked):
400		"""Test if it is changing from backup_path to primary_path."""
401		deploy_mocked.return_value = True
402		deploy_to_path_mocked.return_value = True
403		primary_path = [
404		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
405		metadata={"s_vlan": 5},
406		status=EntityStatus.UP),
407		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
408		metadata={"s_vlan": 6},
409		status=EntityStatus.UP),
410		]
411		backup_path = [
412		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
413		metadata={"s_vlan": 5},
414		status=EntityStatus.UP),
415		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
416		metadata={"s_vlan": 6},
417		status=EntityStatus.UP),
418		]
419		attributes = {
420		"controller": get_controller_mock(),
421		"name": "circuit_name",
422		"uni_a": get_uni_mocked(is_valid=True),
423		"uni_z": get_uni_mocked(is_valid=True),
424		"primary_path": primary_path,
425		"backup_path": backup_path,
426		"enabled": True,
427		"dynamic_backup_path": True
428		}
429
430		evc = EVC(**attributes)
431		evc.current_path = evc.backup_path
432		current_handle_link_up = evc.handle_link_up(primary_path[0])
433		self.assertEqual(deploy_mocked.call_count, 0)
434		self.assertEqual(deploy_to_path_mocked.call_count, 1)
435		deploy_to_path_mocked.assert_called_once_with(evc.primary_path)
436		self.assertTrue(current_handle_link_up)
437
438		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
439		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
440		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
441		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
442		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
443		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
444		def test_handle_link_up_case_3(self, _install_uni_flows_mocked,
445		_install_nni_flows_mocked,
446		get_best_path_mocked,
447		deploy_to_path_mocked, deploy_mocked):
448		"""Test if it is deployed after the backup is up."""
449		deploy_mocked.return_value = True
450		deploy_to_path_mocked.return_value = True
451		primary_path = [
452		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
453		metadata={"s_vlan": 5},
454		status=EntityStatus.DOWN),
455		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
456		metadata={"s_vlan": 6},
457		status=EntityStatus.UP),
458		]
459		backup_path = [
460		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
461		metadata={"s_vlan": 5},
462		status=EntityStatus.DOWN),
463		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
464		metadata={"s_vlan": 6},
465		status=EntityStatus.UP),
466		]
467		attributes = {
468		"controller": get_controller_mock(),
469		"name": "circuit_name",
470		"uni_a": get_uni_mocked(is_valid=True),
471		"uni_z": get_uni_mocked(is_valid=True),
472		"primary_path": primary_path,
473		"backup_path": backup_path,
474		"enabled": True,
475		"dynamic_backup_path": True
476		}
477
478		evc = EVC(**attributes)
479
480		# storehouse initialization mock
481		evc._storehouse.box = Mock() # pylint: disable=protected-access
482		evc._storehouse.box.data = {} # pylint: disable=protected-access
483
484		evc.current_path = Path([])
485		current_handle_link_up = evc.handle_link_up(backup_path[0])
486
487		self.assertEqual(get_best_path_mocked.call_count, 0)
488		self.assertEqual(deploy_mocked.call_count, 0)
489		self.assertEqual(deploy_to_path_mocked.call_count, 1)
490		deploy_to_path_mocked.assert_called_once_with(evc.backup_path)
491		self.assertTrue(current_handle_link_up)
492
493		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
494		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
495		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
496		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
497		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
498		def test_handle_link_up_case_4(self, *args):
499		"""Test if not path is found a dynamic path is used."""
500		(_install_uni_flows_mocked, _install_nni_flows_mocked,
501		get_best_path_mocked, deploy_to_path_mocked) = args
502
503		deploy_to_path_mocked.return_value = True
504
505		primary_path = [
506		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
507		metadata={"s_vlan": 5},
508		status=EntityStatus.UP),
509		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
510		metadata={"s_vlan": 6},
511		status=EntityStatus.DOWN),
512		]
513		backup_path = [
514		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
515		metadata={"s_vlan": 5},
516		status=EntityStatus.DOWN),
517		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
518		metadata={"s_vlan": 6},
519		status=EntityStatus.DOWN),
520		]
521
522		# Setup best_path mock
523		best_path = Path()
524		best_path.append(primary_path[0])
525		get_best_path_mocked.return_value = best_path
526
527		attributes = {
528		"controller": get_controller_mock(),
529		"name": "circuit_name",
530		"uni_a": get_uni_mocked(is_valid=True),
531		"uni_z": get_uni_mocked(is_valid=True),
532		"primary_path": primary_path,
533		"backup_path": backup_path,
534		"enabled": True,
535		"dynamic_backup_path": True
536		}
537
538		evc = EVC(**attributes)
539		evc.current_path = Path([])
540
541		# storehouse mock
542		evc._storehouse.box = Mock() # pylint: disable=protected-access
543		evc._storehouse.box.data = {} # pylint: disable=protected-access
544
545		current_handle_link_up = evc.handle_link_up(backup_path[0])
546
547		self.assertEqual(get_best_path_mocked.call_count, 0)
548		self.assertEqual(deploy_to_path_mocked.call_count, 1)
549		deploy_to_path_mocked.assert_called_once_with()
550		self.assertTrue(current_handle_link_up)
551

kytos / mef_eline

Push — master ( 1f710a...437722 )

build.tests.unit.models.test_link_protection A

Complexity

Size/Duplication

Importance

14 Methods

How to fix Duplicated Code

Duplicated Code

Duplication Side-by-Side

Filter issues like