Code Duplication - kytos-ng/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Code Duplication Length = 99-102 lines in 2 locations

tests/unit/test_main.py 2 locations


        response = await self.api_client.post(url, json=payload)
        assert 400 == response.status_code, response.data

    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
    async def test_create_a_circuit_case_5(
        self,
        validate_mock,
        evc_as_dict_mock,
        mongo_controller_upsert_mock,
        uni_from_dict_mock,
        sched_add_mock,
        evc_deploy_mock,
        mock_use_uni_tags,
        mock_tags_equal,
        mock_check_duplicate
    ):
        """Test create a new intra circuit with a disabled switch"""
        # pylint: disable=too-many-locals
        self.napp.controller.loop = asyncio.get_running_loop()
        validate_mock.return_value = True
        mongo_controller_upsert_mock.return_value = True
        evc_deploy_mock.return_value = True
        mock_use_uni_tags.return_value = True
        mock_tags_equal.return_value = True
        mock_check_duplicate.return_value = True
        uni1 = create_autospec(UNI)
        uni2 = create_autospec(UNI)
        uni1.interface = create_autospec(Interface)
        uni2.interface = create_autospec(Interface)
        switch = MagicMock()
        switch.status = EntityStatus.DISABLED
        switch.return_value = "00:00:00:00:00:00:00:01"
        uni1.interface.switch = switch
        uni2.interface.switch = switch
        uni_from_dict_mock.side_effect = [uni1, uni2]
        evc_as_dict_mock.return_value = {}
        sched_add_mock.return_value = True
        self.napp.mongo_controller.get_circuits.return_value = {}

        url = f"{self.base_endpoint}/v2/evc/"
        payload = {
            "name": "my evc1",
            "frequency": "* * * * *",
            "uni_a": {
                "interface_id": "00:00:00:00:00:00:00:01:1",
                "tag": {"tag_type": 'vlan', "value": 80},
            },
            "uni_z": {
                "interface_id": "00:00:00:00:00:00:00:01:2",
                "tag": {"tag_type": 'vlan', "value": 1},
            },
            "dynamic_backup_path": True,
            "primary_constraints": {
                "spf_max_path_cost": 8,
                "mandatory_metrics": {
                    "ownership": "red"
                }
            },
            "secondary_constraints": {
                "spf_attribute": "priority",
                "mandatory_metrics": {
                    "ownership": "blue"
                }
            }
        }

        response = await self.api_client.post(url, json=payload)
        current_data = response.json()

        # verify expected result from request
        assert 201 == response.status_code
        assert "circuit_id" in current_data

        # verify uni called
        uni_from_dict_mock.called_twice()
        uni_from_dict_mock.assert_any_call(payload["uni_z"])
        uni_from_dict_mock.assert_any_call(payload["uni_a"])

        # verify validation called
        validate_mock.assert_called_once()
        validate_mock.assert_called_with(
            table_group={'evpl': 0, 'epl': 0},
            frequency="* * * * *",
            name="my evc1",
            uni_a=uni1,
            uni_z=uni2,
            dynamic_backup_path=True,
            primary_constraints=payload["primary_constraints"],
            secondary_constraints=payload["secondary_constraints"],
        )
        # verify save method is called
        mongo_controller_upsert_mock.assert_called_once()

        # verify evc as dict is called to save in the box
        evc_as_dict_mock.assert_called()
        # verify add circuit in sched
        sched_add_mock.assert_called_once()

    async def test_create_a_circuit_invalid_queue_id(self):
        """Test create a new circuit with invalid queue_id."""

        expected_result = "circuit_id 3 not found"
        assert response.json()["description"] == expected_result

    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
    async def test_create_a_circuit_case_1(
        self,
        validate_mock,
        evc_as_dict_mock,
        mongo_controller_upsert_mock,
        uni_from_dict_mock,
        sched_add_mock,
        evc_deploy_mock,
        mock_use_uni_tags,
        mock_tags_equal,
        mock_check_duplicate
    ):
        """Test create a new circuit."""
        # pylint: disable=too-many-locals
        self.napp.controller.loop = asyncio.get_running_loop()
        validate_mock.return_value = True
        mongo_controller_upsert_mock.return_value = True
        evc_deploy_mock.return_value = True
        mock_use_uni_tags.return_value = True
        mock_tags_equal.return_value = True
        mock_check_duplicate.return_value = True
        uni1 = create_autospec(UNI)
        uni2 = create_autospec(UNI)
        uni1.interface = create_autospec(Interface)
        uni2.interface = create_autospec(Interface)
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
        uni_from_dict_mock.side_effect = [uni1, uni2]
        evc_as_dict_mock.return_value = {}
        sched_add_mock.return_value = True
        self.napp.mongo_controller.get_circuits.return_value = {}

        url = f"{self.base_endpoint}/v2/evc/"
        payload = {
            "name": "my evc1",
            "frequency": "* * * * *",
            "uni_a": {
                "interface_id": "00:00:00:00:00:00:00:01:1",
                "tag": {"tag_type": 'vlan', "value": 80},
            },
            "uni_z": {
                "interface_id": "00:00:00:00:00:00:00:02:2",
                "tag": {"tag_type": 'vlan', "value": 1},
            },
            "dynamic_backup_path": True,
            "primary_constraints": {
                "spf_max_path_cost": 8,
                "mandatory_metrics": {
                    "ownership": "red"
                }
            },
            "secondary_constraints": {
                "spf_attribute": "priority",
                "mandatory_metrics": {
                    "ownership": "blue"
                }
            }
        }

        response = await self.api_client.post(url, json=payload)
        current_data = response.json()

        # verify expected result from request
        assert 201 == response.status_code
        assert "circuit_id" in current_data

        # verify uni called
        uni_from_dict_mock.called_twice()
        uni_from_dict_mock.assert_any_call(payload["uni_z"])
        uni_from_dict_mock.assert_any_call(payload["uni_a"])

        # verify validation called
        validate_mock.assert_called_once()
        validate_mock.assert_called_with(
            table_group={'evpl': 0, 'epl': 0},
            frequency="* * * * *",
            name="my evc1",
            uni_a=uni1,
            uni_z=uni2,
            dynamic_backup_path=True,
            primary_constraints=payload["primary_constraints"],
            secondary_constraints=payload["secondary_constraints"],
        )
        # verify save method is called
        mongo_controller_upsert_mock.assert_called_once()

        # verify evc as dict is called to save in the box
        evc_as_dict_mock.assert_called()
        # verify add circuit in sched
        sched_add_mock.assert_called_once()

    async def test_create_a_circuit_case_2(self):
        """Test create a new circuit trying to send request without a json."""

		@@ 572-673 (lines=102) @@
