build.tests.unit.test_main.TestGenericFlow.test_from_replies_flows() - Code Metrics - Inspection of "Listen to kytos/of_core.flow_stats.received" - kytos-ng/flow_stats - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#28)

unknown

created 2022-08-31 17:01 UTC

TestGenericFlow.test_from_replies_flows() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	34
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	26
dl	0
loc	34
rs	9.256
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 (
    MatchDLVLAN,
    MatchFieldFactory,
)
from pyof.foundation.basic_types import UBInt32
from pyof.v0x04.common.flow_instructions import InstructionType


# 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_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()

        self.napp.controller = MagicMock()

        event_switch = MagicMock()
        event_switch.dpid = 111
        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)

    # pylint: disable=no-member
    def test_from_replies_flows(self):
        """Test from_replies_flows function."""
        replies_flow = MagicMock()

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

        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()]
        replies_flow.instructions = [instruction]

        result = GenericFlow.from_replies_flows(replies_flow)

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

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

kytos-ng / flow_stats

Pull Request — master (#28)

TestGenericFlow.test_from_replies_flows() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like