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

Passed

Pull Request — master (#34)

unknown

created 2022-09-15 03:51 UTC

TestMain.test_verify_api_urls() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	47
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	36
dl	0
loc	47
rs	9.016
c	0
b	0
f	0
cc	1
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 MatchFieldFactory
from napps.kytos.of_core.v0x04.flow import Match as Match40
from pyof.foundation.basic_types import UBInt32
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_replies_flows(self):
        """Test from_replies_flows method 0x04 version."""
        replies_flow = MagicMock()

        action_dict = {
            "action_type": "output",
            "port": UBInt32(1),
        }

        actions = Action40.from_dict(action_dict).as_of_action()

        instruction = MagicMock()
        instruction.instruction_type = 'apply_actions'
        instruction.actions = [actions]
        replies_flow.instructions = [instruction]
        match = Match40(42)
        replies_flow.match = match
        result = GenericFlow.from_replies_flows(replies_flow)

        self.assertEqual(result.idle_timeout, replies_flow.idle_timeout)
        self.assertEqual(result.hard_timeout, replies_flow.hard_timeout)
        self.assertEqual(result.priority, replies_flow.priority)
        self.assertEqual(result.table_id, replies_flow.table_id)
        self.assertEqual(result.cookie, replies_flow.cookie)
        self.assertEqual(result.duration_sec, replies_flow.stats.duration_sec)
        self.assertEqual(result.packet_count, replies_flow.stats.packet_count)
        self.assertEqual(result.byte_count, replies_flow.stats.byte_count)

        # pylint: disable=too-many-public-methods, too-many-lines
        match_expect = MatchFieldFactory.from_of_tlv(
            match.as_of_match().oxm_match_fields[0]
            )
        self.assertEqual(result.actions[0], actions)
        self.assertEqual(result.match["in_port"], match_expect)

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

kytos-ng / flow_stats

Pull Request — master (#34)

TestMain.test_verify_api_urls() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like