569		response = await self.api_client.post(url, json=payload)
570		assert 400 == response.status_code, response.data
571
572		@patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
573		@patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
574		@patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
575		@patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
576		@patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
577		@patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
578		@patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
579		@patch("napps.kytos.mef_eline.main.EVC.as_dict")
580		@patch("napps.kytos.mef_eline.models.evc.EVC._validate")
581		async def test_create_a_circuit_case_5(
582		self,
583		validate_mock,
584		evc_as_dict_mock,
585		mongo_controller_upsert_mock,
586		uni_from_dict_mock,
587		sched_add_mock,
588		evc_deploy_mock,
589		mock_use_uni_tags,
590		mock_tags_equal,
591		mock_check_duplicate
592		):
593		"""Test create a new intra circuit with a disabled switch"""
594		# pylint: disable=too-many-locals
595		self.napp.controller.loop = asyncio.get_running_loop()
596		validate_mock.return_value = True
597		mongo_controller_upsert_mock.return_value = True
598		evc_deploy_mock.return_value = True
599		mock_use_uni_tags.return_value = True
600		mock_tags_equal.return_value = True
601		mock_check_duplicate.return_value = True
602		uni1 = create_autospec(UNI)
603		uni2 = create_autospec(UNI)
604		uni1.interface = create_autospec(Interface)
605		uni2.interface = create_autospec(Interface)
606		switch = MagicMock()
607		switch.status = EntityStatus.DISABLED
608		switch.return_value = "00:00:00:00:00:00:00:01"
609		uni1.interface.switch = switch
610		uni2.interface.switch = switch
611		uni_from_dict_mock.side_effect = [uni1, uni2]
612		evc_as_dict_mock.return_value = {}
613		sched_add_mock.return_value = True
614		self.napp.mongo_controller.get_circuits.return_value = {}
615
616		url = f"{self.base_endpoint}/v2/evc/"
617		payload = {
618		"name": "my evc1",
619		"frequency": "* * * * *",
620		"uni_a": {
621		"interface_id": "00:00:00:00:00:00:00:01:1",
622		"tag": {"tag_type": 'vlan', "value": 80},
623		},
624		"uni_z": {
625		"interface_id": "00:00:00:00:00:00:00:01:2",
626		"tag": {"tag_type": 'vlan', "value": 1},
627		},
628		"dynamic_backup_path": True,
629		"primary_constraints": {
630		"spf_max_path_cost": 8,
631		"mandatory_metrics": {
632		"ownership": "red"
633		}
634		},
635		"secondary_constraints": {
636		"spf_attribute": "priority",
637		"mandatory_metrics": {
638		"ownership": "blue"
639		}
640		}
641		}
642
643		response = await self.api_client.post(url, json=payload)
644		current_data = response.json()
645
646		# verify expected result from request
647		assert 201 == response.status_code
648		assert "circuit_id" in current_data
649
650		# verify uni called
651		uni_from_dict_mock.called_twice()
652		uni_from_dict_mock.assert_any_call(payload["uni_z"])
653		uni_from_dict_mock.assert_any_call(payload["uni_a"])
654
655		# verify validation called
656		validate_mock.assert_called_once()
657		validate_mock.assert_called_with(
658		table_group={'evpl': 0, 'epl': 0},
659		frequency="* * * * *",
660		name="my evc1",
661		uni_a=uni1,
662		uni_z=uni2,
663		dynamic_backup_path=True,
664		primary_constraints=payload["primary_constraints"],
665		secondary_constraints=payload["secondary_constraints"],
666		)
667		# verify save method is called
668		mongo_controller_upsert_mock.assert_called_once()
669
670		# verify evc as dict is called to save in the box
671		evc_as_dict_mock.assert_called()
672		# verify add circuit in sched
673		sched_add_mock.assert_called_once()
674
675		async def test_create_a_circuit_invalid_queue_id(self):
676		"""Test create a new circuit with invalid queue_id."""
		@@ 410-508 (lines=99) @@
