build.tests.models.test_link_protection - Code Metrics - Inspection of "Update Scrutinizer, Tox configuration and unittest..." - kytos/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#112)

by Rogerio

created 2019-02-14 16:02 UTC

build.tests.models.test_link_protection A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	526
Duplicated Lines	32.7 %

Importance

Changes

Metric	Value
eloc	422
dl	172
loc	526
rs	10
c	0
b	0
f	0
wmc	14

14 Methods

Rating	Name	Duplication	Size	Complexity
B	TestLinkProtection.test_handle_link_down_case_4()	0	56	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_handle_link_up_case_1()	39	39	1
B	TestLinkProtection.test_handle_link_up_case_4()	0	59	1
A	TestLinkProtection.test_deploy_to_case_3()	0	18	1
B	TestLinkProtection.test_handle_link_down_case_3()	46	46	1
B	TestLinkProtection.test_handle_link_down_case_2()	43	43	1
A	TestLinkProtection.test_deploy_to_case_2()	0	32	1
B	TestLinkProtection.test_handle_link_up_case_3()	0	43	1
B	TestLinkProtection.test_handle_link_down_case_1()	44	44	1
A	TestLinkProtection.test_is_using_backup_path()	0	21	1
A	TestLinkProtection.test_deploy_to_case_1()	0	23	1
A	TestLinkProtection._mocked_requests_get_path_down()	0	2	1

How to fix Duplicated Code

