build.tests.unit.models.test_link_protection.TestLinkProtection.test_is_using_primary_path() - Code Metrics - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

TestLinkProtection.test_is_using_primary_path() A
last analyzed 2021-12-02 22:20 UTC

↳ Parent: build.tests.unit.models.test_link_protection

Complexity

Conditions

Size

Total Lines	21
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	16
nop	1
dl	0
loc	21
rs	9.6
c	0
b	0
f	0

"""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('requests.post')
    @patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
    @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)

        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,
                                               'enabled': True},
                                       'def': {'active': True,
                                               'enabled': 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_to_path')
    @patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
    @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,
                                     _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(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('requests.post')
92		@patch('napps.kytos.mef_eline.storehouse.StoreHouse.save_evc')
93		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
94		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
95		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
96		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
97		def test_deploy_to_case_2(self, install_uni_flows_mocked,
98		install_nni_flows_mocked,
99		deploy_mocked, *_):
100		"""Test deploy with all links up."""
101		deploy_mocked.return_value = True
102
103		primary_path = [
104		get_link_mocked(status=EntityStatus.UP),
105		get_link_mocked(status=EntityStatus.UP)
106		]
107		attributes = {
108		"controller": get_controller_mock(),
109		"name": "circuit_name",
110		"uni_a": get_uni_mocked(is_valid=True),
111		"uni_z": get_uni_mocked(is_valid=True),
112		"primary_path": primary_path,
113		"enabled": True
114		}
115		evc = EVC(**attributes)
116
117		deployed = evc.deploy_to('primary_path', evc.primary_path)
118		install_uni_flows_mocked.assert_called_with(evc.primary_path)
119		install_nni_flows_mocked.assert_called_with(evc.primary_path)
120		self.assertTrue(deployed)
121
122		# This method will be used by the mock to replace requests.get
123		def _mocked_requests_get_path_down(self):
124		# pylint: disable=no-self-use
125		return MockResponse({'links': {'abc': {'active': False,
126		'enabled': True},
127		'def': {'active': True,
128		'enabled': 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		@patch('napps.kytos.mef_eline.models.log')
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_to_path')
301		@patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
302		@patch(DEPLOY_TO_PRIMARY_PATH)
303		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
304		def test_handle_link_down_case_4(self, deploy_to_mocked,
305		_send_flow_mods_mocked,
306		deploy_mocked,
307		log_mocked):
308		"""Test if circuit with dynamic path is return success."""
309		deploy_mocked.return_value = True
310		deploy_to_mocked.return_value = False
311		primary_path = [
312		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
313		metadata={"s_vlan": 5},
314		status=EntityStatus.DOWN),
315		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
316		metadata={"s_vlan": 6},
317		status=EntityStatus.UP),
318		]
319		backup_path = [
320		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
321		metadata={"s_vlan": 5},
322		status=EntityStatus.DOWN),
323		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
324		metadata={"s_vlan": 6},
325		status=EntityStatus.UP),
326		]
327		attributes = {
328		"controller": get_controller_mock(),
329		"name": "circuit_name",
330		"uni_a": get_uni_mocked(is_valid=True),
331		"uni_z": get_uni_mocked(is_valid=True),
332		"primary_path": primary_path,
333		"backup_path": backup_path,
334		"enabled": True,
335		"dynamic_backup_path": True
336		}
337
338		evc = EVC(**attributes)
339		evc.current_path = evc.backup_path
340
341		# storehouse mock
342		evc._storehouse.box = Mock() # pylint: disable=protected-access
343		evc._storehouse.box.data = {} # pylint: disable=protected-access
344
345		current_handle_link_down = evc.handle_link_down()
346		self.assertEqual(deploy_to_mocked.call_count, 1)
347
348		self.assertTrue(current_handle_link_down)
349		msg = f"{evc} deployed after link down."
350		log_mocked.debug.assert_called_with(msg)
351
352		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
353		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
354		def test_handle_link_up_case_1(self, deploy_to_mocked, deploy_mocked):
355		"""Test if handle link up do nothing when is using primary path."""
356		deploy_mocked.return_value = True
357		deploy_to_mocked.return_value = True
358		primary_path = [
359		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
360		metadata={"s_vlan": 5},
361		status=EntityStatus.UP),
362		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
363		metadata={"s_vlan": 6},
364		status=EntityStatus.UP),
365		]
366		backup_path = [
367		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
368		metadata={"s_vlan": 5},
369		status=EntityStatus.UP),
370		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
371		metadata={"s_vlan": 6},
372		status=EntityStatus.UP),
373		]
374		attributes = {
375		"controller": get_controller_mock(),
376		"name": "circuit_name",
377		"uni_a": get_uni_mocked(is_valid=True),
378		"uni_z": get_uni_mocked(is_valid=True),
379		"primary_path": primary_path,
380		"backup_path": backup_path,
381		"enabled": True,
382		"dynamic_backup_path": True
383		}
384
385		evc = EVC(**attributes)
386		evc.current_path = evc.primary_path
387		current_handle_link_up = evc.handle_link_up(backup_path[0])
388		self.assertEqual(deploy_mocked.call_count, 0)
389		self.assertEqual(deploy_to_mocked.call_count, 0)
390		self.assertTrue(current_handle_link_up)
391
392		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
393		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
394		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
395		def test_handle_link_up_case_2(self, deploy_to_path_mocked, deploy_mocked):
396		"""Test if it is changing from backup_path to primary_path."""
397		deploy_mocked.return_value = True
398		deploy_to_path_mocked.return_value = True
399		primary_path = [
400		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
401		metadata={"s_vlan": 5},
402		status=EntityStatus.UP),
403		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
404		metadata={"s_vlan": 6},
405		status=EntityStatus.UP),
406		]
407		backup_path = [
408		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
409		metadata={"s_vlan": 5},
410		status=EntityStatus.UP),
411		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
412		metadata={"s_vlan": 6},
413		status=EntityStatus.UP),
414		]
415		attributes = {
416		"controller": get_controller_mock(),
417		"name": "circuit_name",
418		"uni_a": get_uni_mocked(is_valid=True),
419		"uni_z": get_uni_mocked(is_valid=True),
420		"primary_path": primary_path,
421		"backup_path": backup_path,
422		"enabled": True,
423		"dynamic_backup_path": True
424		}
425
426		evc = EVC(**attributes)
427		evc.current_path = evc.backup_path
428		current_handle_link_up = evc.handle_link_up(primary_path[0])
429		self.assertEqual(deploy_mocked.call_count, 0)
430		self.assertEqual(deploy_to_path_mocked.call_count, 1)
431		deploy_to_path_mocked.assert_called_once_with(evc.primary_path)
432		self.assertTrue(current_handle_link_up)
433
434		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
435		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
436		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
437		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
438		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
439		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
440		def test_handle_link_up_case_3(self, _install_uni_flows_mocked,
441		_install_nni_flows_mocked,
442		get_best_path_mocked,
443		deploy_to_path_mocked, deploy_mocked):
444		"""Test if it is deployed after the backup is up."""
445		deploy_mocked.return_value = True
446		deploy_to_path_mocked.return_value = True
447		primary_path = [
448		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
449		metadata={"s_vlan": 5},
450		status=EntityStatus.DOWN),
451		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
452		metadata={"s_vlan": 6},
453		status=EntityStatus.UP),
454		]
455		backup_path = [
456		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
457		metadata={"s_vlan": 5},
458		status=EntityStatus.DOWN),
459		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
460		metadata={"s_vlan": 6},
461		status=EntityStatus.UP),
462		]
463		attributes = {
464		"controller": get_controller_mock(),
465		"name": "circuit_name",
466		"uni_a": get_uni_mocked(is_valid=True),
467		"uni_z": get_uni_mocked(is_valid=True),
468		"primary_path": primary_path,
469		"backup_path": backup_path,
470		"enabled": True,
471		"dynamic_backup_path": True
472		}
473
474		evc = EVC(**attributes)
475
476		# storehouse initialization mock
477		evc._storehouse.box = Mock() # pylint: disable=protected-access
478		evc._storehouse.box.data = {} # pylint: disable=protected-access
479
480		evc.current_path = Path([])
481		current_handle_link_up = evc.handle_link_up(backup_path[0])
482
483		self.assertEqual(get_best_path_mocked.call_count, 0)
484		self.assertEqual(deploy_mocked.call_count, 0)
485		self.assertEqual(deploy_to_path_mocked.call_count, 1)
486		deploy_to_path_mocked.assert_called_once_with(evc.backup_path)
487		self.assertTrue(current_handle_link_up)
488
489		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy_to_path')
490		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
491		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
492		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
493		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
494		def test_handle_link_up_case_4(self, *args):
495		"""Test if not path is found a dynamic path is used."""
496		(_install_uni_flows_mocked, _install_nni_flows_mocked,
497		get_best_path_mocked, deploy_to_path_mocked) = args
498
499		deploy_to_path_mocked.return_value = True
500
501		primary_path = [
502		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
503		metadata={"s_vlan": 5},
504		status=EntityStatus.UP),
505		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
506		metadata={"s_vlan": 6},
507		status=EntityStatus.DOWN),
508		]
509		backup_path = [
510		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
511		metadata={"s_vlan": 5},
512		status=EntityStatus.DOWN),
513		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
514		metadata={"s_vlan": 6},
515		status=EntityStatus.DOWN),
516		]
517
518		# Setup best_path mock
519		best_path = Path()
520		best_path.append(primary_path[0])
521		get_best_path_mocked.return_value = best_path
522
523		attributes = {
524		"controller": get_controller_mock(),
525		"name": "circuit_name",
526		"uni_a": get_uni_mocked(is_valid=True),
527		"uni_z": get_uni_mocked(is_valid=True),
528		"primary_path": primary_path,
529		"backup_path": backup_path,
530		"enabled": True,
531		"dynamic_backup_path": True
532		}
533
534		evc = EVC(**attributes)
535		evc.current_path = Path([])
536
537		# storehouse mock
538		evc._storehouse.box = Mock() # pylint: disable=protected-access
539		evc._storehouse.box.data = {} # pylint: disable=protected-access
540
541		current_handle_link_up = evc.handle_link_up(backup_path[0])
542
543		self.assertEqual(get_best_path_mocked.call_count, 0)
544		self.assertEqual(deploy_to_path_mocked.call_count, 1)
545		deploy_to_path_mocked.assert_called_once_with()
546		self.assertTrue(current_handle_link_up)
547

kytos / mef_eline

TestLinkProtection.test_is_using_primary_path() A last analyzed 2021-12-02 22:20 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestLinkProtection.test_is_using_primary_path() A
last analyzed 2021-12-02 22:20 UTC