Test Failed
Pull Request — master (#62)
by Gleyberson
03:47
created

build.tests.test_main   A

Complexity

Total Complexity 17

Size/Duplication

Total Lines 317
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 241
dl 0
loc 317
rs 10
c 0
b 0
f 0
wmc 17

16 Methods

Rating   Name   Duplication   Size   Complexity  
A TestMain.test_handle_01_features_reply() 0 23 1
A TestMain.test_handle_port_status_raw_in() 0 17 1
A TestMain.test_handle_echo_request() 0 13 1
A TestMain._get_interface() 0 7 1
A TestMain.test_handle_multipart_reply() 0 37 1
A TestMain.test_handle_packet_in_raw_in() 0 21 1
A TestMain.test_handle_features_request_sent() 0 14 1
A TestMain.test_handle_stats_reply() 0 21 1
A TestMain.test_handle_04_features_reply() 0 17 1
A TestMain.test_handle_hello_raw_in() 0 12 1
A TestMain._get_switch_mock() 0 13 1
A TestMain.test_get_event_listeners() 0 16 2
A TestMain.test_handle_port_desc_multipart_reply() 0 36 1
A TestMain.setUp() 0 9 1
A TestMain._get_controller_mock() 0 7 1
A TestMain.test_execute() 0 11 1
1
"""Test main."""
2
3
from unittest import TestCase
4
from unittest.mock import Mock
5
6
from pyof.utils import unpack
7
from pyof.v0x01.controller2switch.features_reply import \
8
    FeaturesReply as FReply_v0x01
9
from pyof.v0x01.controller2switch.stats_reply import StatsReply
10
from pyof.v0x04.controller2switch.features_reply import \
11
    FeaturesReply as FReply_v0x04
12
from pyof.v0x04.controller2switch.features_request import FeaturesRequest
13
from pyof.v0x04.controller2switch.multipart_reply import MultipartReply
14
from pyof.v0x04.symmetric.echo_request import EchoRequest
15
16
from kytos.core import Controller
17
from kytos.core.config import KytosConfig
18
from kytos.core.connection import Connection, ConnectionState
19
from kytos.core.events import KytosEvent
20
from kytos.core.interface import Interface
21
from kytos.core.switch import Switch
22
from napps.kytos.of_core.main import Main
23
24
25
class TestMain(TestCase):
26
    """docstring for TestMain."""
27
28
    def setUp(self):
29
        """Execute steps before each tests.
30
31
        Set the server_name_url from kytos/of_core
32
        """
33
        self.server_name_url = 'http://localhost:8181/api/kytos/of_core'
34
        self.controller = self._get_controller_mock()
35
        self.napp = Main(self.controller)
36
        self.patched_events = []
37
38
    @staticmethod
39
    def _get_controller_mock():
40
        """Return a controller mock."""
41
        options = KytosConfig().options['daemon']
42
        controller = Controller(options)
43
        controller.log = Mock()
44
        return controller
45
46
    @staticmethod
47
    def _get_switch_mock(of_version, connection_state=ConnectionState.NEW,
48
                         dpid="00:00:00:00:00:00:00:01"):
49
        """Return a switch mock."""
50
        switch = Switch(dpid)
51
        address = Mock()
52
        port = Mock()
53
        socket = Mock()
54
        switch.connection = Connection(address, port, socket)
55
        switch.connection.protocol.unpack = unpack
56
        switch.connection.protocol.version = of_version
57
        switch.connection.state = connection_state
58
        return switch
59
60
    def _get_interface(self, interface_name, port, *args, **kwargs):
61
        """Return a interface mock."""
62
        switch1 = self._get_switch_mock
63
        switch1.connection = Mock()
64
        switch1.connection.protocol.version = 0x04
65
        iface1 = Interface(interface_name, port, switch1, *args, **kwargs)
66
        return iface1
67
68
    def test_get_event_listeners(self):
69
        """Verify all event listeners registered."""
70
        expected_events = [
71
            'kytos/of_core.v0x01.messages.in.ofpt_stats_reply',
72
            'kytos/of_core.v0x0[14].messages.in.ofpt_features_reply',
73
            'kytos/of_core.v0x04.messages.in.ofpt_multipart_reply',
74
            'kytos/core.openflow.raw.in',
75
            'kytos/of_core.v0x0[14].messages.in.ofpt_echo_request',
76
            'kytos/of_core.v0x0[14].messages.out.ofpt_echo_reply',
77
            'kytos/of_core.v0x[0-9a-f]{2}.messages.in.hello_failed',
78
            'kytos/of_core.v0x0[14].messages.out.hello_failed',
79
        ]
80
81
        actual_events = self.napp.listeners()
82
        for _event in expected_events:
83
            self.assertIn(_event, actual_events, '%s' % _event)
84
85
    def test_execute(self):
86
        """Test 'execute' main method."""
87
        dpid_01 = "00:00:00:00:00:00:00:01"
88
        sw_01 = self._get_switch_mock(0x01, ConnectionState.ESTABLISHED,
89
                                      dpid_01)
90
        self.napp.controller.get_switch_or_create(dpid_01, sw_01.connection)
91
        dpid_02 = "00:00:00:00:00:00:00:02"
92
        sw_04 = self._get_switch_mock(0x04, ConnectionState.ESTABLISHED,
93
                                      dpid_02)
94
        self.napp.controller.get_switch_or_create(dpid_02, sw_04.connection)
95
        self.napp.execute()
96
97
    def test_handle_stats_reply(self):
98
        """Test handling stats reply message."""
99
        event_name = 'kytos/of_core.v0x01.messages.in.ofpt_stats_reply'
100
        switch = self._get_switch_mock(0x01)
101
        switch.connection = Mock()
102
103
        stats_data = b'\x01\x11\x00\x0c\x00\x00\x00\x01\x00\x01\x00\x01'
104
        stats_reply = StatsReply()
105
        stats_reply.unpack(stats_data[8:])
106
        stats_event = KytosEvent(name=event_name,
107
                                 content={'source': switch.connection,
108
                                          'message': stats_reply})
109
        self.napp.handle_stats_reply(stats_event)
110
111
        desc_stats_data = b'\x01\x11\x00\x0c\x00\x00\x00\x0e\x00\x00\x00\x00'
112
        desc_stats_reply = StatsReply()
113
        desc_stats_reply.unpack(desc_stats_data[8:])
114
        desc_stats_event = KytosEvent(name=event_name,
115
                                      content={'source': switch.connection,
116
                                               'message': desc_stats_reply})
117
        self.napp.handle_stats_reply(desc_stats_event)
118
119
    def test_handle_04_features_reply(self):
120
        """Test handling features reply message."""
121
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_features_reply'
122
        switch = self._get_switch_mock(0x04, ConnectionState.SETUP)
123
        switch.connection.protocol.state = 'waiting_features_reply'
124
125
        data = b'\x04\x06\x00\x20\x00\x00\x00\x00\x00\x00\x08\x60\x6e\x7f\x74'
126
        data += b'\xe7\x00\x00\x00\x00\xff\x63\x00\x00\x00\x00\x00\x4f\x00\x00'
127
        data += b'\x00\x00'
128
129
        features_reply = FReply_v0x04()
130
        features_reply.unpack(data)
131
132
        event = KytosEvent(name=event_name,
133
                           content={'source': switch.connection,
134
                                    'message': features_reply})
135
        self.napp.handle_features_reply(event)
136
137
    def test_handle_01_features_reply(self):
138
        """Test handling features reply message."""
139
        event_name = 'kytos/of_core.v0x01.messages.in.ofpt_features_reply'
140
        switch = self._get_switch_mock(0x01, ConnectionState.SETUP)
141
        switch.connection.protocol.state = 'waiting_features_reply'
142
143
        data = b'\x01\x06\x00\x80\x00\x00\x00\x00\x00\x00\x00\xff\x12\x34\x56'
144
        data += b'\x78\x00\x00\x00\x00\xff\x00\x00\x00\x00\x00\x00\xa9\x00\x00'
145
        data += b'\x08\x43\x00\x07\xf2\x0b\xa4\xd0\x3f\x70\x50\x6f\x72\x74\x37'
146
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
147
        data += b'\x00\x00\x00\x10\x00\x00\x02\x88\x00\x00\x02\x80\x00\x00\x02'
148
        data += b'\x88\x00\x00\x02\x88\x00\x06\xf2\x0b\xa4\x7d\xf8\xea\x50\x6f'
149
        data += b'\x72\x74\x36\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
150
        data += b'\x00\x00\x00\x00\x00\x00\x02\x00\x00\x02\x88\x00\x00\x02\x80'
151
        data += b'\x00\x00\x02\x88\x00\x00\x02\x88'
152
153
        features_reply = FReply_v0x01()
154
        features_reply.unpack(data[8:])
155
156
        event = KytosEvent(name=event_name,
157
                           content={'source': switch.connection,
158
                                    'message': features_reply})
159
        self.napp.handle_features_reply(event)
160
161
    def test_handle_features_request_sent(self):
162
        """Test handling features request sent message."""
163
        event_name = 'kytos/of_core.v0x01.messages.out.ofpt_features_request'
164
        switch = self._get_switch_mock(0x01)
165
        switch.connection.protocol.state = 'sending_features'
166
167
        data = b'\x04\x05\x00\x08\x00\x00\x00\x03'
168
        features_request = FeaturesRequest()
169
        features_request.unpack(data)
170
171
        event = KytosEvent(name=event_name,
172
                           content={'destination': switch.connection,
173
                                    'message': features_request})
174
        self.napp.handle_features_request_sent(event)
175
176
    def test_handle_echo_request(self):
177
        """Test handling echo request message."""
178
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_echo_request'
179
        switch = self._get_switch_mock(0x04)
180
181
        data = b'\x04\x02\x00\x0c\x00\x00\x00\x00\x68\x6f\x67\x65'
182
        echo_request = EchoRequest()
183
        echo_request.unpack(data)
184
185
        event = KytosEvent(name=event_name,
186
                           content={'source': switch.connection,
187
                                    'message': echo_request})
188
        self.napp.handle_echo_request(event)
189
190
    def test_handle_hello_raw_in(self):
191
        """Test handling hello raw in message."""
192
        event_name = 'kytos/core.openflow.raw.in'
193
        switch = self._get_switch_mock(0x04)
194
195
        data = b'\x04\x00\x00\x10\x00\x00\x00\x3e'
196
        data += b'\x00\x01\x00\x08\x00\x00\x00\x10'
197
198
        event = KytosEvent(name=event_name,
199
                           content={'source': switch.connection,
200
                                    'new_data': data})
201
        self.napp.handle_raw_in(event)
202
203
    def test_handle_port_status_raw_in(self):
204
        """Test handling port_status raw in message."""
205
        event_name = 'kytos/core.openflow.raw.in'
206
        switch = self._get_switch_mock(0x04, ConnectionState.ESTABLISHED)
207
        switch.connection.switch = self._get_switch_mock(0x04)
208
209
        data = b'\x04\x0c\x00\x50\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00'
210
        data += b'\x00\x00\x00\x00\x01\x00\x00\x00\x00\x62\x43\xe5\xdb\x35\x0a'
211
        data += b'\x00\x00\x73\x31\x2d\x65\x74\x68\x31\x00\x00\x00\x00\x00\x00'
212
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00\x00\x08\x40'
213
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x98\x96'
214
        data += b'\x80\x00\x00\x00\x00'
215
216
        event = KytosEvent(name=event_name,
217
                           content={'source': switch.connection,
218
                                    'new_data': data})
219
        self.napp.handle_raw_in(event)
220
221
    def test_handle_packet_in_raw_in(self):
222
        """Test handling packet_in raw in message."""
223
        event_name = 'kytos/core.openflow.raw.in'
224
        switch = self._get_switch_mock(0x04, ConnectionState.ESTABLISHED)
225
        switch.connection.switch = self._get_switch_mock(0x04)
226
227
        data = b'\x04\x0a\x00\x94\x00\x00\x00\x00\x00\x00\x00\x02\x00\x2a\x01'
228
        data += b'\x01\x00\x01\x02\x03\x00\x00\x00\x00\x00\x01\x00\x50\x80\x00'
229
        data += b'\x00\x04\x00\x00\x00\x06\x80\x00\x0a\x02\x08\x06\x80\x00\x06'
230
        data += b'\x06\xff\xff\xff\xff\xff\xff\x80\x00\x08\x06\xf2\x0b\xa4\x7d'
231
        data += b'\xf8\xea\x80\x00\x2a\x02\x00\x01\x80\x00\x2c\x04\x0a\x00\x00'
232
        data += b'\x01\x80\x00\x2e\x04\x0a\x00\x00\x03\x80\x00\x30\x06\xf2\x0b'
233
        data += b'\xa4\x7d\xf8\xea\x80\x00\x32\x06\x00\x00\x00\x00\x00\x00\x00'
234
        data += b'\x00\xff\xff\xff\xff\xff\xff\xf2\x0b\xa4\x7d\xf8\xea\x08\x06'
235
        data += b'\x00\x01\x08\x00\x06\x04\x00\x01\xf2\x0b\xa4\x7d\xf8\xea\x0a'
236
        data += b'\x00\x00\x01\x00\x00\x00\x00\x00\x00\x0a\x00\x00\x03'
237
238
        event = KytosEvent(name=event_name,
239
                           content={'source': switch.connection,
240
                                    'new_data': data})
241
        self.napp.handle_raw_in(event)
242
243
    def test_handle_multipart_reply(self):
244
        """Test handling ofpt_multipart_reply."""
245
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_multipart_reply'
246
        switch = self._get_switch_mock(0x04, dpid='00:00:00:00:00:00:00:02')
247
        switch.connection.switch = self._get_switch_mock(0x04)
248
        self.napp.controller.get_switch_or_create(switch.dpid,
249
                                                  switch.connection)
250
251
        data = b'\x04\x13\x00\x68\xac\xc8\xdf\x58\x00\x01\x00\x00\x00\x00\x00'
252
        data += b'\x00\x00\x58\x00\x00\x00\x00\x00\x38\x25\xd9\x54\xc0\x03\xe8'
253
        data += b'\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00'
254
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x12\x00\x00\x00\x00'
255
        data += b'\x00\x00\x02\xf4\x00\x01\x00\x10\x80\x00\x0a\x02\x88\xcc\x80'
256
        data += b'\x00\x0c\x02\x1e\xd7\x00\x04\x00\x18\x00\x00\x00\x00\x00\x00'
257
        data += b'\x00\x10\xff\xff\xff\xfd\xff\xff\x00\x00\x00\x00\x00\x00'
258
259
        # pylint: disable=protected-access
260
        xid = self.napp._multipart_replies_xids[switch.dpid]
261
        # pylint: enable=protected-access
262
        multipart_reply = MultipartReply(xid=xid)
263
        multipart_reply.unpack(data[8:])
264
        stats_event = KytosEvent(name=event_name,
265
                                 content={'source': switch.connection,
266
                                          'message': multipart_reply})
267
268
        self.napp.handle_multipart_reply(stats_event)
269
270
        # test ofpmp_desc
271
        data = b'\x04\x12\x00\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
272
        data += b'\x00'
273
        multipart_desc = MultipartReply()
274
        multipart_desc.unpack(data[8:])
275
        stats_desc_event = KytosEvent(name=event_name,
276
                                      content={'source': switch.connection,
277
                                               'message': multipart_desc})
278
279
        self.napp.handle_multipart_reply(stats_desc_event)
280
281
    def test_handle_port_desc_multipart_reply(self):
282
        """Test handling to ofpt_PORT_DESC."""
283
284
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_multipart_reply'
285
        switch = self._get_switch_mock(0x04)
286
        switch.connection.switch = self._get_switch_mock(0x04)
287
        data = b'\x04\x13\x00\x90\x00\x00\x00\x00\x00\x0d\x00\x00\x00\x00\x00'
288
        data += b'\x00\x00\x00\x00\x07\x00\x00\x00\x00\xf2\x0b\xa4\xd0\x3f\x70'
289
        data += b'\x00\x00\x50\x6f\x72\x74\x37\x00\x00\x00\x00\x00\x00\x00\x00'
290
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00\x00\x28\x08'
291
        data += b'\x00\x00\x28\x00\x00\x00\x28\x08\x00\x00\x28\x08\x00\x00\x13'
292
        data += b'\x88\x00\x00\x13\x88\x00\x00\x00\x06\x00\x00\x00\x00\xf2\x0b'
293
        data += b'\xa4\x7d\xf8\xea\x00\x00\x50\x6f\x72\x74\x36\x00\x00\x00\x00'
294
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04'
295
        data += b'\x00\x00\x28\x08\x00\x00\x28\x00\x00\x00\x28\x08\x00\x00\x28'
296
        data += b'\x08\x00\x00\x13\x88\x00\x00\x13\x88'
297
298
        port_desc = MultipartReply()
299
        port_desc.unpack(data[8:])
300
        interface_1 = self._get_interface("interface1", 6)
301
        interface_2 = self._get_interface("interface2", 7)
302
        switch.connection.switch.interfaces = {6: interface_1, 7: interface_2}
303
304
        stats_event = KytosEvent(name=event_name,
305
                                 content={'source': switch.connection,
306
                                          'message': port_desc})
307
        self.napp.handle_multipart_reply(stats_event)
308
309
        # Send port_desc pack without interface
310
        switch = self._get_switch_mock(0x04)
311
        switch.connection.switch = self._get_switch_mock(0x04)
312
        stats_event = KytosEvent(name=event_name,
313
                                 content={'source': switch.connection,
314
                                          'message': port_desc})
315
316
        self.napp.handle_multipart_reply(stats_event)
317