407		expected_result = "circuit_id 3 not found"
408		assert response.json()["description"] == expected_result
409
410		@patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
411		@patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
412		@patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
413		@patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
414		@patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
415		@patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
416		@patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
417		@patch("napps.kytos.mef_eline.main.EVC.as_dict")
418		@patch("napps.kytos.mef_eline.models.evc.EVC._validate")
419		async def test_create_a_circuit_case_1(
420		self,
421		validate_mock,
422		evc_as_dict_mock,
423		mongo_controller_upsert_mock,
424		uni_from_dict_mock,
425		sched_add_mock,
426		evc_deploy_mock,
427		mock_use_uni_tags,
428		mock_tags_equal,
429		mock_check_duplicate
430		):
431		"""Test create a new circuit."""
432		# pylint: disable=too-many-locals
433		self.napp.controller.loop = asyncio.get_running_loop()
434		validate_mock.return_value = True
435		mongo_controller_upsert_mock.return_value = True
436		evc_deploy_mock.return_value = True
437		mock_use_uni_tags.return_value = True
438		mock_tags_equal.return_value = True
439		mock_check_duplicate.return_value = True
440		uni1 = create_autospec(UNI)
441		uni2 = create_autospec(UNI)
442		uni1.interface = create_autospec(Interface)
443		uni2.interface = create_autospec(Interface)
444		uni1.interface.switch = "00:00:00:00:00:00:00:01"
445		uni2.interface.switch = "00:00:00:00:00:00:00:02"
446		uni_from_dict_mock.side_effect = [uni1, uni2]
447		evc_as_dict_mock.return_value = {}
448		sched_add_mock.return_value = True
449		self.napp.mongo_controller.get_circuits.return_value = {}
450
451		url = f"{self.base_endpoint}/v2/evc/"
452		payload = {
453		"name": "my evc1",
454		"frequency": "* * * * *",
455		"uni_a": {
456		"interface_id": "00:00:00:00:00:00:00:01:1",
457		"tag": {"tag_type": 'vlan', "value": 80},
458		},
459		"uni_z": {
460		"interface_id": "00:00:00:00:00:00:00:02:2",
461		"tag": {"tag_type": 'vlan', "value": 1},
462		},
463		"dynamic_backup_path": True,
464		"primary_constraints": {
465		"spf_max_path_cost": 8,
466		"mandatory_metrics": {
467		"ownership": "red"
468		}
469		},
470		"secondary_constraints": {
471		"spf_attribute": "priority",
472		"mandatory_metrics": {
473		"ownership": "blue"
474		}
475		}
476		}
477
478		response = await self.api_client.post(url, json=payload)
479		current_data = response.json()
480
481		# verify expected result from request
482		assert 201 == response.status_code
483		assert "circuit_id" in current_data
484
485		# verify uni called
486		uni_from_dict_mock.called_twice()
487		uni_from_dict_mock.assert_any_call(payload["uni_z"])
488		uni_from_dict_mock.assert_any_call(payload["uni_a"])
489
490		# verify validation called
491		validate_mock.assert_called_once()
492		validate_mock.assert_called_with(
493		table_group={'evpl': 0, 'epl': 0},
494		frequency="* * * * *",
495		name="my evc1",
496		uni_a=uni1,
497		uni_z=uni2,
498		dynamic_backup_path=True,
499		primary_constraints=payload["primary_constraints"],
500		secondary_constraints=payload["secondary_constraints"],
501		)
502		# verify save method is called
503		mongo_controller_upsert_mock.assert_called_once()
504
505		# verify evc as dict is called to save in the box
506		evc_as_dict_mock.assert_called()
507		# verify add circuit in sched
508		sched_add_mock.assert_called_once()
509
510		async def test_create_a_circuit_case_2(self):
511		"""Test create a new circuit trying to send request without a json."""

kytos-ng / mef_eline

Code Duplication Length = 99-102 lines in 2 locations

tests/unit/test_main.py 2 locations