build.tests.unit.test_main.TestMain.test_handle_stats_reply_0x04() - Code Metrics - Inspection of "Increase unit test coverage" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#22)

by Rogerio

created 2022-08-10 22:07 UTC

TestMain.test_handle_stats_reply_0x04() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	17
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	17
rs	9.8
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 importlib import reload
import sys
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."""
    @classmethod
    def setUpClass(cls):
        # The decorator run_on_thread is patched, so methods that listen
        # for events do not run on threads while tested.
        # Decorators have to be patched before the methods that are
        # decorated with them are imported.
        reload(sys.modules["kytos.core.helpers"])
        patch("kytos.core.helpers.run_on_thread", lambda x: x).start()

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

kytos-ng / flow_stats

Pull Request — master (#22)

TestMain.test_handle_stats_reply_0x04() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like