build.tests.unit.test_main.TestMain.setUpClass() - Code Metrics - Inspection of "Merge pull request #25 from kytos-ng/fix/revert_re..." - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 645373...00c642 )

by Vinicius

created 2022-08-11 13:59 UTC

build.tests.unit.test_main.TestMain.setUpClass() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	8
rs	10
c	0
b	0
f	0
cc	2
nop	1

"""Module to test the main napp file."""
import json
from unittest import TestCase
from unittest.mock import MagicMock, patch
from kytos.lib.helpers import (
    get_controller_mock,
    get_test_client,
    get_kytos_event_mock,
    get_switch_mock,
)
from napps.amlight.flow_stats.main import GenericFlow, Main
from napps.kytos.of_core.v0x01.flow import Action as Action10
from napps.kytos.of_core.v0x04.flow import Action as Action40
from napps.kytos.of_core.v0x04.match_fields import (
    MatchDLVLAN,
    MatchFieldFactory,
)
from pyof.foundation.basic_types import UBInt32
from pyof.v0x04.common.flow_instructions import InstructionType
from pyof.v0x01.controller2switch.common import StatsType
from pyof.v0x04.controller2switch.common import MultipartType


# pylint: disable=too-many-public-methods, too-many-lines
class TestMain(TestCase):
    """Test the Main class."""

    def setUp(self):
        """Execute steps before each tests.

        Set the server_name_url_url from amlight/flow_stats
        """
        self.server_name_url = "http://localhost:8181/api/amlight/flow_stats"
        self.napp = Main(get_controller_mock())

    @staticmethod
    def get_napp_urls(napp):
        """Return the amlight/flow_stats urls.

        The urls will be like:

        urls = [
            (options, methods, url)
        ]

        """
        controller = napp.controller
        controller.api_server.register_napp_endpoints(napp)

        urls = []
        for rule in controller.api_server.app.url_map.iter_rules():
            options = {}
            for arg in rule.arguments:
                options[arg] = f"[{0}]".format(arg)

            if f"{napp.username}/{napp.name}" in str(rule):
                urls.append((options, rule.methods, f"{str(rule)}"))

        return urls

    def test_verify_api_urls(self):
        """Verify all APIs registered."""

        expected_urls = [
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/flow/match/ ",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/flow/stats/",
            ),
            (
                {"flow_id": "[flow_id]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/packet_count/",
            ),
            (
                {"flow_id": "[flow_id]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/bytes_count/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/packet_count/per_flow/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/packet_count/sum/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/bytes_count/per_flow/",
            ),
            (
                {"dpid": "[dpid]"},
                {"OPTIONS", "HEAD", "GET"},
                "/api/amlight/flow_stats/bytes_count/sum/",
            ),
        ]
        urls = self.get_napp_urls(self.napp)
        self.assertEqual(len(expected_urls), len(urls))

    def test_packet_count__fail(self):
        """Test packet_count rest call with wrong flow_id."""
        flow_id = "123456789"
        rest_name = "packet_count"
        response = self._get_rest_response(rest_name, flow_id)

        self.assertEqual(response.data, b"Flow does not exist")

    def test_packet_count(self):
        """Test packet_count rest call."""
        flow_id = "1"
        rest_name = "packet_count"
        self._patch_switch_flow(flow_id)
        response = self._get_rest_response(rest_name, flow_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response["flow_id"], flow_id)
        self.assertEqual(json_response["packet_counter"], 40)
        self.assertEqual(json_response["packet_per_second"], 2.0)

    def test_bytes_count_fail(self):
        """Test bytes_count rest call with wrong flow_id."""
        flow_id = "123456789"
        rest_name = "bytes_count"
        response = self._get_rest_response(rest_name, flow_id)

        self.assertEqual(response.data, b"Flow does not exist")

    def test_bytes_count(self):
        """Test bytes_count rest call."""
        flow_id = "1"
        rest_name = "bytes_count"
        self._patch_switch_flow(flow_id)
        response = self._get_rest_response(rest_name, flow_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response["flow_id"], flow_id)
        self.assertEqual(json_response["bytes_counter"], 10)
        self.assertEqual(json_response["bits_per_second"], 4.0)

    def test_packet_count_per_flow(self):

        """Test packet_count_per_flow rest call."""
        flow_id = "1"
        rest_name = "packet_count/per_flow"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response[0]["flow_id"], flow_id)
        self.assertEqual(json_response[0]["packet_counter"], 40)
        self.assertEqual(json_response[0]["packet_per_second"], 2.0)

    def test_packet_count_sum(self):
        """Test packet_count_sum rest call."""
        flow_id = "1"
        rest_name = "packet_count/sum"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(json_response, 40)

    def test_bytes_count_per_flow(self):

        """Test bytes_count_per_flow rest call."""
        flow_id = "1"
        rest_name = "bytes_count/per_flow"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)

        json_response = json.loads(response.data)
        self.assertEqual(json_response[0]["flow_id"], flow_id)
        self.assertEqual(json_response[0]["bytes_counter"], 10)
        self.assertEqual(json_response[0]["bits_per_second"], 4.0)

    def test_bytes_count_sum(self):
        """Test bytes_count_sum rest call."""
        flow_id = "1"
        rest_name = "bytes_count/sum"
        self._patch_switch_flow(flow_id)

        dpid_id = 111
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(json_response, 10)

    @patch("napps.amlight.flow_stats.main.Main.match_flows")
    def test_flow_match(self, mock_match_flows):
        """Test flow_match rest call."""
        flow = GenericFlow()
        flow.actions = [
            Action10.from_dict(
                {
                    "action_type": "output",
                    "port": "1",
                }
            ),
        ]
        flow.version = "0x04"
        mock_match_flows.return_value = flow

        flow_id = "1"
        rest_name = "flow/match"
        self._patch_switch_flow(flow_id)

        dpid_id = "aa:00:00:00:00:00:00:11"
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(json_response["actions"][0]["action_type"], "output")
        self.assertEqual(json_response["actions"][0]["port"], "1")
        self.assertEqual(json_response["version"], "0x04")

    @patch("napps.amlight.flow_stats.main.Main.match_flows")
    def test_flow_match_fail(self, mock_match_flows):
        """Test flow_match rest call."""
        mock_match_flows.return_value = None

        flow_id = "1"
        rest_name = "flow/match"
        self._patch_switch_flow(flow_id)

        dpid_id = "aa:00:00:00:00:00:00:11"
        response = self._get_rest_response(rest_name, dpid_id)

        self.assertEqual(response.status_code, 404)

    @patch("napps.amlight.flow_stats.main.Main.match_flows")
    def test_flow_stats(self, mock_match_flows):
        """Test flow_match rest call."""
        flow = GenericFlow()
        flow.actions = [
            Action10.from_dict(
                {
                    "action_type": "output",
                    "port": "1",
                }
            ),
        ]
        flow.version = "0x04"
        mock_match_flows.return_value = [flow]

        flow_id = "1"
        rest_name = "flow/stats"
        self._patch_switch_flow(flow_id)

        dpid_id = "aa:00:00:00:00:00:00:11"
        response = self._get_rest_response(rest_name, dpid_id)
        json_response = json.loads(response.data)

        self.assertEqual(response.status_code, 200)
        print(json_response)
        self.assertEqual(json_response[0]["actions"][0]["action_type"],
                         "output")
        self.assertEqual(json_response[0]["actions"][0]["port"], "1")
        self.assertEqual(json_response[0]["version"], "0x04")

    def _patch_switch_flow(self, flow_id):
        """Helper method to patch controller to return switch/flow data."""
        # patching the flow_stats object in the switch
        self.napp.controller.switches = MagicMock()
        flow = self._get_mocked_flow_stats()
        flow.id = flow_id
        switch = MagicMock()
        switch.generic_flows = [flow]
        self.napp.controller.switches.values.return_value = [switch]
        self.napp.controller.get_switch_by_dpid = MagicMock()
        self.napp.controller.get_switch_by_dpid.return_value = switch

    def _get_rest_response(self, rest_name, url_id):
        """Helper method to call a rest endpoint."""
        # call rest
        api = get_test_client(get_controller_mock(), self.napp)
        url = f"{self.server_name_url}/{rest_name}/{url_id}"
        response = api.get(url, content_type="application/json")

        return response

    # pylint: disable=no-self-use
    def _get_mocked_flow_stats(self):
        """Helper method to create a mock flow_stats object."""
        flow_stats = MagicMock()
        flow_stats.id = 123
        flow_stats.byte_count = 10
        flow_stats.duration_sec = 20
        flow_stats.duration_nsec = 30
        flow_stats.packet_count = 40
        return flow_stats

    def _get_mocked_flow_base(self):
        """Helper method to create a mock flow object."""
        flow = MagicMock()
        flow.id = 456
        flow.switch = None
        flow.table_id = None
        flow.match = None
        flow.priority = None
        flow.idle_timeout = None
        flow.hard_timeout = None
        flow.cookie = None
        flow.stats = self._get_mocked_flow_stats()
        return flow

    @patch("napps.amlight.flow_stats.main.GenericFlow.from_flow_stats")
    def test_handle_stats_reply(self, mock_from_flow):
        """Test handle_stats_reply rest call."""
        mock_from_flow.return_value = self._get_mocked_flow_base()

        def side_effect(event):
            self.assertTrue(f"{event}", "amlight/flow_stats.flows_updated")
            self.assertTrue(event.content["switch"], 111)

        self.napp.controller = MagicMock()
        self.napp.controller.buffers.app.put.side_effect = side_effect

        msg = MagicMock()
        msg.flags.value = 2
        msg.body = [self._get_mocked_flow_stats()]
        event_switch = MagicMock()
        event_switch.generic_flows = []
        event_switch.dpid = 111
        self.napp.handle_stats_reply(msg, event_switch)

        # Check if important trace dont trigger the event
        # It means that the CP trace is the same to the DP trace
        self.napp.controller.buffers.app.put.assert_called_once()

        # Check mocked flow id
        self.assertEqual(event_switch.generic_flows[0].id, 456)

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
    def test_handle_stats_reply_0x01(self, mock_handle_stats):
        """Test handle stats reply."""
        flow_msg = MagicMock()
        flow_msg.body = "A"
        flow_msg.body_type = StatsType.OFPST_FLOW

        switch_v0x01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x01)

        name = "kytos/of_core.v0x01.messages.in.ofpt_stats_reply"
        content = {"source": switch_v0x01.connection, "message": flow_msg}
        event = get_kytos_event_mock(name=name, content=content)

        self.napp.handle_stats_reply_0x01(event)
        mock_handle_stats.assert_called_once()

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
    def test_handle_stats_reply_0x01__fail(self, mock_handle_stats):
        """Test handle stats reply."""
        flow_msg = MagicMock()
        flow_msg.body = "A"
        flow_msg.body_type = StatsType.OFPST_DESC

        switch_v0x01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x01)

        name = "kytos/of_core.v0x01.messages.in.ofpt_stats_reply"
        content = {"source": switch_v0x01.connection, "message": flow_msg}
        event = get_kytos_event_mock(name=name, content=content)

        self.napp.handle_stats_reply_0x01(event)

        mock_handle_stats.assert_not_called()

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
    def test_handle_stats_reply_0x04(self, mock_handle_stats):
        """Test handle stats reply."""
        flow_msg = MagicMock()
        flow_msg.body = "A"
        flow_msg.multipart_type = MultipartType.OFPMP_FLOW

        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)

        name = "kytos/of_core.v0x04.messages.in.ofpt_multipart_reply"
        content = {"source": switch_v0x04.connection, "message": flow_msg}
        event = get_kytos_event_mock(name=name, content=content)
        event.content["message"] = flow_msg

        self.napp.handle_stats_reply_0x04(event)

        mock_handle_stats.assert_called_once()

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
    def test_handle_stats_reply_0x04_fail(self, mock_handle_stats):
        """Test handle stats reply."""
        flow_msg = MagicMock()
        flow_msg.body = "A"

        flow_msg.multipart_type = MultipartType.OFPMP_PORT_DESC

        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)

        name = "kytos/of_core.v0x04.messages.in.ofpt_multipart_reply"
        content = {"source": switch_v0x04.connection, "message": flow_msg}

        event = get_kytos_event_mock(name=name, content=content)
        event.content["message"] = flow_msg

        self.napp.handle_stats_reply_0x04(event)

        mock_handle_stats.assert_not_called()


