1
|
|
|
""" Tests for /backends/openflow13.py """ |
2
|
|
|
from unittest import TestCase |
3
|
|
|
from unittest.mock import MagicMock, patch |
4
|
|
|
|
5
|
|
|
from napps.amlight.sdntrace import settings |
6
|
|
|
from napps.amlight.sdntrace.backends.openflow13 import ( |
7
|
|
|
packet_in, |
8
|
|
|
send_packet_out, |
9
|
|
|
) |
10
|
|
|
|
11
|
|
|
|
12
|
|
|
# pylint: disable=too-many-public-methods, too-many-lines |
13
|
|
|
class TestOpenflow13(TestCase): |
14
|
|
|
"""Unit tests for backends.openflow13 functions""" |
15
|
|
|
|
16
|
|
|
def test_packet_in_with_color(self): |
17
|
|
|
"""Test packet in with color ee:ee:ee:ee:ee:01.""" |
18
|
|
|
raw = b"\x00\x15\xaf\xd58\x98\xee\xee\xee\xee\xee\x01\x08\x00testdata" |
19
|
|
|
# Ethernet(destination='00:15:af:d5:38:98', |
20
|
|
|
# source='ee:ee:ee:ee:ee:01', ether_type=0x800, |
21
|
|
|
# data=b'testdata') |
22
|
|
|
packet_in_msg = MagicMock() |
23
|
|
|
packet_in_msg.data.value = raw |
24
|
|
|
packet_in_msg.in_port.value = 1 |
25
|
|
|
|
26
|
|
|
event = MagicMock() |
27
|
|
|
event.source.switch = "ee:ee:ee:ee:ee:01" |
28
|
|
|
event.message.in_port = 1 |
29
|
|
|
|
30
|
|
|
ethernet, in_port, switch = packet_in(event, packet_in_msg) |
31
|
|
|
|
32
|
|
|
self.assertEqual(ethernet.source.value, "ee:ee:ee:ee:ee:01") |
33
|
|
|
self.assertEqual(in_port, 1) |
34
|
|
|
self.assertEqual(switch, "ee:ee:ee:ee:ee:01") |
35
|
|
|
|
36
|
|
|
def test_normal_packet_in(self): |
37
|
|
|
"""Test packet in without color.""" |
38
|
|
|
raw = b"\x00\x15\xaf\xd58\x98\x00\x1f:>\x9a\xcf\x08\x00testdata" |
39
|
|
|
# Ethernet(destination='00:15:af:d5:38:98', |
40
|
|
|
# source='00:1f:3a:3e:9a:cf', ether_type=0x800, |
41
|
|
|
# data=b'testdata') |
42
|
|
|
packet_in_msg = MagicMock() |
43
|
|
|
packet_in_msg.data.value = raw |
44
|
|
|
packet_in_msg.in_port.value = 1 |
45
|
|
|
|
46
|
|
|
event = MagicMock() |
47
|
|
|
event.source.switch = "00:1f:3a:3e:9a:cf" |
48
|
|
|
|
49
|
|
|
ethernet, in_port, switch = packet_in(event, packet_in_msg) |
50
|
|
|
|
51
|
|
|
self.assertEqual(ethernet, 0) |
52
|
|
|
self.assertEqual(in_port, 0) |
53
|
|
|
self.assertEqual(switch, 0) |
54
|
|
|
|
55
|
|
|
def test_packet_out_with_color(self): |
56
|
|
|
"""Test packet in with color ee:ee:ee:ee:ee:01.""" |
57
|
|
|
data = b"\x00\x15\xaf\xd58\x98\xee\xee\xee\xee\xee\x01\x08\x00testdata" |
58
|
|
|
# Ethernet(destination='00:15:af:d5:38:98', |
59
|
|
|
# source='ee:ee:ee:ee:ee:01', ether_type=0x800, |
60
|
|
|
# data=b'testdata') |
61
|
|
|
controller = MagicMock() |
62
|
|
|
switch = MagicMock() |
63
|
|
|
switch.connection.value = "00:15:af:d5:38:98" |
64
|
|
|
port = 1 |
65
|
|
|
|
66
|
|
|
send_packet_out(controller, switch, port, data) |
67
|
|
|
|
68
|
|
|
# Verify that the controller sent the packet_out |
69
|
|
|
controller.buffers.msg_out.put.assert_called_once() |
70
|
|
|
called_event = controller.buffers.msg_out.put.call_args.args[0] |
71
|
|
|
|
72
|
|
|
# Verify the packet_out values |
73
|
|
|
self.assertEqual( |
74
|
|
|
called_event.name, "kytos/of_lldp.messages.out.ofpt_packet_out" |
75
|
|
|
) |
76
|
|
|
self.assertEqual( |
77
|
|
|
called_event.content["destination"].value, switch.connection.value |
78
|
|
|
) |
79
|
|
|
self.assertEqual(called_event.content["message"].in_port, port) |
80
|
|
|
self.assertEqual(called_event.content["message"].data, data) |
81
|
|
|
self.assertEqual( |
82
|
|
|
called_event.content["message"].actions[0].port, |
83
|
|
|
settings.OFPP_TABLE_13, |
84
|
|
|
) |
85
|
|
|
|
86
|
|
|
@patch("napps.amlight.sdntrace.backends.openflow13.of_msg_prio") |
87
|
|
|
def test_send_packet_out(self, mock_of_msg_prio) -> None: |
88
|
|
|
"""Test send_packet_out.""" |
89
|
|
|
controller = MagicMock() |
90
|
|
|
send_packet_out(controller, MagicMock(), MagicMock(), MagicMock()) |
91
|
|
|
assert controller.buffers.msg_out.put.call_count == 1 |
92
|
|
|
assert mock_of_msg_prio.call_count == 1 |
93
|
|
|
|