build.tests.unit.test_main.TestMain.test_handle_stats_received() - Code Metrics - Inspection of "Fix error in from_replies_flows" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#34)

unknown

created 2022-09-15 01:38 UTC

TestMain.test_handle_stats_received() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	13
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	9
dl	0
loc	13
rs	9.95
c	0
b	0
f	0
cc	1
nop	2

"""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


# 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
        flow = self._get_mocked_flow_stats()
        flow.id = flow_id
        switch = MagicMock()
        switch.generic_flows = [flow]
        self.napp.controller.switches = {"1": 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_multipart_replies_flows(self):
        """Helper method to create mock multipart replies flows"""
        flow = self._get_mocked_flow_base()

        instruction = MagicMock()
        flow.instructions = [instruction]

        replies_flows = [flow]
        return replies_flows

    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_received")
    def test_handle_stats_received(self, mock_handle_stats):
        """Test handle_stats_received function."""

        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        replies_flows = self._get_mocked_multipart_replies_flows()
        name = "kytos/of_core.flow_stats.received"
        content = {"switch": switch_v0x04, "replies_flows": replies_flows}

        event = get_kytos_event_mock(name=name, content=content)

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

    @patch("napps.amlight.flow_stats.main.Main.handle_stats_reply_received")
    def test_handle_stats_received_fail(self, mock_handle_stats):
        """Test handle_stats_received function for
        fail when replies_flows is not in content."""

        switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
        name = "kytos/of_core.flow_stats.received"
        content = {"switch": switch_v0x04}

        event = get_kytos_event_mock(name=name, content=content)

        self.napp.handle_stats_received(event)
        mock_handle_stats.assert_not_called()

    @patch("napps.amlight.flow_stats.main.GenericFlow.from_replies_flows")
    def test_handle_stats_reply_received(self, mock_from_flow):
        """Test handle_stats_reply_received call."""
        mock_from_flow.return_value = self._get_mocked_flow_base()

        event_switch = MagicMock()
        flows_mock = self._get_mocked_multipart_replies_flows()
        self.napp.handle_stats_reply_received(event_switch, flows_mock)

        self.assertEqual(event_switch.generic_flows[0].id, 456)


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

kytos-ng / flow_stats

Pull Request — master (#34)

TestMain.test_handle_stats_received() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like