"""Module to test the LinkProtection class."""
import sys
from tests.helpers import MockResponse
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
from tests.helpers import get_link_mocked,\
    get_uni_mocked, get_controller_mock  # NOQA



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()
        evc._storehouse.box.data = {}

        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(*args, **kwargs):
        return MockResponse({}, 200)

    @patch('requests.get', side_effect=_mocked_requests_get_path_down)
    def test_deploy_to_case_3(self, requests_get_path_down_mocked):
        """Test deploy with one link down."""
        primary_path = [
                 get_link_mocked(status=EntityStatus.DOWN),
                 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)
        self.assertFalse(deployed)


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

    @patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
    def test_handle_link_down_case_1(self, deploy_mocked, deploy_to_mocked,
                                     log_mocked):
        """Test if deploy_to backup path is called."""
        deploy_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=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_with('backup_path',
                                                 evc.backup_path)
        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('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    def test_handle_link_down_case_2(self, 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
        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.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_with('primary_path',
                                                 evc.primary_path)
        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('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    @patch('napps.kytos.mef_eline.models.DynamicPathManager.get_paths')
    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=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)
        self.assertEqual(deploy_to_mocked.call_count, 1)
        deploy_to_mocked.assert_called_once_with('primary_path',
                                                 evc.primary_path)
        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('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    @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,
                                     get_best_path_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=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

        #storehouse mock
        evc._storehouse.box = Mock()
        evc._storehouse.box.data = {}

        current_handle_link_down = evc.handle_link_down()
        #self.assertEqual(deploy_mocked.call_count, 1)
        self.assertEqual(deploy_to_mocked.call_count, 1)
        deploy_to_mocked.assert_called_once_with('primary_path',
                                                 evc.primary_path)
        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.LinkProtection.deploy_to')
    def test_handle_link_up_case_2(self, deploy_to_mocked, deploy_mocked):
        """Test if it is changing from backup_path to 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.backup_path
        current_handle_link_up = evc.handle_link_up(primary_path[0])
        self.assertEqual(deploy_mocked.call_count, 0)
        self.assertEqual(deploy_to_mocked.call_count, 1)
        deploy_to_mocked.assert_called_once_with('primary_path',
                                                 evc.primary_path)
        self.assertTrue(current_handle_link_up)

    @patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
    @patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    @patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
    def test_handle_link_up_case_3(self, get_best_path_mocked,
                                   deploy_to_mocked, deploy_mocked):
        """Test if it is deployed after the backup is up."""
        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.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)
        evc.current_path = 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, 1)
        deploy_to_mocked.assert_called_once_with('backup_path',
                                                 evc.backup_path)
        self.assertTrue(current_handle_link_up)

    @patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
    @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_mocked) = args
        """Test if it is deployed after the dynamic_backup_path deploy."""
        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=11, endpoint_b_port=12,
                                metadata={"s_vlan": 6},
                                status=EntityStatus.UP),
        ]

        # 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()
        evc._storehouse.box.data = {}

        current_handle_link_up = evc.handle_link_up(backup_path[0])

        self.assertEqual(get_best_path_mocked.call_count, 1)
        self.assertEqual(deploy_to_mocked.call_count, 1)
        deploy_to_mocked.assert_called_once_with('backup_path',
                                                 evc.backup_path)
        self.assertTrue(current_handle_link_up)


1		"""Module to test the LinkProtection class."""
2		import sys
3		from tests.helpers import MockResponse
4		from unittest import TestCase
5		from unittest.mock import patch
6		from unittest.mock import Mock
7
8		from kytos.core.common import EntityStatus
9
10		# pylint: disable=wrong-import-position
11		sys.path.insert(0, '/var/lib/kytos/napps/..')
12		# pylint: enable=wrong-import-position
13
14		from napps.kytos.mef_eline.models import EVC, Path # NOQA
15		from tests.helpers import get_link_mocked,\
16		get_uni_mocked, get_controller_mock # NOQA
17
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()
117		evc._storehouse.box.data = {}
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(args, *kwargs):
126		return MockResponse({}, 200)
127
128		@patch('requests.get', side_effect=_mocked_requests_get_path_down)
129		def test_deploy_to_case_3(self, requests_get_path_down_mocked):
130		"""Test deploy with one link down."""
131		primary_path = [
132		get_link_mocked(status=EntityStatus.DOWN),
133		get_link_mocked(status=EntityStatus.UP)
134		]
135		attributes = {
136		"controller": get_controller_mock(),
137		"name": "circuit_name",
138		"uni_a": get_uni_mocked(is_valid=True),
139		"uni_z": get_uni_mocked(is_valid=True),
140		"primary_path": primary_path,
141		"enabled": True
142		}
143		evc = EVC(**attributes)
144		deployed = evc.deploy_to('primary_path', evc.primary_path)
145		self.assertFalse(deployed)
146
147
148	View Code Duplication	@patch('napps.kytos.mef_eline.models.log')
		0 ignored issues – show Duplication introduced 2018-09-04 18:20 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
149		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
150		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
151		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.UP)
152		def test_handle_link_down_case_1(self, deploy_mocked, deploy_to_mocked,
153		log_mocked):
154		"""Test if deploy_to backup path is called."""
155		deploy_mocked.return_value = True
156
157		primary_path = [
158		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
159		metadata={"s_vlan": 5},
160		status=EntityStatus.DOWN),
161		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
162		metadata={"s_vlan": 6},
163		status=EntityStatus.UP),
164		]
165		backup_path = [
166		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
167		metadata={"s_vlan": 5},
168		status=EntityStatus.UP),
169		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
170		metadata={"s_vlan": 6},
171		status=EntityStatus.UP),
172		]
173		attributes = {
174		"controller": get_controller_mock(),
175		"name": "circuit_name",
176		"uni_a": get_uni_mocked(is_valid=True),
177		"uni_z": get_uni_mocked(is_valid=True),
178		"primary_path": primary_path,
179		"backup_path": backup_path,
180		"enabled": True
181		}
182		evc = EVC(**attributes)
183
184		evc.current_path = evc.primary_path
185		current_handle_link_down = evc.handle_link_down()
186		self.assertEqual(deploy_mocked.call_count, 0)
187		deploy_to_mocked.assert_called_once_with('backup_path',
188		evc.backup_path)
189		self.assertTrue(current_handle_link_down)
190		msg = f"{evc} deployed after link down."
191		log_mocked.debug.assert_called_once_with(msg)
192
193	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...
194		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
195		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
196		def test_handle_link_down_case_2(self, deploy_to_mocked, deploy_mocked,
197		log_mocked):
198		"""Test if deploy_to backup path is called."""
199		deploy_mocked.return_value = True
200		deploy_to_mocked.return_value = True
201		primary_path = [
202		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
203		metadata={"s_vlan": 5},
204		status=EntityStatus.UP),
205		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
206		metadata={"s_vlan": 6},
207		status=EntityStatus.UP),
208		]
209		backup_path = [
210		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
211		metadata={"s_vlan": 5},
212		status=EntityStatus.DOWN),
213		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
214		metadata={"s_vlan": 6},
215		status=EntityStatus.UP),
216		]
217		attributes = {
218		"controller": get_controller_mock(),
219		"name": "circuit_name",
220		"uni_a": get_uni_mocked(is_valid=True),
221		"uni_z": get_uni_mocked(is_valid=True),
222		"primary_path": primary_path,
223		"backup_path": backup_path,
224		"enabled": True
225		}
226
227		evc = EVC(**attributes)
228		evc.current_path = evc.backup_path
229		current_handle_link_down = evc.handle_link_down()
230		self.assertEqual(deploy_mocked.call_count, 0)
231		deploy_to_mocked.assert_called_once_with('primary_path',
232		evc.primary_path)
233		self.assertTrue(current_handle_link_down)
234		msg = f"{evc} deployed after link down."
235		log_mocked.debug.assert_called_once_with(msg)
236
237	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...
238		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
239		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
240		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_paths')
241		def test_handle_link_down_case_3(self, get_paths_mocked,
242		deploy_to_mocked, deploy_mocked,
243		log_mocked):
244		"""Test if circuit without dynamic path is return failed."""
245		deploy_mocked.return_value = False
246		deploy_to_mocked.return_value = False
247		primary_path = [
248		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
249		metadata={"s_vlan": 5},
250		status=EntityStatus.DOWN),
251		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
252		metadata={"s_vlan": 6},
253		status=EntityStatus.UP),
254		]
255		backup_path = [
256		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
257		metadata={"s_vlan": 5},
258		status=EntityStatus.DOWN),
259		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
260		metadata={"s_vlan": 6},
261		status=EntityStatus.UP),
262		]
263		attributes = {
264		"controller": get_controller_mock(),
265		"name": "circuit_name",
266		"uni_a": get_uni_mocked(is_valid=True),
267		"uni_z": get_uni_mocked(is_valid=True),
268		"primary_path": primary_path,
269		"backup_path": backup_path,
270		"enabled": True
271		}
272
273		evc = EVC(**attributes)
274		evc.current_path = evc.backup_path
275		current_handle_link_down = evc.handle_link_down()
276		self.assertEqual(deploy_mocked.call_count, 0)
277		self.assertEqual(deploy_to_mocked.call_count, 1)
278		deploy_to_mocked.assert_called_once_with('primary_path',
279		evc.primary_path)
280		self.assertFalse(current_handle_link_down)
281		msg = f'Failed to re-deploy {evc} after link down.'
282		log_mocked.debug.assert_called_once_with(msg)
283
284		@patch('napps.kytos.mef_eline.models.log')
285		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
286		@patch('napps.kytos.mef_eline.models.EVCDeploy._send_flow_mods')
287		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
288		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
289		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
290		def test_handle_link_down_case_4(self, deploy_to_mocked,
291		_send_flow_mods_mocked,
292		get_best_path_mocked,
293		deploy_mocked,
294		log_mocked):
295		"""Test if circuit with dynamic path is return success."""
296		deploy_mocked.return_value = True
297		deploy_to_mocked.return_value = False
298		primary_path = [
299		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
300		metadata={"s_vlan": 5},
301		status=EntityStatus.DOWN),
302		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
303		metadata={"s_vlan": 6},
304		status=EntityStatus.UP),
305		]
306		backup_path = [
307		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
308		metadata={"s_vlan": 5},
309		status=EntityStatus.DOWN),
310		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
311		metadata={"s_vlan": 6},
312		status=EntityStatus.UP),
313		]
314		attributes = {
315		"controller": get_controller_mock(),
316		"name": "circuit_name",
317		"uni_a": get_uni_mocked(is_valid=True),
318		"uni_z": get_uni_mocked(is_valid=True),
319		"primary_path": primary_path,
320		"backup_path": backup_path,
321		"enabled": True,
322		"dynamic_backup_path": True
323		}
324
325		evc = EVC(**attributes)
326		evc.current_path = evc.backup_path
327
328		#storehouse mock
329		evc._storehouse.box = Mock()
330		evc._storehouse.box.data = {}
331
332		current_handle_link_down = evc.handle_link_down()
333		#self.assertEqual(deploy_mocked.call_count, 1)
334		self.assertEqual(deploy_to_mocked.call_count, 1)
335		deploy_to_mocked.assert_called_once_with('primary_path',
336		evc.primary_path)
337		self.assertTrue(current_handle_link_down)
338		msg = f"{evc} deployed after link down."
339		log_mocked.debug.assert_called_with(msg)
340
341	View Code Duplication	@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
		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...
342		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
343		def test_handle_link_up_case_1(self, deploy_to_mocked, deploy_mocked):
344		"""Test if handle link up do nothing when is using primary path."""
345		deploy_mocked.return_value = True
346		deploy_to_mocked.return_value = True
347		primary_path = [
348		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
349		metadata={"s_vlan": 5},
350		status=EntityStatus.UP),
351		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
352		metadata={"s_vlan": 6},
353		status=EntityStatus.UP),
354		]
355		backup_path = [
356		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
357		metadata={"s_vlan": 5},
358		status=EntityStatus.UP),
359		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
360		metadata={"s_vlan": 6},
361		status=EntityStatus.UP),
362		]
363		attributes = {
364		"controller": get_controller_mock(),
365		"name": "circuit_name",
366		"uni_a": get_uni_mocked(is_valid=True),
367		"uni_z": get_uni_mocked(is_valid=True),
368		"primary_path": primary_path,
369		"backup_path": backup_path,
370		"enabled": True,
371		"dynamic_backup_path": True
372		}
373
374		evc = EVC(**attributes)
375		evc.current_path = evc.primary_path
376		current_handle_link_up = evc.handle_link_up(backup_path[0])
377		self.assertEqual(deploy_mocked.call_count, 0)
378		self.assertEqual(deploy_to_mocked.call_count, 0)
379		self.assertTrue(current_handle_link_up)
380
381		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
382		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
383		def test_handle_link_up_case_2(self, deploy_to_mocked, deploy_mocked):
384		"""Test if it is changing from backup_path to primary_path."""
385		deploy_mocked.return_value = True
386		deploy_to_mocked.return_value = True
387		primary_path = [
388		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
389		metadata={"s_vlan": 5},
390		status=EntityStatus.UP),
391		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
392		metadata={"s_vlan": 6},
393		status=EntityStatus.UP),
394		]
395		backup_path = [
396		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
397		metadata={"s_vlan": 5},
398		status=EntityStatus.UP),
399		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
400		metadata={"s_vlan": 6},
401		status=EntityStatus.UP),
402		]
403		attributes = {
404		"controller": get_controller_mock(),
405		"name": "circuit_name",
406		"uni_a": get_uni_mocked(is_valid=True),
407		"uni_z": get_uni_mocked(is_valid=True),
408		"primary_path": primary_path,
409		"backup_path": backup_path,
410		"enabled": True,
411		"dynamic_backup_path": True
412		}
413
414		evc = EVC(**attributes)
415		evc.current_path = evc.backup_path
416		current_handle_link_up = evc.handle_link_up(primary_path[0])
417		self.assertEqual(deploy_mocked.call_count, 0)
418		self.assertEqual(deploy_to_mocked.call_count, 1)
419		deploy_to_mocked.assert_called_once_with('primary_path',
420		evc.primary_path)
421		self.assertTrue(current_handle_link_up)
422
423		@patch('napps.kytos.mef_eline.models.EVCDeploy.deploy')
424		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
425		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
426		def test_handle_link_up_case_3(self, get_best_path_mocked,
427		deploy_to_mocked, deploy_mocked):
428		"""Test if it is deployed after the backup is up."""
429		deploy_mocked.return_value = True
430		deploy_to_mocked.return_value = True
431		primary_path = [
432		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
433		metadata={"s_vlan": 5},
434		status=EntityStatus.DOWN),
435		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
436		metadata={"s_vlan": 6},
437		status=EntityStatus.UP),
438		]
439		backup_path = [
440		get_link_mocked(endpoint_a_port=13, endpoint_b_port=14,
441		metadata={"s_vlan": 5},
442		status=EntityStatus.DOWN),
443		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
444		metadata={"s_vlan": 6},
445		status=EntityStatus.UP),
446		]
447		attributes = {
448		"controller": get_controller_mock(),
449		"name": "circuit_name",
450		"uni_a": get_uni_mocked(is_valid=True),
451		"uni_z": get_uni_mocked(is_valid=True),
452		"primary_path": primary_path,
453		"backup_path": backup_path,
454		"enabled": True,
455		"dynamic_backup_path": True
456		}
457
458		evc = EVC(**attributes)
459		evc.current_path = Path([])
460		current_handle_link_up = evc.handle_link_up(backup_path[0])
461		self.assertEqual(deploy_mocked.call_count, 0)
462		self.assertEqual(deploy_to_mocked.call_count, 1)
463		deploy_to_mocked.assert_called_once_with('backup_path',
464		evc.backup_path)
465		self.assertTrue(current_handle_link_up)
466
467		@patch('napps.kytos.mef_eline.models.LinkProtection.deploy_to')
468		@patch('napps.kytos.mef_eline.models.DynamicPathManager.get_best_path')
469		@patch('napps.kytos.mef_eline.models.EVC._install_nni_flows')
470		@patch('napps.kytos.mef_eline.models.EVC._install_uni_flows')
471		@patch('napps.kytos.mef_eline.models.Path.status', EntityStatus.DOWN)
472		def test_handle_link_up_case_4(self, *args):
473		"""Test if not path is found a dynamic path is used."""
474		(_install_uni_flows_mocked, _install_nni_flows_mocked,
475		get_best_path_mocked, deploy_to_mocked) = args
476		"""Test if it is deployed after the dynamic_backup_path deploy."""
477		deploy_to_mocked.return_value = False
478
479		primary_path = [
480		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
481		metadata={"s_vlan": 5},
482		status=EntityStatus.DOWN),
483		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
484		metadata={"s_vlan": 6},
485		status=EntityStatus.UP),
486		]
487		backup_path = [
488		get_link_mocked(endpoint_a_port=9, endpoint_b_port=10,
489		metadata={"s_vlan": 5},
490		status=EntityStatus.DOWN),
491		get_link_mocked(endpoint_a_port=11, endpoint_b_port=12,
492		metadata={"s_vlan": 6},
493		status=EntityStatus.UP),
494		]
495
496		# Setup best_path mock
497		best_path = Path()
498		best_path.append(primary_path[0])
499		get_best_path_mocked.return_value = best_path
500
501		attributes = {
502		"controller": get_controller_mock(),
503		"name": "circuit_name",
504		"uni_a": get_uni_mocked(is_valid=True),
505		"uni_z": get_uni_mocked(is_valid=True),
506		"primary_path": primary_path,
507		"backup_path": backup_path,
508		"enabled": True,
509		"dynamic_backup_path": True
510		}
511
512		evc = EVC(**attributes)
513		evc.current_path = Path([])
514
515		#storehouse mock
516		evc._storehouse.box = Mock()
517		evc._storehouse.box.data = {}
518
519		current_handle_link_up = evc.handle_link_up(backup_path[0])
520
521		self.assertEqual(get_best_path_mocked.call_count, 1)
522		self.assertEqual(deploy_to_mocked.call_count, 1)
523		deploy_to_mocked.assert_called_once_with('backup_path',
524		evc.backup_path)
525		self.assertTrue(current_handle_link_up)
526

kytos / mef_eline

Pull Request — master (#112)

build.tests.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