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

build.tests.test_main   A

Complexity

Total Complexity 17

Size/Duplication

Total Lines 312
Duplicated Lines 0 %

Importance

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