# pylint: disable=too-many-public-methods, too-many-lines
class TestGenericFlow(TestCase):
    """Test the GenericFlow class."""

    # pylint: disable=no-member
    def test_from_flow_stats__x01(self):
        """Test from_flow_stats method 0x01 version."""
        flow_stats = MagicMock()

        flow_stats.actions = [
            Action10.from_dict(
                {
                    "action_type": "output",
                    "port": UBInt32(1),
                }
            ).as_of_action(),
        ]

        result = GenericFlow.from_flow_stats(flow_stats, version="0x01")

        self.assertEqual(result.idle_timeout, flow_stats.idle_timeout.value)
        self.assertEqual(result.hard_timeout, flow_stats.hard_timeout.value)
        self.assertEqual(result.priority, flow_stats.priority.value)
        self.assertEqual(result.table_id, flow_stats.table_id.value)
        self.assertEqual(result.duration_sec, flow_stats.duration_sec.value)
        self.assertEqual(result.packet_count, flow_stats.packet_count.value)
        self.assertEqual(result.byte_count, flow_stats.byte_count.value)

        exp_match = flow_stats.match
        self.assertEqual(result.match["wildcards"], exp_match.wildcards.value)
        self.assertEqual(result.match["in_port"], exp_match.in_port.value)
        self.assertEqual(result.match["eth_src"], exp_match.dl_src.value)
        self.assertEqual(result.match["eth_dst"], exp_match.dl_dst.value)
        self.assertEqual(result.match["vlan_vid"], exp_match.dl_vlan.value)
        self.assertEqual(result.match["vlan_pcp"], exp_match.dl_vlan_pcp.value)
        self.assertEqual(result.match["eth_type"], exp_match.dl_type.value)
        self.assertEqual(result.match["ip_tos"], exp_match.nw_tos.value)
        self.assertEqual(result.match["ipv4_src"], exp_match.nw_src.value)
        self.assertEqual(result.match["ipv4_dst"], exp_match.nw_dst.value)
        self.assertEqual(result.match["ip_proto"], exp_match.nw_proto.value)
        self.assertEqual(result.match["tcp_src"], exp_match.tp_src.value)
        self.assertEqual(result.match["tcp_dst"], exp_match.tp_dst.value)

    def test_from_flow_stats__x04_match(self):
        """Test from_flow_stats method 0x04 version with match."""
        flow_stats = MagicMock()
        flow_stats.actions = [
            Action10.from_dict(
                {
                    "action_type": "output",
                    "port": UBInt32(1),
                }
            ).as_of_action(),
        ]

        flow_stats.match.oxm_match_fields = [MatchDLVLAN(42).as_of_tlv()]

        result = GenericFlow.from_flow_stats(flow_stats, version="0x04")

        match_expect = MatchFieldFactory.from_of_tlv(
            flow_stats.match.oxm_match_fields[0]
        )
        self.assertEqual(result.match["vlan_vid"], match_expect)

    def test_from_flow_stats__x04_action(self):
        """Test from_flow_stats method 0x04 version with action."""
        flow_stats = MagicMock()

        action_dict = {
            "action_type": "output",
            "port": UBInt32(1),
        }
        instruction = MagicMock()
        instruction.instruction_type = InstructionType.OFPIT_APPLY_ACTIONS
        instruction.actions = [Action40.from_dict(action_dict).as_of_action()]
        flow_stats.instructions = [instruction]

        result = GenericFlow.from_flow_stats(flow_stats, version="0x04")
        self.assertEqual(result.actions[0].as_dict(), action_dict)

    def test_to_dict__x01(self):
        """Test to_dict method 0x01 version."""
        action = Action10.from_dict(
            {
                "action_type": "set_vlan",
                "vlan_id": 6,
            }
        )
        match = {}
        match["in_port"] = 11

        generic_flow = GenericFlow(
            version="0x01",
            match=match,
            idle_timeout=1,
            hard_timeout=2,
            duration_sec=3,
            packet_count=4,
            byte_count=5,
            priority=6,
            table_id=7,
            cookie=8,
            buffer_id=9,
            actions=[action],
        )

        result = generic_flow.to_dict()

        expected = {
            "version": "0x01",
            "idle_timeout": 1,
            "in_port": 11,
            "hard_timeout": 2,
            "priority": 6,
            "table_id": 7,
            "cookie": 8,
            "buffer_id": 9,
            "actions": [{"vlan_id": 6, "action_type": "set_vlan"}],
        }

        self.assertEqual(result, expected)

    def test_to_dict__x04(self):
        """Test to_dict method 0x04 version."""
        match = {}
        match["in_port"] = MagicMock()
        match["in_port"].value = 22

        generic_flow = GenericFlow(
            version="0x04",
            match=match,
        )

        result = generic_flow.to_dict()
        expected = {
            "version": "0x04",
            "in_port": 22,
            "idle_timeout": 0,
            "hard_timeout": 0,
            "priority": 0,
            "table_id": 255,
            "cookie": None,
            "buffer_id": None,
            "actions": [],
        }

        self.assertEqual(result, expected)

    def test_match_to_dict(self):
        """Test match_to_dict method for 0x04 version."""
        match = {}
        match["in_port"] = MagicMock()
        match["in_port"].value = 22
        match["vlan_vid"] = MagicMock()
        match["vlan_vid"].value = 123

        generic_flow = GenericFlow(
            version="0x04",
            match=match,
        )
        result = generic_flow.match_to_dict()
        expected = {"in_port": 22, "vlan_vid": 123}

        self.assertEqual(result, expected)

    def test_match_to_dict__empty_match(self):
        """Test match_to_dict method for 0x04 version, with empty matches."""
        generic_flow = GenericFlow(version="0x04")
        result = generic_flow.match_to_dict()
        self.assertEqual(result, {})

    def test_id__x01(self):
        """Test id method 0x01 version."""
        action = Action10.from_dict(
            {
                "action_type": "set_vlan",
                "vlan_id": 6,
            }
        )
        match = {}
        match["in_port"] = 11

        generic_flow = GenericFlow(
            version="0x01",
            match=match,
            idle_timeout=1,
            hard_timeout=2,
            priority=6,
            table_id=7,
            cookie=8,
            buffer_id=9,
            actions=[action],
        )

        self.assertEqual(generic_flow.id, "78d18f2cffd3eaa069afbf0995b90db9")

    def test_id__x04(self):
        """Test id method 0x04 version."""
        match = {}
        match["in_port"] = MagicMock()
        match["in_port"].value = 22
        match["vlan_vid"] = MagicMock()
        match["vlan_vid"].value = 123

        generic_flow = GenericFlow(
            version="0x04",
            match=match,
            idle_timeout=1,
            hard_timeout=2,
            priority=6,
            table_id=7,
            cookie=8,
            buffer_id=9,
        )

        self.assertEqual(generic_flow.id, "2d843f76b8b254fad6c6e2a114590440")


