build.tests.unit.test_main.TestMain.test_negotiate() - Code Metrics - Inspection of "Merge pull request #66 from cmagnobarbosa/unittest..." - kytos/of_core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1b459e...75d007 )

by Humberto

created 2020-03-31 17:41 UTC

TestMain.test_negotiate() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	18
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	17
nop	2
dl	0
loc	18
rs	9.55
c	0
b	0
f	0

"""Test Main methods."""
from unittest import TestCase
from unittest.mock import MagicMock, patch, create_autospec

from pyof.v0x01.controller2switch.common import StatsType
from pyof.v0x04.controller2switch.common import MultipartType
from pyof.foundation.network_types import Ethernet

from kytos.core.connection import ConnectionState
from tests.unit.helpers import (get_connection_mock, get_controller_mock,
                                get_kytos_event_mock, get_switch_mock)


# pylint: disable=protected-access
class TestMain(TestCase):
    """docstring for TestMain."""

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

        Set the server_name_url from kytos/of_core
        """
        self.switch_v0x01 = get_switch_mock("00:00:00:00:00:00:00:01")
        self.switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:02")
        self.switch_v0x01.connection = get_connection_mock(
            0x01, get_switch_mock("00:00:00:00:00:00:00:03"))
        self.switch_v0x04.connection = get_connection_mock(
            0x04, get_switch_mock("00:00:00:00:00:00:00:04"))

        patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
        from napps.kytos.of_core.main import Main
        self.addCleanup(patch.stopall)

        self.napp = Main(get_controller_mock())

    @patch('napps.kytos.of_core.v0x04.utils.update_flow_list')
    @patch('napps.kytos.of_core.v0x01.utils.update_flow_list')
    def test_request_flow_list(self, *args):
        """Test request flow list."""
        (mock_update_flow_list_v0x01, mock_update_flow_list_v0x04) = args
        mock_update_flow_list_v0x04.return_value = "ABC"

        self.napp._request_flow_list(self.switch_v0x01)
        mock_update_flow_list_v0x01.assert_called_with(self.napp.controller,
                                                       self.switch_v0x01)
        self.napp._request_flow_list(self.switch_v0x04)
        mock_update_flow_list_v0x04.assert_called_with(self.napp.controller,
                                                       self.switch_v0x04)

    @patch('napps.kytos.of_core.v0x01.flow.Flow.from_of_flow_stats')
    @patch('kytos.core.switch.Switch.update_description')
    def test_handle_stats_reply(self, *args):
        """Test handle stats reply."""
        (mock_update_description, mock_from_of_flow_stats_v0x01) = args
        mock_from_of_flow_stats_v0x01.return_value = "ABC"

        flow_msg = MagicMock()
        flow_msg.body = "A"
        flow_msg.body_type = StatsType.OFPST_FLOW
        event = get_kytos_event_mock(source=self.switch_v0x01.connection,
                                     message=flow_msg)
        self.napp.handle_stats_reply(event)
        mock_from_of_flow_stats_v0x01.assert_called_with(
            flow_msg.body, self.switch_v0x01.connection.switch)

        desc_msg = MagicMock()
        desc_msg.body = "A"
        desc_msg.body_type = StatsType.OFPST_DESC
        event = get_kytos_event_mock(source=self.switch_v0x01.connection,
                                     message=desc_msg)
        self.napp.handle_stats_reply(event)
        mock_update_description.assert_called_with(desc_msg.body)

    @patch('kytos.core.switch.Switch.update_description')
    @patch('napps.kytos.of_core.main.Main._handle_multipart_flow_stats')
    @patch('napps.kytos.of_core.v0x04.utils.handle_port_desc')
    def test_handle_multipart_reply(self, *args):
        """Test handle multipart reply."""
        (mock_of_core_v0x04_utils, mock_from_of_flow_stats_v0x04,
         mock_update_description) = args

        flow_msg = MagicMock()
        flow_msg.multipart_type = MultipartType.OFPMP_FLOW
        event = get_kytos_event_mock(source=self.switch_v0x01.connection,
                                     message=flow_msg)

        self.napp.handle_multipart_reply(event)
        mock_from_of_flow_stats_v0x04.assert_called_with(
            flow_msg, self.switch_v0x01.connection.switch)

        ofpmp_port_desc = MagicMock()
        ofpmp_port_desc.body = "A"
        ofpmp_port_desc.multipart_type = MultipartType.OFPMP_PORT_DESC
        event = get_kytos_event_mock(source=self.switch_v0x01.connection,
                                     message=ofpmp_port_desc)
        self.napp.handle_multipart_reply(event)
        mock_of_core_v0x04_utils.assert_called_with(
            self.napp.controller, self.switch_v0x01.connection.switch,
            ofpmp_port_desc.body)

        ofpmp_desc = MagicMock()
        ofpmp_desc.body = "A"
        ofpmp_desc.multipart_type = MultipartType.OFPMP_DESC
        event = get_kytos_event_mock(source=self.switch_v0x01.connection,
                                     message=ofpmp_desc)
        self.napp.handle_multipart_reply(event)
        mock_update_description.assert_called_with(ofpmp_desc.body)

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_core.v0x04.utils.send_set_config')
    @patch('napps.kytos.of_core.v0x01.utils.send_set_config')
    @patch('napps.kytos.of_core.v0x04.utils.send_desc_request')
    @patch('napps.kytos.of_core.v0x01.utils.send_desc_request')
    @patch('kytos.core.connection.Connection.set_established_state')
    @patch('kytos.core.connection.Connection.is_during_setup')
    @patch('napps.kytos.of_core.v0x04.utils.handle_features_reply')
    @patch('napps.kytos.of_core.v0x01.utils.handle_features_reply')
    def test_handle_features_reply(self, *args):
        """Test handle features reply."""
        (mock_freply_v0x01, mock_freply_v0x04, mock_is_during_setup,
         mock_set_established_state, mock_send_desc_request_v0x01,
         mock_send_desc_request_v0x04, mock_send_set_config_v0x01,
         mock_send_set_config_v0x04, mock_buffers_put) = args
        mock_freply_v0x01.return_value = self.switch_v0x01.connection.switch
        mock_freply_v0x04.return_value = self.switch_v0x04.connection.switch
        mock_is_during_setup.return_value = True

        self.switch_v0x01.connection.state = ConnectionState.SETUP
        self.switch_v0x01.connection.protocol.state = 'waiting_features_reply'
        event = get_kytos_event_mock(source=self.switch_v0x01.connection)
        self.napp.handle_features_reply(event)
        mock_freply_v0x01.assert_called_with(self.napp.controller, event)
        mock_send_desc_request_v0x01.assert_called_with(
            self.napp.controller, self.switch_v0x01.connection.switch)
        mock_send_set_config_v0x01.assert_called_with(
            self.napp.controller, self.switch_v0x01.connection.switch)

        self.switch_v0x04.connection.state = ConnectionState.SETUP
        self.switch_v0x04.connection.protocol.state = 'waiting_features_reply'
        event = get_kytos_event_mock(source=self.switch_v0x04.connection)
        self.napp.handle_features_reply(event)
        mock_freply_v0x04.assert_called_with(self.napp.controller, event)
        mock_send_desc_request_v0x04.assert_called_with(
            self.napp.controller, self.switch_v0x04.connection.switch)
        mock_send_set_config_v0x04.assert_called_with(
            self.napp.controller, self.switch_v0x04.connection.switch)

        mock_is_during_setup.assert_called()
        mock_set_established_state.assert_called()
        mock_buffers_put.assert_called()

    @patch('napps.kytos.of_core.main.Main._update_switch_flows')
    @patch('napps.kytos.of_core.v0x04.flow.Flow.from_of_flow_stats')
    @patch('napps.kytos.of_core.main.Main._is_multipart_reply_ours')
    def test_handle_multipart_flow_stats(self, *args):
        """Test handle multipart flow stats."""
        (mock_is_multipart_reply_ours, mock_from_of_flow_stats_v0x01,
         mock_update_switch_flows) = args
        mock_is_multipart_reply_ours.return_value = True
        mock_from_of_flow_stats_v0x01.return_value = "ABC"

        flow_msg = MagicMock()
        flow_msg.body = "A"
        flow_msg.flags.value = 2
        flow_msg.body_type = StatsType.OFPST_FLOW

        self.napp._handle_multipart_flow_stats(flow_msg, self.switch_v0x04)

        mock_is_multipart_reply_ours.assert_called_with(flow_msg,
                                                        self.switch_v0x04)
        mock_from_of_flow_stats_v0x01.assert_called_with(flow_msg.body,
                                                         self.switch_v0x04)
        mock_update_switch_flows.assert_called_with(self.switch_v0x04)

    @patch('napps.kytos.of_core.main.Main.update_port_status')
    @patch('napps.kytos.of_core.main.Main.update_links')
    def test_emit_message_in(self, *args):
        """Test emit_message_in."""
        (mock_update_links, mock_update_port_status) = args

        mock_port_connection = MagicMock()
        msg_port_mock = MagicMock()
        msg_port_mock.header.message_type.name = 'ofpt_port_status'
        mock_port_connection.side_effect = True
        self.napp.emit_message_in(mock_port_connection,
                                  msg_port_mock)
        mock_update_port_status.assert_called_with(msg_port_mock,
                                                   mock_port_connection)

        mock_packet_in_connection = MagicMock()
        msg_packet_in_mock = MagicMock()
        mock_packet_in_connection.side_effect = True
        msg_packet_in_mock.header.message_type.name = 'ofpt_packet_in'
        self.napp.emit_message_in(mock_packet_in_connection,
                                  msg_packet_in_mock)
        mock_update_links.assert_called_with(msg_packet_in_mock,
                                             mock_packet_in_connection)

    @patch('pyof.utils.v0x04.symmetric.echo_reply.EchoReply')
    @patch('napps.kytos.of_core.main.Main.emit_message_out')
    def test_handle_echo_request(self, *args):
        """Test handle echo request messages."""
        (mock_emit_message_out, mock_echo_reply) = args
        mock_event = MagicMock()
        mock_echo_request = MagicMock()
        mock_echo_reply.return_value = "A"
        mock_echo_request.header.xid = "A"
        mock_echo_request.data = "A"
        mock_event.source.protocol.version = 4
        mock_event.message = mock_echo_request
        self.napp.handle_echo_request(mock_event)
        mock_echo_reply.assert_called_with(xid=mock_echo_request.header.xid,
                                           data=mock_echo_request.data)
        mock_emit_message_out.assert_called_with(mock_event.source, "A")

    @patch('napps.kytos.of_core.main.Main.send_features_request')
    @patch('napps.kytos.of_core.v0x04.utils.say_hello')
    @patch('napps.kytos.of_core.main._get_version_from_bitmask')
    @patch('napps.kytos.of_core.main._get_version_from_header')
    def test_negotiate(self, *args):
        """Test negotiate."""
        (mock_version_header, mock_version_bitmask, mock_say_hello,
         mock_features_request) = args
        mock_version_header.return_value = 4
        mock_version_bitmask.return_value = 4
        mock_connection = MagicMock()
        mock_message = MagicMock()
        mock_message.versions = 4
        self.napp._negotiate(mock_connection, mock_message)
        mock_version_bitmask.assert_called_with(mock_message.versions)
        mock_say_hello.assert_called_with(self.napp.controller,
                                          mock_connection)
        mock_features_request.assert_called_with(mock_connection)

    @patch('pyof.utils.v0x04.asynchronous.error_msg.ErrorMsg')
    @patch('napps.kytos.of_core.main.Main.emit_message_out')
    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def tests_fail_negotiation(self, *args):
        """Test fail_negotiation."""
        (mock_event_buffer, mock_emit_message_out,
         mock_error_msg) = args
        mock_connection = MagicMock()
        mock_message = MagicMock()
        mock_connection.id = "A"
        mock_message.side_effect = 4
        self.napp.fail_negotiation(mock_connection, mock_message)
        mock_event_buffer.assert_called()
        mock_emit_message_out.assert_called_with(mock_connection,
                                                 mock_error_msg.return_value)

    @patch('napps.kytos.of_core.settings.SEND_FEATURES_REQUEST_ON_ECHO')
    @patch('napps.kytos.of_core.main.Main.send_features_request')
    def test_handle_queued_openflow_echo_reply(self, *args):
        """Test handle queued OpenFlow echo reply messages."""
        (mock_send_features_request, mock_settings) = args
        mock_settings.return_value = True
        mock_event = MagicMock()
        self.napp.handle_queued_openflow_echo_reply(mock_event)
        mock_send_features_request.assert_called_with(mock_event.destination)

    @patch('pyof.utils.v0x04.controller2switch.'
           'features_request.FeaturesRequest')
    @patch('napps.kytos.of_core.main.Main.emit_message_out')
    def test_send_features_request(self, *args):
        """Test send send_features_request."""
        (mock_emit_message_out, mock_features_request) = args
        mock_destination = MagicMock()
        mock_destination.protocol.version = 4
        mock_features_request.return_value = "A"
        self.napp.send_features_request(mock_destination)
        mock_features_request.assert_called()
        mock_emit_message_out.assert_called_with(mock_destination, "A")

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_core.main.Ethernet')
    def test_update_links(self, *args):
        """Test update_links."""
        (mock_ethernet, mock_buffer_put) = args
        ethernet = create_autospec(Ethernet)
        ethernet.ether_type = "A"
        mock_ethernet.side_effect = ethernet
        mock_message = MagicMock()
        mock_source = MagicMock()
        self.napp.update_links(mock_message, mock_source)
        mock_ethernet.assert_called()
        mock_buffer_put.assert_called()

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    def test_send_specific_port_mod(self, mock_buffer_put):
        """Test send specific port."""
        mock_port = MagicMock()
        mock_interface = MagicMock()
        current_state = 2
        self.napp._send_specific_port_mod(mock_port,
                                          mock_interface, current_state)
        mock_buffer_put.assert_called()

    @patch('kytos.core.buffers.KytosEventBuffer.put')
    @patch('napps.kytos.of_core.main.Interface')
    @patch('napps.kytos.of_core.main.Main._send_specific_port_mod')
    def test_update_port_status(self, *args):
        """Test update_port_status."""
        (mock_port_mod, mock_interface, mock_buffer_put) = args
        mock_port_status = MagicMock()
        mock_source = MagicMock()
        mock_port_status.reason.value.side_effect = [0, 1]
        mock_port_status.reason.enum_ref(0).name = 'OFPPR_ADD'
        self.napp.update_port_status(mock_port_status, mock_source)
        mock_interface.assert_called()

        # check OFPRR_MODIFY
        mock_port_status.reason.enum_ref(1).name = 'OFPPR_MODIFY'
        self.napp.update_port_status(mock_port_status, mock_source)
        mock_port_mod.assert_called()
        mock_buffer_put.assert_called()


1			"""Test Main methods."""
2			from unittest import TestCase
3			from unittest.mock import MagicMock, patch, create_autospec
4
5			from pyof.v0x01.controller2switch.common import StatsType
6			from pyof.v0x04.controller2switch.common import MultipartType
7			from pyof.foundation.network_types import Ethernet
8
9			from kytos.core.connection import ConnectionState
10			from tests.unit.helpers import (get_connection_mock, get_controller_mock,
11			get_kytos_event_mock, get_switch_mock)
12
13
14			# pylint: disable=protected-access
15			class TestMain(TestCase):
16			"""docstring for TestMain."""
17
18			def setUp(self):
19			"""Execute steps before each tests.
20
21			Set the server_name_url from kytos/of_core
22			"""
23			self.switch_v0x01 = get_switch_mock("00:00:00:00:00:00:00:01")
24			self.switch_v0x04 = get_switch_mock("00:00:00:00:00:00:00:02")
25			self.switch_v0x01.connection = get_connection_mock(
26			0x01, get_switch_mock("00:00:00:00:00:00:00:03"))
27			self.switch_v0x04.connection = get_connection_mock(
28			0x04, get_switch_mock("00:00:00:00:00:00:00:04"))
29
30			patch('kytos.core.helpers.run_on_thread', lambda x: x).start()
31			from napps.kytos.of_core.main import Main
32			self.addCleanup(patch.stopall)
33
34			self.napp = Main(get_controller_mock())
35
36			@patch('napps.kytos.of_core.v0x04.utils.update_flow_list')
37			@patch('napps.kytos.of_core.v0x01.utils.update_flow_list')
38			def test_request_flow_list(self, *args):
39			"""Test request flow list."""
40			(mock_update_flow_list_v0x01, mock_update_flow_list_v0x04) = args
41			mock_update_flow_list_v0x04.return_value = "ABC"
42
43			self.napp._request_flow_list(self.switch_v0x01)
44			mock_update_flow_list_v0x01.assert_called_with(self.napp.controller,
45			self.switch_v0x01)
46			self.napp._request_flow_list(self.switch_v0x04)
47			mock_update_flow_list_v0x04.assert_called_with(self.napp.controller,
48			self.switch_v0x04)
49
50			@patch('napps.kytos.of_core.v0x01.flow.Flow.from_of_flow_stats')
51			@patch('kytos.core.switch.Switch.update_description')
52			def test_handle_stats_reply(self, *args):
53			"""Test handle stats reply."""
54			(mock_update_description, mock_from_of_flow_stats_v0x01) = args
55			mock_from_of_flow_stats_v0x01.return_value = "ABC"
56
57			flow_msg = MagicMock()
58			flow_msg.body = "A"
59			flow_msg.body_type = StatsType.OFPST_FLOW
60			event = get_kytos_event_mock(source=self.switch_v0x01.connection,
61			message=flow_msg)
62			self.napp.handle_stats_reply(event)
63			mock_from_of_flow_stats_v0x01.assert_called_with(
64			flow_msg.body, self.switch_v0x01.connection.switch)
65
66			desc_msg = MagicMock()
67			desc_msg.body = "A"
68			desc_msg.body_type = StatsType.OFPST_DESC
69			event = get_kytos_event_mock(source=self.switch_v0x01.connection,
70			message=desc_msg)
71			self.napp.handle_stats_reply(event)
72			mock_update_description.assert_called_with(desc_msg.body)
73
74			@patch('kytos.core.switch.Switch.update_description')
75			@patch('napps.kytos.of_core.main.Main._handle_multipart_flow_stats')
76			@patch('napps.kytos.of_core.v0x04.utils.handle_port_desc')
77			def test_handle_multipart_reply(self, *args):
78			"""Test handle multipart reply."""
79			(mock_of_core_v0x04_utils, mock_from_of_flow_stats_v0x04,
80			mock_update_description) = args
81
82			flow_msg = MagicMock()
83			flow_msg.multipart_type = MultipartType.OFPMP_FLOW
84			event = get_kytos_event_mock(source=self.switch_v0x01.connection,
85			message=flow_msg)
86
87			self.napp.handle_multipart_reply(event)
88			mock_from_of_flow_stats_v0x04.assert_called_with(
89			flow_msg, self.switch_v0x01.connection.switch)
90
91			ofpmp_port_desc = MagicMock()
92			ofpmp_port_desc.body = "A"
93			ofpmp_port_desc.multipart_type = MultipartType.OFPMP_PORT_DESC
94			event = get_kytos_event_mock(source=self.switch_v0x01.connection,
95			message=ofpmp_port_desc)
96			self.napp.handle_multipart_reply(event)
97			mock_of_core_v0x04_utils.assert_called_with(
98			self.napp.controller, self.switch_v0x01.connection.switch,
99			ofpmp_port_desc.body)
100
101			ofpmp_desc = MagicMock()
102			ofpmp_desc.body = "A"
103			ofpmp_desc.multipart_type = MultipartType.OFPMP_DESC
104			event = get_kytos_event_mock(source=self.switch_v0x01.connection,
105			message=ofpmp_desc)
106			self.napp.handle_multipart_reply(event)
107			mock_update_description.assert_called_with(ofpmp_desc.body)
108
109			@patch('kytos.core.buffers.KytosEventBuffer.put')
110			@patch('napps.kytos.of_core.v0x04.utils.send_set_config')
111			@patch('napps.kytos.of_core.v0x01.utils.send_set_config')
112			@patch('napps.kytos.of_core.v0x04.utils.send_desc_request')
113			@patch('napps.kytos.of_core.v0x01.utils.send_desc_request')
114			@patch('kytos.core.connection.Connection.set_established_state')
115			@patch('kytos.core.connection.Connection.is_during_setup')
116			@patch('napps.kytos.of_core.v0x04.utils.handle_features_reply')
117			@patch('napps.kytos.of_core.v0x01.utils.handle_features_reply')
118			def test_handle_features_reply(self, *args):
119			"""Test handle features reply."""
120			(mock_freply_v0x01, mock_freply_v0x04, mock_is_during_setup,
121			mock_set_established_state, mock_send_desc_request_v0x01,
122			mock_send_desc_request_v0x04, mock_send_set_config_v0x01,
123			mock_send_set_config_v0x04, mock_buffers_put) = args
124			mock_freply_v0x01.return_value = self.switch_v0x01.connection.switch
125			mock_freply_v0x04.return_value = self.switch_v0x04.connection.switch
126			mock_is_during_setup.return_value = True
127
128			self.switch_v0x01.connection.state = ConnectionState.SETUP
129			self.switch_v0x01.connection.protocol.state = 'waiting_features_reply'
130			event = get_kytos_event_mock(source=self.switch_v0x01.connection)
131			self.napp.handle_features_reply(event)
132			mock_freply_v0x01.assert_called_with(self.napp.controller, event)
133			mock_send_desc_request_v0x01.assert_called_with(
134			self.napp.controller, self.switch_v0x01.connection.switch)
135			mock_send_set_config_v0x01.assert_called_with(
136			self.napp.controller, self.switch_v0x01.connection.switch)
137
138			self.switch_v0x04.connection.state = ConnectionState.SETUP
139			self.switch_v0x04.connection.protocol.state = 'waiting_features_reply'
140			event = get_kytos_event_mock(source=self.switch_v0x04.connection)
141			self.napp.handle_features_reply(event)
142			mock_freply_v0x04.assert_called_with(self.napp.controller, event)
143			mock_send_desc_request_v0x04.assert_called_with(
144			self.napp.controller, self.switch_v0x04.connection.switch)
145			mock_send_set_config_v0x04.assert_called_with(
146			self.napp.controller, self.switch_v0x04.connection.switch)
147
148			mock_is_during_setup.assert_called()
149			mock_set_established_state.assert_called()
150			mock_buffers_put.assert_called()
151
152			@patch('napps.kytos.of_core.main.Main._update_switch_flows')
153			@patch('napps.kytos.of_core.v0x04.flow.Flow.from_of_flow_stats')
154			@patch('napps.kytos.of_core.main.Main._is_multipart_reply_ours')
155			def test_handle_multipart_flow_stats(self, *args):
156			"""Test handle multipart flow stats."""
157			(mock_is_multipart_reply_ours, mock_from_of_flow_stats_v0x01,
158			mock_update_switch_flows) = args
159			mock_is_multipart_reply_ours.return_value = True
160			mock_from_of_flow_stats_v0x01.return_value = "ABC"
161
162			flow_msg = MagicMock()
163			flow_msg.body = "A"
164			flow_msg.flags.value = 2
165			flow_msg.body_type = StatsType.OFPST_FLOW
166
167			self.napp._handle_multipart_flow_stats(flow_msg, self.switch_v0x04)
168
169			mock_is_multipart_reply_ours.assert_called_with(flow_msg,
170			self.switch_v0x04)
171			mock_from_of_flow_stats_v0x01.assert_called_with(flow_msg.body,
172			self.switch_v0x04)
173			mock_update_switch_flows.assert_called_with(self.switch_v0x04)
174
175			@patch('napps.kytos.of_core.main.Main.update_port_status')
176			@patch('napps.kytos.of_core.main.Main.update_links')
177			def test_emit_message_in(self, *args):
178			"""Test emit_message_in."""
179			(mock_update_links, mock_update_port_status) = args
180
181			mock_port_connection = MagicMock()
182			msg_port_mock = MagicMock()
183			msg_port_mock.header.message_type.name = 'ofpt_port_status'
184			mock_port_connection.side_effect = True
185			self.napp.emit_message_in(mock_port_connection,
186			msg_port_mock)
187			mock_update_port_status.assert_called_with(msg_port_mock,
188			mock_port_connection)
189
190			mock_packet_in_connection = MagicMock()
191			msg_packet_in_mock = MagicMock()
192			mock_packet_in_connection.side_effect = True
193			msg_packet_in_mock.header.message_type.name = 'ofpt_packet_in'
194			self.napp.emit_message_in(mock_packet_in_connection,
195			msg_packet_in_mock)
196			mock_update_links.assert_called_with(msg_packet_in_mock,
197			mock_packet_in_connection)
198
199			@patch('pyof.utils.v0x04.symmetric.echo_reply.EchoReply')
200			@patch('napps.kytos.of_core.main.Main.emit_message_out')
201			def test_handle_echo_request(self, *args):
202			"""Test handle echo request messages."""
203			(mock_emit_message_out, mock_echo_reply) = args
204			mock_event = MagicMock()
205			mock_echo_request = MagicMock()
206			mock_echo_reply.return_value = "A"
207			mock_echo_request.header.xid = "A"
208			mock_echo_request.data = "A"
209			mock_event.source.protocol.version = 4
210			mock_event.message = mock_echo_request
211			self.napp.handle_echo_request(mock_event)
212			mock_echo_reply.assert_called_with(xid=mock_echo_request.header.xid,
213			data=mock_echo_request.data)
214			mock_emit_message_out.assert_called_with(mock_event.source, "A")
215
216			@patch('napps.kytos.of_core.main.Main.send_features_request')
217			@patch('napps.kytos.of_core.v0x04.utils.say_hello')
218			@patch('napps.kytos.of_core.main._get_version_from_bitmask')
219			@patch('napps.kytos.of_core.main._get_version_from_header')
220			def test_negotiate(self, *args):
221			"""Test negotiate."""
222			(mock_version_header, mock_version_bitmask, mock_say_hello,
223			mock_features_request) = args
224			mock_version_header.return_value = 4
225			mock_version_bitmask.return_value = 4
226			mock_connection = MagicMock()
227			mock_message = MagicMock()
228			mock_message.versions = 4
229			self.napp._negotiate(mock_connection, mock_message)
230			mock_version_bitmask.assert_called_with(mock_message.versions)
231			mock_say_hello.assert_called_with(self.napp.controller,
232			mock_connection)
233			mock_features_request.assert_called_with(mock_connection)
234
235			@patch('pyof.utils.v0x04.asynchronous.error_msg.ErrorMsg')
236			@patch('napps.kytos.of_core.main.Main.emit_message_out')
237			@patch('kytos.core.buffers.KytosEventBuffer.put')
238			def tests_fail_negotiation(self, *args):
239			"""Test fail_negotiation."""
240			(mock_event_buffer, mock_emit_message_out,
241			mock_error_msg) = args
242			mock_connection = MagicMock()
243			mock_message = MagicMock()
244			mock_connection.id = "A"
245			mock_message.side_effect = 4
246			self.napp.fail_negotiation(mock_connection, mock_message)
247			mock_event_buffer.assert_called()
248			mock_emit_message_out.assert_called_with(mock_connection,
249			mock_error_msg.return_value)
250
251			@patch('napps.kytos.of_core.settings.SEND_FEATURES_REQUEST_ON_ECHO')
252			@patch('napps.kytos.of_core.main.Main.send_features_request')
253			def test_handle_queued_openflow_echo_reply(self, *args):
254			"""Test handle queued OpenFlow echo reply messages."""
255			(mock_send_features_request, mock_settings) = args
256			mock_settings.return_value = True
257			mock_event = MagicMock()
258			self.napp.handle_queued_openflow_echo_reply(mock_event)
259			mock_send_features_request.assert_called_with(mock_event.destination)
260
261			@patch('pyof.utils.v0x04.controller2switch.'
262			'features_request.FeaturesRequest')
263			@patch('napps.kytos.of_core.main.Main.emit_message_out')
264			def test_send_features_request(self, *args):
265			"""Test send send_features_request."""
266			(mock_emit_message_out, mock_features_request) = args
267			mock_destination = MagicMock()
268			mock_destination.protocol.version = 4
269			mock_features_request.return_value = "A"
270			self.napp.send_features_request(mock_destination)
271			mock_features_request.assert_called()
272			mock_emit_message_out.assert_called_with(mock_destination, "A")
273
274			@patch('kytos.core.buffers.KytosEventBuffer.put')
275			@patch('napps.kytos.of_core.main.Ethernet')
276			def test_update_links(self, *args):
277			"""Test update_links."""
278			(mock_ethernet, mock_buffer_put) = args
279			ethernet = create_autospec(Ethernet)
280			ethernet.ether_type = "A"
281			mock_ethernet.side_effect = ethernet
282			mock_message = MagicMock()
283			mock_source = MagicMock()
284			self.napp.update_links(mock_message, mock_source)
285			mock_ethernet.assert_called()
286			mock_buffer_put.assert_called()
287
288			@patch('kytos.core.buffers.KytosEventBuffer.put')
289			def test_send_specific_port_mod(self, mock_buffer_put):
290			"""Test send specific port."""
291			mock_port = MagicMock()
292			mock_interface = MagicMock()
293			current_state = 2
294			self.napp._send_specific_port_mod(mock_port,
295			mock_interface, current_state)
296			mock_buffer_put.assert_called()
297
298			@patch('kytos.core.buffers.KytosEventBuffer.put')
299			@patch('napps.kytos.of_core.main.Interface')
300			@patch('napps.kytos.of_core.main.Main._send_specific_port_mod')
301			def test_update_port_status(self, *args):
302			"""Test update_port_status."""
303			(mock_port_mod, mock_interface, mock_buffer_put) = args
304			mock_port_status = MagicMock()
305			mock_source = MagicMock()
306			mock_port_status.reason.value.side_effect = [0, 1]
307			mock_port_status.reason.enum_ref(0).name = 'OFPPR_ADD'
308			self.napp.update_port_status(mock_port_status, mock_source)
309			mock_interface.assert_called()
310
311			# check OFPRR_MODIFY
312			mock_port_status.reason.enum_ref(1).name = 'OFPPR_MODIFY'
313			self.napp.update_port_status(mock_port_status, mock_source)
314			mock_port_mod.assert_called()
315			mock_buffer_put.assert_called()
316

kytos / of_core

Push — master ( 1b459e...75d007 )

TestMain.test_negotiate() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like