1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
"""group_mod tests."""
from unittest import TestCase
from pyof.v0x04.controller2switch.group_mod import GroupMod
from pyof.v0x04.common.action import (
ActionExperimenter, ActionSetField, ListOfActions)
from pyof.v0x04.common.flow_match import OxmClass, OxmOfbMatchField, OxmTLV
from pyof.v0x04.common.port import PortNo
from pyof.v0x04.controller2switch.common import Bucket
from pyof.v0x04.controller2switch.group_mod import ListOfBuckets
class TestGroupMod(TestCase):
"""group_mod tests."""
def test_min_size(self):
"""Test minimum struct size."""
self.assertEqual(16, GroupMod().get_size())
class TestBucket(TestCase):
"""bucket tests."""
def test_min_size(self):
"""Test minimum struct size."""
self.assertEqual(16, Bucket().get_size())
class TestListBuckets(TestCase):
def setUp(self):
"""Configure raw file and its object in parent class (TestDump)."""
super().setUp()
self.oxmtlv1 = OxmTLV(oxm_class=OxmClass.OFPXMC_OPENFLOW_BASIC,
oxm_field=OxmOfbMatchField.OFPXMT_OFB_METADATA,
oxm_hasmask=False,
oxm_value=b'\x00\x00\x00\x00\x00\x00\x00\x01')
self.oxmtlv2 = OxmTLV(oxm_class=OxmClass.OFPXMC_OPENFLOW_BASIC,
oxm_field=OxmOfbMatchField.OFPXMT_OFB_METADATA,
oxm_hasmask=False,
oxm_value=b'\x00\x00\x00\x00\x00\x00\x00\x02')
self.action1 = ActionSetField(field=self.oxmtlv1)
self.action2 = ActionSetField(field=self.oxmtlv2)
self.action3 = ActionExperimenter(length=16, experimenter=0x00002320,
body=b'\x00\x0e\xff\xf8\x28\x00\x00\x00')
self.action4 = ActionExperimenter(length=16, experimenter=0x00001223,
body=b'\x00\x0e\xff\xff\x28\x00\x00\x00')
def test_bucket_list(self):
bucket1 = Bucket(length=48, weight=1, watch_port=PortNo.OFPP_ANY,
watch_group=PortNo.OFPP_ANY,
actions=ListOfActions([self.action1, self.action2]))
bucket2 = Bucket(length=80, weight=2, watch_port=PortNo.OFPP_ANY,
watch_group=PortNo.OFPP_ANY,
actions=ListOfActions([self.action1, self.action2,
self.action3, self.action4]))
bucket3 = Bucket(length=48, weight=3, watch_port=PortNo.OFPP_ANY,
watch_group=PortNo.OFPP_ANY,
actions=ListOfActions([self.action3, self.action4]))
# Packing buckets
buckets = ListOfBuckets([bucket1, bucket2, bucket3])
buff = packed_buff = buckets.pack()
# Unpacking buckets bytes
unpacked_buckets = ListOfBuckets()
unpacked_buckets.unpack(buff)
self.assertEqual(len(unpacked_buckets), 3)
self.assertEqual(unpacked_buckets[0].length, 48)
self.assertEqual(unpacked_buckets[0].weight, 1)
self.assertEqual(len(unpacked_buckets[0].actions), 2)
self.assertEqual(unpacked_buckets[0].actions[0].field.oxm_value,
self.oxmtlv1.oxm_value)
self.assertEqual(unpacked_buckets[0].actions[1].field.oxm_value,
self.oxmtlv2.oxm_value)
self.assertEqual(unpacked_buckets[1].length, 80)
self.assertEqual(unpacked_buckets[1].weight, 2)
self.assertEqual(len(unpacked_buckets[1].actions), 4)
self.assertEqual(unpacked_buckets[1].actions[0].field.oxm_value,
self.oxmtlv1.oxm_value)
self.assertEqual(unpacked_buckets[1].actions[1].field.oxm_value,
self.oxmtlv2.oxm_value)
self.assertEqual(unpacked_buckets[1].actions[2].body,
self.action3.body)
self.assertEqual(unpacked_buckets[1].actions[3].body,
self.action4.body)
self.assertEqual(unpacked_buckets[2].length, 48)
self.assertEqual(unpacked_buckets[2].weight, 3)
self.assertEqual(len(unpacked_buckets[2].actions), 2)
self.assertEqual(unpacked_buckets[2].actions[0].body,
self.action3.body)
self.assertEqual(unpacked_buckets[2].actions[1].body,
self.action4.body)
def test_buckets_one_item(self):
bucket1 = Bucket(length=48, weight=1, watch_port=PortNo.OFPP_ANY,
watch_group=PortNo.OFPP_ANY,
actions=ListOfActions([self.action1, self.action2]))
# Packing buckets
buckets = ListOfBuckets([bucket1])
buff = packed_buff = buckets.pack()
# Unpacking buckets bytes
unpacked_buckets = ListOfBuckets()
unpacked_buckets.unpack(buff)
self.assertEqual(len(unpacked_buckets), 1)
self.assertEqual(unpacked_buckets[0].length, 48)
self.assertEqual(unpacked_buckets[0].weight, 1)
self.assertEqual(len(unpacked_buckets[0].actions), 2)
self.assertEqual(unpacked_buckets[0].actions[0].field.oxm_value,
self.oxmtlv1.oxm_value)
self.assertEqual(unpacked_buckets[0].actions[1].field.oxm_value,
self.oxmtlv2.oxm_value)
def test_buckets_no_action(self):
bucket1 = Bucket(length=48, weight=1, watch_port=PortNo.OFPP_ANY,
watch_group=PortNo.OFPP_ANY,
actions=ListOfActions([self.action1]))
# Packing buckets
buckets = ListOfBuckets([bucket1])
buff = packed_buff = buckets.pack()
# Unpacking buckets bytes
unpacked_buckets = ListOfBuckets()
unpacked_buckets.unpack(buff)
self.assertEqual(len(unpacked_buckets), 1)
self.assertEqual(unpacked_buckets[0].length, 48)
self.assertEqual(unpacked_buckets[0].weight, 1)
self.assertEqual(len(unpacked_buckets[0].actions), 1)
self.assertEqual(unpacked_buckets[0].actions[0].field.oxm_value,
self.oxmtlv1.oxm_value)
|