1		"""Module to test the main napp file."""
2		import json
3		from unittest import TestCase
4		from unittest.mock import MagicMock, patch
5		from kytos.lib.helpers import (
6		get_controller_mock,
7		get_test_client,
8		get_kytos_event_mock,
9		get_switch_mock,
10		)
11		from napps.amlight.flow_stats.main import GenericFlow, Main
12		from napps.kytos.of_core.v0x01.flow import Action as Action10
13		from napps.kytos.of_core.v0x04.flow import Action as Action40
14		from napps.kytos.of_core.v0x04.match_fields import (
15		MatchDLVLAN,
16		MatchFieldFactory,
17		)
18		from pyof.foundation.basic_types import UBInt32
19		from pyof.v0x04.common.flow_instructions import InstructionType
20		from pyof.v0x01.controller2switch.common import StatsType
21		from pyof.v0x04.controller2switch.common import MultipartType
22
23
24		# pylint: disable=too-many-public-methods, too-many-lines
25		class TestMain(TestCase):
26		"""Test the Main class."""
27
28		def setUp(self):
29		"""Execute steps before each tests.
30
31		Set the server_name_url_url from amlight/flow_stats
32		"""
33		self.server_name_url = "http://localhost:8181/api/amlight/flow_stats"
34		self.napp = Main(get_controller_mock())
35
36		@staticmethod
37		def get_napp_urls(napp):
38		"""Return the amlight/flow_stats urls.
39
40		The urls will be like:
41
42		urls = [
43		(options, methods, url)
44		]
45
46		"""
47		controller = napp.controller
48		controller.api_server.register_napp_endpoints(napp)
49
50		urls = []
51		for rule in controller.api_server.app.url_map.iter_rules():
52		options = {}
53		for arg in rule.arguments:
54		options[arg] = f"[{0}]".format(arg)
55
56		if f"{napp.username}/{napp.name}" in str(rule):
57		urls.append((options, rule.methods, f"{str(rule)}"))
58
59		return urls
60
61		def test_verify_api_urls(self):
62		"""Verify all APIs registered."""
63
64		expected_urls = [
65		(
66		{"dpid": "[dpid]"},
67		{"OPTIONS", "HEAD", "GET"},
68		"/api/amlight/flow_stats/flow/match/ ",
69		),
70		(
71		{"dpid": "[dpid]"},
72		{"OPTIONS", "HEAD", "GET"},
73		"/api/amlight/flow_stats/flow/stats/",
74		),
75		(
76		{"flow_id": "[flow_id]"},
77		{"OPTIONS", "HEAD", "GET"},
78		"/api/amlight/flow_stats/packet_count/",
79		),
80		(
81		{"flow_id": "[flow_id]"},
82		{"OPTIONS", "HEAD", "GET"},
83		"/api/amlight/flow_stats/bytes_count/",
84		),
85		(
86		{"dpid": "[dpid]"},
87		{"OPTIONS", "HEAD", "GET"},
88		"/api/amlight/flow_stats/packet_count/per_flow/",
89		),
90		(
91		{"dpid": "[dpid]"},
92		{"OPTIONS", "HEAD", "GET"},
93		"/api/amlight/flow_stats/packet_count/sum/",
94		),
95		(
96		{"dpid": "[dpid]"},
97		{"OPTIONS", "HEAD", "GET"},
98		"/api/amlight/flow_stats/bytes_count/per_flow/",
99		),
100		(
101		{"dpid": "[dpid]"},
102		{"OPTIONS", "HEAD", "GET"},
103		"/api/amlight/flow_stats/bytes_count/sum/",
104		),
105		]
106		urls = self.get_napp_urls(self.napp)
107		self.assertEqual(len(expected_urls), len(urls))
108
109		def test_packet_count__fail(self):
110		"""Test packet_count rest call with wrong flow_id."""
111		flow_id = "123456789"
112		rest_name = "packet_count"
113		response = self._get_rest_response(rest_name, flow_id)
114
115		self.assertEqual(response.data, b"Flow does not exist")
116
117		def test_packet_count(self):
118		"""Test packet_count rest call."""
119		flow_id = "1"
120		rest_name = "packet_count"
121		self._patch_switch_flow(flow_id)
122		response = self._get_rest_response(rest_name, flow_id)
123
124		json_response = json.loads(response.data)
125		self.assertEqual(json_response["flow_id"], flow_id)
126		self.assertEqual(json_response["packet_counter"], 40)
127		self.assertEqual(json_response["packet_per_second"], 2.0)
128
129		def test_bytes_count_fail(self):
130		"""Test bytes_count rest call with wrong flow_id."""
131		flow_id = "123456789"
132		rest_name = "bytes_count"
133		response = self._get_rest_response(rest_name, flow_id)
134
135		self.assertEqual(response.data, b"Flow does not exist")
136
137		def test_bytes_count(self):
138		"""Test bytes_count rest call."""
139		flow_id = "1"
140		rest_name = "bytes_count"
141		self._patch_switch_flow(flow_id)
142		response = self._get_rest_response(rest_name, flow_id)
143
144		json_response = json.loads(response.data)
145		self.assertEqual(json_response["flow_id"], flow_id)
146		self.assertEqual(json_response["bytes_counter"], 10)
147		self.assertEqual(json_response["bits_per_second"], 4.0)
148
149	View Code Duplication	def test_packet_count_per_flow(self):
		0 ignored issues – show Duplication introduced 2022-08-05 19:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
150		"""Test packet_count_per_flow rest call."""
151		flow_id = "1"
152		rest_name = "packet_count/per_flow"
153		self._patch_switch_flow(flow_id)
154
155		dpid_id = 111
156		response = self._get_rest_response(rest_name, dpid_id)
157
158		json_response = json.loads(response.data)
159		self.assertEqual(json_response[0]["flow_id"], flow_id)
160		self.assertEqual(json_response[0]["packet_counter"], 40)
161		self.assertEqual(json_response[0]["packet_per_second"], 2.0)
162
163		def test_packet_count_sum(self):
164		"""Test packet_count_sum rest call."""
165		flow_id = "1"
166		rest_name = "packet_count/sum"
167		self._patch_switch_flow(flow_id)
168
169		dpid_id = 111
170		response = self._get_rest_response(rest_name, dpid_id)
171		json_response = json.loads(response.data)
172
173		self.assertEqual(json_response, 40)
174
175	View Code Duplication	def test_bytes_count_per_flow(self):
		0 ignored issues – show Duplication introduced 2022-08-05 19:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
176		"""Test bytes_count_per_flow rest call."""
177		flow_id = "1"
178		rest_name = "bytes_count/per_flow"
179		self._patch_switch_flow(flow_id)
180
181		dpid_id = 111
182		response = self._get_rest_response(rest_name, dpid_id)
183
184		json_response = json.loads(response.data)
185		self.assertEqual(json_response[0]["flow_id"], flow_id)
186		self.assertEqual(json_response[0]["bytes_counter"], 10)
187		self.assertEqual(json_response[0]["bits_per_second"], 4.0)
188
189		def test_bytes_count_sum(self):
190		"""Test bytes_count_sum rest call."""
191		flow_id = "1"
192		rest_name = "bytes_count/sum"
193		self._patch_switch_flow(flow_id)
194
195		dpid_id = 111
196		response = self._get_rest_response(rest_name, dpid_id)
197		json_response = json.loads(response.data)
198
199		self.assertEqual(json_response, 10)
200
201		@patch("napps.amlight.flow_stats.main.Main.match_flows")
202		def test_flow_match(self, mock_match_flows):
203		"""Test flow_match rest call."""
204		flow = GenericFlow()
205		flow.actions = [
206		Action10.from_dict(
207		{
208		"action_type": "output",
209		"port": "1",
210		}
211		),
212		]
213		flow.version = "0x04"
214		mock_match_flows.return_value = flow
215
216		flow_id = "1"
217		rest_name = "flow/match"
218		self._patch_switch_flow(flow_id)
219
220		dpid_id = "aa:00:00:00:00:00:00:11"
221		response = self._get_rest_response(rest_name, dpid_id)
222		json_response = json.loads(response.data)
223
224		self.assertEqual(response.status_code, 200)
225		self.assertEqual(json_response["actions"][0]["action_type"], "output")
226		self.assertEqual(json_response["actions"][0]["port"], "1")
227		self.assertEqual(json_response["version"], "0x04")
228
229		@patch("napps.amlight.flow_stats.main.Main.match_flows")
230		def test_flow_match_fail(self, mock_match_flows):
231		"""Test flow_match rest call."""
232		mock_match_flows.return_value = None
233
234		flow_id = "1"
235		rest_name = "flow/match"
236		self._patch_switch_flow(flow_id)
237
238		dpid_id = "aa:00:00:00:00:00:00:11"
239		response = self._get_rest_response(rest_name, dpid_id)
240
241		self.assertEqual(response.status_code, 404)
242
243		@patch("napps.amlight.flow_stats.main.Main.match_flows")
244		def test_flow_stats(self, mock_match_flows):
245		"""Test flow_match rest call."""
246		flow = GenericFlow()
247		flow.actions = [
248		Action10.from_dict(
249		{
250		"action_type": "output",
251		"port": "1",
252		}
253		),
254		]
255		flow.version = "0x04"
256		mock_match_flows.return_value = [flow]
257
258		flow_id = "1"
259		rest_name = "flow/stats"
260		self._patch_switch_flow(flow_id)
261
262		dpid_id = "aa:00:00:00:00:00:00:11"
263		response = self._get_rest_response(rest_name, dpid_id)
264		json_response = json.loads(response.data)
265
266		self.assertEqual(response.status_code, 200)
267		print(json_response)
268		self.assertEqual(json_response[0]["actions"][0]["action_type"],
269		"output")
270		self.assertEqual(json_response[0]["actions"][0]["port"], "1")
271		self.assertEqual(json_response[0]["version"], "0x04")
272
273		def _patch_switch_flow(self, flow_id):
274		"""Helper method to patch controller to return switch/flow data."""
275		# patching the flow_stats object in the switch
276		self.napp.controller.switches = MagicMock()
277		flow = self._get_mocked_flow_stats()
278		flow.id = flow_id
279		switch = MagicMock()
280		switch.generic_flows = [flow]
281		self.napp.controller.switches.values.return_value = [switch]
282		self.napp.controller.get_switch_by_dpid = MagicMock()
283		self.napp.controller.get_switch_by_dpid.return_value = switch
284
285		def _get_rest_response(self, rest_name, url_id):
286		"""Helper method to call a rest endpoint."""
287		# call rest
288		api = get_test_client(get_controller_mock(), self.napp)
289		url = f"{self.server_name_url}/{rest_name}/{url_id}"
290		response = api.get(url, content_type="application/json")
291
292		return response
293
294		# pylint: disable=no-self-use
295		def _get_mocked_flow_stats(self):
296		"""Helper method to create a mock flow_stats object."""
297		flow_stats = MagicMock()
298		flow_stats.id = 123
299		flow_stats.byte_count = 10
300		flow_stats.duration_sec = 20
301		flow_stats.duration_nsec = 30
302		flow_stats.packet_count = 40
303		return flow_stats
304
305		def _get_mocked_flow_base(self):
306		"""Helper method to create a mock flow object."""
307		flow = MagicMock()
308		flow.id = 456
309		flow.switch = None
310		flow.table_id = None
311		flow.match = None
312		flow.priority = None
313		flow.idle_timeout = None
314		flow.hard_timeout = None
315		flow.cookie = None
316		flow.stats = self._get_mocked_flow_stats()
317		return flow
318
319		@patch("napps.amlight.flow_stats.main.GenericFlow.from_flow_stats")
320		def test_handle_stats_reply(self, mock_from_flow):
321		"""Test handle_stats_reply rest call."""
322		mock_from_flow.return_value = self._get_mocked_flow_base()
323
324		def side_effect(event):
325		self.assertTrue(f"{event}", "amlight/flow_stats.flows_updated")
326		self.assertTrue(event.content["switch"], 111)
327
328		self.napp.controller = MagicMock()
329		self.napp.controller.buffers.app.put.side_effect = side_effect
330
331		msg = MagicMock()
332		msg.flags.value = 2
333		msg.body = [self._get_mocked_flow_stats()]
334		event_switch = MagicMock()
335		event_switch.generic_flows = []
336		event_switch.dpid = 111
337		self.napp.handle_stats_reply(msg, event_switch)
338
339		# Check if important trace dont trigger the event
340		# It means that the CP trace is the same to the DP trace
341		self.napp.controller.buffers.app.put.assert_called_once()
342
343		# Check mocked flow id
344		self.assertEqual(event_switch.generic_flows[0].id, 456)
345
346		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
347		def test_handle_stats_reply_0x01(self, mock_handle_stats):
348		"""Test handle stats reply."""
349		flow_msg = MagicMock()
350		flow_msg.body = "A"
351		flow_msg.body_type = StatsType.OFPST_FLOW
352
353		switch_v0x01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x01)
354
355		name = "kytos/of_core.v0x01.messages.in.ofpt_stats_reply"
356		content = {"source": switch_v0x01.connection, "message": flow_msg}
357		event = get_kytos_event_mock(name=name, content=content)
358
359		self.napp.handle_stats_reply_0x01(event)
360		mock_handle_stats.assert_called_once()
361
362		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
363		def test_handle_stats_reply_0x01__fail(self, mock_handle_stats):
364		"""Test handle stats reply."""
365		flow_msg = MagicMock()
366		flow_msg.body = "A"
367		flow_msg.body_type = StatsType.OFPST_DESC
368
369		switch_v0x01 = get_switch_mock("00:00:00:00:00:00:00:01", 0x01)
370
371		name = "kytos/of_core.v0x01.messages.in.ofpt_stats_reply"
372		content = {"source": switch_v0x01.connection, "message": flow_msg}
373		event = get_kytos_event_mock(name=name, content=content)
374
375		self.napp.handle_stats_reply_0x01(event)
376
377		mock_handle_stats.assert_not_called()
378
379		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
380		def test_handle_stats_reply_0x04(self, mock_handle_stats):
381		"""Test handle stats reply."""
382		flow_msg = MagicMock()
383		flow_msg.body = "A"
384		flow_msg.multipart_type = MultipartType.OFPMP_FLOW
385
386		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
387
388		name = "kytos/of_core.v0x04.messages.in.ofpt_multipart_reply"
389		content = {"source": switch_v0x04.connection, "message": flow_msg}
390		event = get_kytos_event_mock(name=name, content=content)
391		event.content["message"] = flow_msg
392
393		self.napp.handle_stats_reply_0x04(event)
394
395		mock_handle_stats.assert_called_once()
396
397		@patch("napps.amlight.flow_stats.main.Main.handle_stats_reply")
398		def test_handle_stats_reply_0x04_fail(self, mock_handle_stats):
399		"""Test handle stats reply."""
400		flow_msg = MagicMock()
401		flow_msg.body = "A"
402
403		flow_msg.multipart_type = MultipartType.OFPMP_PORT_DESC
404
405		switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
406
407		name = "kytos/of_core.v0x04.messages.in.ofpt_multipart_reply"
408		content = {"source": switch_v0x04.connection, "message": flow_msg}
409
410		event = get_kytos_event_mock(name=name, content=content)
411		event.content["message"] = flow_msg
412
413		self.napp.handle_stats_reply_0x04(event)
414
415		mock_handle_stats.assert_not_called()
416
417
418		# pylint: disable=too-many-public-methods, too-many-lines
419		class TestGenericFlow(TestCase):
420		"""Test the GenericFlow class."""
421
422		# pylint: disable=no-member
423		def test_from_flow_stats__x01(self):
424		"""Test from_flow_stats method 0x01 version."""
425		flow_stats = MagicMock()
426
427		flow_stats.actions = [
428		Action10.from_dict(
429		{
430		"action_type": "output",
431		"port": UBInt32(1),
432		}
433		).as_of_action(),
434		]
435
436		result = GenericFlow.from_flow_stats(flow_stats, version="0x01")
437
438		self.assertEqual(result.idle_timeout, flow_stats.idle_timeout.value)
439		self.assertEqual(result.hard_timeout, flow_stats.hard_timeout.value)
440		self.assertEqual(result.priority, flow_stats.priority.value)
441		self.assertEqual(result.table_id, flow_stats.table_id.value)
442		self.assertEqual(result.duration_sec, flow_stats.duration_sec.value)
443		self.assertEqual(result.packet_count, flow_stats.packet_count.value)
444		self.assertEqual(result.byte_count, flow_stats.byte_count.value)
445
446		exp_match = flow_stats.match
447		self.assertEqual(result.match["wildcards"], exp_match.wildcards.value)
448		self.assertEqual(result.match["in_port"], exp_match.in_port.value)
449		self.assertEqual(result.match["eth_src"], exp_match.dl_src.value)
450		self.assertEqual(result.match["eth_dst"], exp_match.dl_dst.value)
451		self.assertEqual(result.match["vlan_vid"], exp_match.dl_vlan.value)
452		self.assertEqual(result.match["vlan_pcp"], exp_match.dl_vlan_pcp.value)
453		self.assertEqual(result.match["eth_type"], exp_match.dl_type.value)
454		self.assertEqual(result.match["ip_tos"], exp_match.nw_tos.value)
455		self.assertEqual(result.match["ipv4_src"], exp_match.nw_src.value)
456		self.assertEqual(result.match["ipv4_dst"], exp_match.nw_dst.value)
457		self.assertEqual(result.match["ip_proto"], exp_match.nw_proto.value)
458		self.assertEqual(result.match["tcp_src"], exp_match.tp_src.value)
459		self.assertEqual(result.match["tcp_dst"], exp_match.tp_dst.value)
460
461		def test_from_flow_stats__x04_match(self):
462		"""Test from_flow_stats method 0x04 version with match."""
463		flow_stats = MagicMock()
464		flow_stats.actions = [
465		Action10.from_dict(
466		{
467		"action_type": "output",
468		"port": UBInt32(1),
469		}
470		).as_of_action(),
471		]
472
473		flow_stats.match.oxm_match_fields = [MatchDLVLAN(42).as_of_tlv()]
474
475		result = GenericFlow.from_flow_stats(flow_stats, version="0x04")
476
477		match_expect = MatchFieldFactory.from_of_tlv(
478		flow_stats.match.oxm_match_fields[0]
479		)
480		self.assertEqual(result.match["vlan_vid"], match_expect)
481
482		def test_from_flow_stats__x04_action(self):
483		"""Test from_flow_stats method 0x04 version with action."""
484		flow_stats = MagicMock()
485
486		action_dict = {
487		"action_type": "output",
488		"port": UBInt32(1),
489		}
490		instruction = MagicMock()
491		instruction.instruction_type = InstructionType.OFPIT_APPLY_ACTIONS
492		instruction.actions = [Action40.from_dict(action_dict).as_of_action()]
493		flow_stats.instructions = [instruction]
494
495		result = GenericFlow.from_flow_stats(flow_stats, version="0x04")
496		self.assertEqual(result.actions[0].as_dict(), action_dict)
497
498		def test_to_dict__x01(self):
499		"""Test to_dict method 0x01 version."""
500		action = Action10.from_dict(
501		{
502		"action_type": "set_vlan",
503		"vlan_id": 6,
504		}
505		)
506		match = {}
507		match["in_port"] = 11
508
509		generic_flow = GenericFlow(
510		version="0x01",
511		match=match,
512		idle_timeout=1,
513		hard_timeout=2,
514		duration_sec=3,
515		packet_count=4,
516		byte_count=5,
517		priority=6,
518		table_id=7,
519		cookie=8,
520		buffer_id=9,
521		actions=[action],
522		)
523
524		result = generic_flow.to_dict()
525
526		expected = {
527		"version": "0x01",
528		"idle_timeout": 1,
529		"in_port": 11,
530		"hard_timeout": 2,
531		"priority": 6,
532		"table_id": 7,
533		"cookie": 8,
534		"buffer_id": 9,
535		"actions": [{"vlan_id": 6, "action_type": "set_vlan"}],
536		}
537
538		self.assertEqual(result, expected)
539
540		def test_to_dict__x04(self):
541		"""Test to_dict method 0x04 version."""
542		match = {}
543		match["in_port"] = MagicMock()
544		match["in_port"].value = 22
545
546		generic_flow = GenericFlow(
547		version="0x04",
548		match=match,
549		)
550
551		result = generic_flow.to_dict()
552		expected = {
553		"version": "0x04",
554		"in_port": 22,
555		"idle_timeout": 0,
556		"hard_timeout": 0,
557		"priority": 0,
558		"table_id": 255,
559		"cookie": None,
560		"buffer_id": None,
561		"actions": [],
562		}
563
564		self.assertEqual(result, expected)
565
566		def test_match_to_dict(self):
567		"""Test match_to_dict method for 0x04 version."""
568		match = {}
569		match["in_port"] = MagicMock()
570		match["in_port"].value = 22
571		match["vlan_vid"] = MagicMock()
572		match["vlan_vid"].value = 123
573
574		generic_flow = GenericFlow(
575		version="0x04",
576		match=match,
577		)
578		result = generic_flow.match_to_dict()
579		expected = {"in_port": 22, "vlan_vid": 123}
580
581		self.assertEqual(result, expected)
582
583		def test_match_to_dict__empty_match(self):
584		"""Test match_to_dict method for 0x04 version, with empty matches."""
585		generic_flow = GenericFlow(version="0x04")
586		result = generic_flow.match_to_dict()
587		self.assertEqual(result, {})
588
589		def test_id__x01(self):
590		"""Test id method 0x01 version."""
591		action = Action10.from_dict(
592		{
593		"action_type": "set_vlan",
594		"vlan_id": 6,
595		}
596		)
597		match = {}
598		match["in_port"] = 11
599
600		generic_flow = GenericFlow(
601		version="0x01",
602		match=match,
603		idle_timeout=1,
604		hard_timeout=2,
605		priority=6,
606		table_id=7,
607		cookie=8,
608		buffer_id=9,
609		actions=[action],
610		)
611
612		self.assertEqual(generic_flow.id, "78d18f2cffd3eaa069afbf0995b90db9")
613
614		def test_id__x04(self):
615		"""Test id method 0x04 version."""
616		match = {}
617		match["in_port"] = MagicMock()
618		match["in_port"].value = 22
619		match["vlan_vid"] = MagicMock()
620		match["vlan_vid"].value = 123
621
622		generic_flow = GenericFlow(
623		version="0x04",
624		match=match,
625		idle_timeout=1,
626		hard_timeout=2,
627		priority=6,
628		table_id=7,
629		cookie=8,
630		buffer_id=9,
631		)
632
633		self.assertEqual(generic_flow.id, "2d843f76b8b254fad6c6e2a114590440")
634

kytos-ng / flow_stats

Push — master ( 645373...00c642 )

build.tests.unit.test_main.TestMain.setUpClass() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like