build.tests.unit.test_flow_serializer_v0x04.TestFlowSerializer13.setUp() - Code Metrics - Inspection of "Merge pull request #81 from gleybersonandrade/unit..." - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 151596...743f3c )

by Humberto

created 2020-07-07 14:49 UTC

TestFlowSerializer13.setUp() A

↳ Parent: build.tests.unit.test_flow_serializer_v0x04

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	2
dl	0
loc	3
rs	10
c	0
b	0
f	0
cc	1
nop	1

"""Test Flow serializer for OF 1.0 methods."""
from unittest import TestCase
from unittest.mock import MagicMock, patch

from napps.kytos.flow_manager.serializers.v0x04 import FlowSerializer13


# pylint: disable=protected-access, unused-argument, no-member
class TestFlowSerializer13(TestCase):
    """Test the FlowSerializer13 class."""

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

    @patch('napps.kytos.flow_manager.serializers.v0x04.ActionOutput')
    @patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
    @patch('napps.kytos.flow_manager.serializers.v0x04.InstructionApplyAction')
    @patch('napps.kytos.flow_manager.serializers.v0x04.FlowMod')
    def test_from_dict(self, *args):
        """Test from_dict method."""
        (mock_flow_mod, mock_instruction, _, _) = args
        flow_mod = MagicMock()
        flow_mod.instructions = []
        flow_mod.match.oxm_match_fields = []
        instruction = MagicMock()
        instruction.actions = []

        mock_flow_mod.return_value = flow_mod
        mock_instruction.return_value = instruction

        self.napp.flow_attributes = {'flow_attr'}
        self.napp._match_names = {'match_attr': 123}
        dictionary = {'flow_attr': 'any_data',
                      'match': {'match_attr': 123},
                      'actions': [{'action_type': 'output',
                                   'port': 'controller'}]}
        return_flow_mod = self.napp.from_dict(dictionary)

        self.assertEqual(return_flow_mod.flow_attr, 'any_data')
        self.assertEqual(len(return_flow_mod.match.oxm_match_fields), 1)
        self.assertEqual(len(return_flow_mod.instructions), 1)
        self.assertEqual(len(return_flow_mod.instructions[0].actions), 1)

    @patch('napps.kytos.flow_manager.serializers.v0x04.IPAddress')
    @patch('napps.kytos.flow_manager.serializers.v0x04.HWAddress')
    @patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
    def test_match_from_dict(self, *args):
        """Test _match_from_dict method."""
        (_, mock_hw_address, mock_ip_address) = args
        hw_address = MagicMock()
        hw_address.pack.return_value = 'hw_address'
        mock_hw_address.return_value = hw_address
        ip_address = MagicMock()
        ip_address.pack.return_value = 'ip_address'
        mock_ip_address.return_value = ip_address

        dictionary = {'dl_vlan_pcp': 0, 'nw_proto': 1, 'dl_vlan': 2,
                      'dl_src': 3, 'dl_dst': 4, 'nw_src': 5, 'nw_dst': 6,
                      'in_port': 7, 'dl_type': 8}

        tlv_generator = self.napp._match_from_dict(dictionary)

        self.assertEqual(next(tlv_generator).oxm_value, b'\x00')
        self.assertEqual(next(tlv_generator).oxm_value, b'\x01')
        self.assertEqual(next(tlv_generator).oxm_value, b'\x10\x02')
        self.assertEqual(next(tlv_generator).oxm_value, 'hw_address')
        self.assertEqual(next(tlv_generator).oxm_value, 'hw_address')
        self.assertEqual(next(tlv_generator).oxm_value, 'ip_address')
        self.assertEqual(next(tlv_generator).oxm_value, 'ip_address')
        self.assertEqual(next(tlv_generator).oxm_value, b'\x00\x00\x00\x07')
        self.assertEqual(next(tlv_generator).oxm_value, b'\x00\x08')

    @patch('napps.kytos.flow_manager.serializers.v0x04.ActionOutput')
    def test_actions_from_list(self, mock_action_output):
        """Test _actions_from_list method."""
        actions = [{'action_type': 'output', 'port': 'controller'}]
        generator = self.napp._actions_from_list(actions)
        return_action = next(generator)

        self.assertEqual(return_action, mock_action_output.return_value)

    @patch('napps.kytos.flow_manager.serializers.v0x04.ActionPopVLAN',
           return_value='action_pop_vlan')
    @patch('napps.kytos.flow_manager.serializers.v0x04.ActionPush',
           return_value='action_push')
    @patch('napps.kytos.flow_manager.serializers.v0x04.ActionOutput',
           return_value='action_output')
    @patch('napps.kytos.flow_manager.serializers.v0x04.ActionSetField',
           return_value='action_set_field')
    @patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
    def test_action_from_dict(self, *args):
        """Test _action_from_dict method."""
        (mock_oxm_tlv, mock_action_set_field, mock_action_output,
         mock_action_push, _) = args
        oxm_tlv = MagicMock()
        mock_oxm_tlv.return_value = oxm_tlv

        action_1 = {'action_type': 'set_vlan', 'vlan_id': 123}
        return_1 = self.napp._action_from_dict(action_1)
        mock_action_set_field.assert_called_with(field=oxm_tlv)
        self.assertEqual(return_1, 'action_set_field')

        action_2 = {'action_type': 'output', 'port': 'controller'}
        return_2 = self.napp._action_from_dict(action_2)
        mock_action_output.assert_called_with(port=0xfffffffd)
        self.assertEqual(return_2, 'action_output')

        action_3 = {'action_type': 'output', 'port': 'any'}
        return_3 = self.napp._action_from_dict(action_3)
        mock_action_output.assert_called_with(port='any')
        self.assertEqual(return_3, 'action_output')

        action_4 = {'action_type': 'push_vlan', 'tag_type': 's'}
        return_4 = self.napp._action_from_dict(action_4)
        mock_action_push.assert_called_with(action_type=17, ethertype=0x88a8)
        self.assertEqual(return_4, 'action_push')

        action_5 = {'action_type': 'push_vlan', 'tag_type': 'c'}
        return_5 = self.napp._action_from_dict(action_5)
        mock_action_push.assert_called_with(action_type=17, ethertype=0x8100)
        self.assertEqual(return_5, 'action_push')

        action_6 = {'action_type': 'pop_vlan'}
        return_6 = self.napp._action_from_dict(action_6)
        self.assertEqual(return_6, 'action_pop_vlan')

    @patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
    def test_create_vlan_tlv(self, mock_oxm_tlv):
        """Test _create_vlan_tlv method."""
        tlv = self.napp._create_vlan_tlv(1)

        self.assertEqual(tlv.oxm_field, 6)
        self.assertEqual(tlv.oxm_value, b'\x10\x01')

    @patch('napps.kytos.flow_manager.serializers.v0x04.FlowSerializer13.'
           '_actions_to_list', return_value='actions_to_list')
    @patch('napps.kytos.flow_manager.serializers.v0x04.FlowSerializer13.'
           '_match_to_dict', return_value='match_to_dict')
    def test_to_dict(self, *args):
        """Test to_dict method."""
        (_, _) = args
        flow_stats = MagicMock()
        flow_stats.flow_attr = MagicMock()
        self.napp.flow_attributes = {'flow_attr'}

        flow_dict = self.napp.to_dict(flow_stats)

        expected = {'flow_attr': flow_stats.flow_attr.value,
                    'match': 'match_to_dict',
                    'actions': 'actions_to_list'}
        self.assertEqual(flow_dict, expected)

    @patch('napps.kytos.flow_manager.serializers.v0x04.HWAddress')
    @patch('napps.kytos.flow_manager.serializers.v0x04.IPAddress')
    def test_match_to_dict(self, *args):
        """Test _match_to_dict method."""
        (mock_ip_address, mock_hw_address) = args
        ip_address = MagicMock()
        hw_address = MagicMock()
        mock_ip_address.return_value = ip_address
        mock_hw_address.return_value = hw_address

        flow_stats = MagicMock()
        flow_stats.match.oxm_match_fields = []
        for oxm_field in [0, 6, 4, 3, 11, 12]:
            field = MagicMock()
            field.oxm_field = oxm_field
            flow_stats.match.oxm_match_fields.append(field)

        match_dict = self.napp._match_to_dict(flow_stats)
        expected = {'in_port': 0, 'dl_vlan': 0, 'dl_src': str(hw_address),
                    'dl_dst': str(hw_address), 'nw_src': str(ip_address),
                    'nw_dst': str(ip_address)}

        self.assertEqual(match_dict, expected)

    @patch('napps.kytos.flow_manager.serializers.v0x04.FlowSerializer13.'
           '_filter_actions')
    def test_actions_to_list(self, mock_filter_actions):
        """Test _actions_to_list method."""
        flow_stats = MagicMock()
        action = MagicMock()
        action.action_type = 0
        action.port = 0xfffffffd
        mock_filter_actions.return_value = [action]

        actions_list = self.napp._actions_to_list(flow_stats)

        expected = [{'action_type': 'output', 'port': 'controller'}]
        self.assertEqual(actions_list, expected)

    def test_action_to_dict(self):
        """Test _action_to_dict method."""
        action_1 = MagicMock()
        action_1.action_type = 25
        action_1.field.oxm_field = 6
        action_1.field.oxm_value = b'\x01'
        return_1 = self.napp._action_to_dict(action_1)
        expected = {'action_type': 'set_vlan', 'vlan_id': 1}
        self.assertEqual(return_1, expected)

        action_2 = MagicMock()
        action_2.action_type = 0
        action_2.port = 0xfffffffd
        return_2 = self.napp._action_to_dict(action_2)
        expected = {'action_type': 'output', 'port': 'controller'}
        self.assertEqual(return_2, expected)

        action_3 = MagicMock()
        action_3.action_type = 0
        action_3.port.value = 'any'
        return_3 = self.napp._action_to_dict(action_3)
        expected = {'action_type': 'output', 'port': 'any'}
        self.assertEqual(return_3, expected)

        action_4 = MagicMock()
        action_4.action_type = 17
        action_4.ethertype = 0x88a8
        return_4 = self.napp._action_to_dict(action_4)
        expected = {'action_type': 'push_vlan', 'tag_type': 's'}
        self.assertEqual(return_4, expected)

        action_5 = MagicMock()
        action_5.action_type = 17
        action_5.ethertype = 'any'
        return_5 = self.napp._action_to_dict(action_5)
        expected = {'action_type': 'push_vlan', 'tag_type': 'c'}
        self.assertEqual(return_5, expected)

        action_6 = MagicMock()
        action_6.action_type = 18
        return_6 = self.napp._action_to_dict(action_6)
        expected = {'action_type': 'pop_vlan'}
        self.assertEqual(return_6, expected)

        action_7 = MagicMock()
        action_7.action_type = 'any'
        return_7 = self.napp._action_to_dict(action_7)
        self.assertEqual(return_7, {})

    def test_filter_actions(self):
        """Test _filter_actions method."""
        action = MagicMock()
        action.action_type = 0
        action.port = 0xfffffffd
        instruction = MagicMock()
        instruction.instruction_type = 4
        instruction.actions = [action]
        flow_stats = MagicMock()
        flow_stats.instructions = [instruction]

        action_list = self.napp._filter_actions(flow_stats)

        self.assertEqual(list(action_list), instruction.actions)


1			"""Test Flow serializer for OF 1.0 methods."""
2			from unittest import TestCase
3			from unittest.mock import MagicMock, patch
4
5			from napps.kytos.flow_manager.serializers.v0x04 import FlowSerializer13
6
7
8			# pylint: disable=protected-access, unused-argument, no-member
9			class TestFlowSerializer13(TestCase):
10			"""Test the FlowSerializer13 class."""
11
12			def setUp(self):
13			"""Execute steps before each tests."""
14			self.napp = FlowSerializer13()
15
16			@patch('napps.kytos.flow_manager.serializers.v0x04.ActionOutput')
17			@patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
18			@patch('napps.kytos.flow_manager.serializers.v0x04.InstructionApplyAction')
19			@patch('napps.kytos.flow_manager.serializers.v0x04.FlowMod')
20			def test_from_dict(self, *args):
21			"""Test from_dict method."""
22			(mock_flow_mod, mock_instruction, _, _) = args
23			flow_mod = MagicMock()
24			flow_mod.instructions = []
25			flow_mod.match.oxm_match_fields = []
26			instruction = MagicMock()
27			instruction.actions = []
28
29			mock_flow_mod.return_value = flow_mod
30			mock_instruction.return_value = instruction
31
32			self.napp.flow_attributes = {'flow_attr'}
33			self.napp._match_names = {'match_attr': 123}
34			dictionary = {'flow_attr': 'any_data',
35			'match': {'match_attr': 123},
36			'actions': [{'action_type': 'output',
37			'port': 'controller'}]}
38			return_flow_mod = self.napp.from_dict(dictionary)
39
40			self.assertEqual(return_flow_mod.flow_attr, 'any_data')
41			self.assertEqual(len(return_flow_mod.match.oxm_match_fields), 1)
42			self.assertEqual(len(return_flow_mod.instructions), 1)
43			self.assertEqual(len(return_flow_mod.instructions[0].actions), 1)
44
45			@patch('napps.kytos.flow_manager.serializers.v0x04.IPAddress')
46			@patch('napps.kytos.flow_manager.serializers.v0x04.HWAddress')
47			@patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
48			def test_match_from_dict(self, *args):
49			"""Test _match_from_dict method."""
50			(_, mock_hw_address, mock_ip_address) = args
51			hw_address = MagicMock()
52			hw_address.pack.return_value = 'hw_address'
53			mock_hw_address.return_value = hw_address
54			ip_address = MagicMock()
55			ip_address.pack.return_value = 'ip_address'
56			mock_ip_address.return_value = ip_address
57
58			dictionary = {'dl_vlan_pcp': 0, 'nw_proto': 1, 'dl_vlan': 2,
59			'dl_src': 3, 'dl_dst': 4, 'nw_src': 5, 'nw_dst': 6,
60			'in_port': 7, 'dl_type': 8}
61
62			tlv_generator = self.napp._match_from_dict(dictionary)
63
64			self.assertEqual(next(tlv_generator).oxm_value, b'\x00')
65			self.assertEqual(next(tlv_generator).oxm_value, b'\x01')
66			self.assertEqual(next(tlv_generator).oxm_value, b'\x10\x02')
67			self.assertEqual(next(tlv_generator).oxm_value, 'hw_address')
68			self.assertEqual(next(tlv_generator).oxm_value, 'hw_address')
69			self.assertEqual(next(tlv_generator).oxm_value, 'ip_address')
70			self.assertEqual(next(tlv_generator).oxm_value, 'ip_address')
71			self.assertEqual(next(tlv_generator).oxm_value, b'\x00\x00\x00\x07')
72			self.assertEqual(next(tlv_generator).oxm_value, b'\x00\x08')
73
74			@patch('napps.kytos.flow_manager.serializers.v0x04.ActionOutput')
75			def test_actions_from_list(self, mock_action_output):
76			"""Test _actions_from_list method."""
77			actions = [{'action_type': 'output', 'port': 'controller'}]
78			generator = self.napp._actions_from_list(actions)
79			return_action = next(generator)
80
81			self.assertEqual(return_action, mock_action_output.return_value)
82
83			@patch('napps.kytos.flow_manager.serializers.v0x04.ActionPopVLAN',
84			return_value='action_pop_vlan')
85			@patch('napps.kytos.flow_manager.serializers.v0x04.ActionPush',
86			return_value='action_push')
87			@patch('napps.kytos.flow_manager.serializers.v0x04.ActionOutput',
88			return_value='action_output')
89			@patch('napps.kytos.flow_manager.serializers.v0x04.ActionSetField',
90			return_value='action_set_field')
91			@patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
92			def test_action_from_dict(self, *args):
93			"""Test _action_from_dict method."""
94			(mock_oxm_tlv, mock_action_set_field, mock_action_output,
95			mock_action_push, _) = args
96			oxm_tlv = MagicMock()
97			mock_oxm_tlv.return_value = oxm_tlv
98
99			action_1 = {'action_type': 'set_vlan', 'vlan_id': 123}
100			return_1 = self.napp._action_from_dict(action_1)
101			mock_action_set_field.assert_called_with(field=oxm_tlv)
102			self.assertEqual(return_1, 'action_set_field')
103
104			action_2 = {'action_type': 'output', 'port': 'controller'}
105			return_2 = self.napp._action_from_dict(action_2)
106			mock_action_output.assert_called_with(port=0xfffffffd)
107			self.assertEqual(return_2, 'action_output')
108
109			action_3 = {'action_type': 'output', 'port': 'any'}
110			return_3 = self.napp._action_from_dict(action_3)
111			mock_action_output.assert_called_with(port='any')
112			self.assertEqual(return_3, 'action_output')
113
114			action_4 = {'action_type': 'push_vlan', 'tag_type': 's'}
115			return_4 = self.napp._action_from_dict(action_4)
116			mock_action_push.assert_called_with(action_type=17, ethertype=0x88a8)
117			self.assertEqual(return_4, 'action_push')
118
119			action_5 = {'action_type': 'push_vlan', 'tag_type': 'c'}
120			return_5 = self.napp._action_from_dict(action_5)
121			mock_action_push.assert_called_with(action_type=17, ethertype=0x8100)
122			self.assertEqual(return_5, 'action_push')
123
124			action_6 = {'action_type': 'pop_vlan'}
125			return_6 = self.napp._action_from_dict(action_6)
126			self.assertEqual(return_6, 'action_pop_vlan')
127
128			@patch('napps.kytos.flow_manager.serializers.v0x04.OxmTLV')
129			def test_create_vlan_tlv(self, mock_oxm_tlv):
130			"""Test _create_vlan_tlv method."""
131			tlv = self.napp._create_vlan_tlv(1)
132
133			self.assertEqual(tlv.oxm_field, 6)
134			self.assertEqual(tlv.oxm_value, b'\x10\x01')
135
136			@patch('napps.kytos.flow_manager.serializers.v0x04.FlowSerializer13.'
137			'_actions_to_list', return_value='actions_to_list')
138			@patch('napps.kytos.flow_manager.serializers.v0x04.FlowSerializer13.'
139			'_match_to_dict', return_value='match_to_dict')
140			def test_to_dict(self, *args):
141			"""Test to_dict method."""
142			(_, _) = args
143			flow_stats = MagicMock()
144			flow_stats.flow_attr = MagicMock()
145			self.napp.flow_attributes = {'flow_attr'}
146
147			flow_dict = self.napp.to_dict(flow_stats)
148
149			expected = {'flow_attr': flow_stats.flow_attr.value,
150			'match': 'match_to_dict',
151			'actions': 'actions_to_list'}
152			self.assertEqual(flow_dict, expected)
153
154			@patch('napps.kytos.flow_manager.serializers.v0x04.HWAddress')
155			@patch('napps.kytos.flow_manager.serializers.v0x04.IPAddress')
156			def test_match_to_dict(self, *args):
157			"""Test _match_to_dict method."""
158			(mock_ip_address, mock_hw_address) = args
159			ip_address = MagicMock()
160			hw_address = MagicMock()
161			mock_ip_address.return_value = ip_address
162			mock_hw_address.return_value = hw_address
163
164			flow_stats = MagicMock()
165			flow_stats.match.oxm_match_fields = []
166			for oxm_field in [0, 6, 4, 3, 11, 12]:
167			field = MagicMock()
168			field.oxm_field = oxm_field
169			flow_stats.match.oxm_match_fields.append(field)
170
171			match_dict = self.napp._match_to_dict(flow_stats)
172			expected = {'in_port': 0, 'dl_vlan': 0, 'dl_src': str(hw_address),
173			'dl_dst': str(hw_address), 'nw_src': str(ip_address),
174			'nw_dst': str(ip_address)}
175
176			self.assertEqual(match_dict, expected)
177
178			@patch('napps.kytos.flow_manager.serializers.v0x04.FlowSerializer13.'
179			'_filter_actions')
180			def test_actions_to_list(self, mock_filter_actions):
181			"""Test _actions_to_list method."""
182			flow_stats = MagicMock()
183			action = MagicMock()
184			action.action_type = 0
185			action.port = 0xfffffffd
186			mock_filter_actions.return_value = [action]
187
188			actions_list = self.napp._actions_to_list(flow_stats)
189
190			expected = [{'action_type': 'output', 'port': 'controller'}]
191			self.assertEqual(actions_list, expected)
192
193			def test_action_to_dict(self):
194			"""Test _action_to_dict method."""
195			action_1 = MagicMock()
196			action_1.action_type = 25
197			action_1.field.oxm_field = 6
198			action_1.field.oxm_value = b'\x01'
199			return_1 = self.napp._action_to_dict(action_1)
200			expected = {'action_type': 'set_vlan', 'vlan_id': 1}
201			self.assertEqual(return_1, expected)
202
203			action_2 = MagicMock()
204			action_2.action_type = 0
205			action_2.port = 0xfffffffd
206			return_2 = self.napp._action_to_dict(action_2)
207			expected = {'action_type': 'output', 'port': 'controller'}
208			self.assertEqual(return_2, expected)
209
210			action_3 = MagicMock()
211			action_3.action_type = 0
212			action_3.port.value = 'any'
213			return_3 = self.napp._action_to_dict(action_3)
214			expected = {'action_type': 'output', 'port': 'any'}
215			self.assertEqual(return_3, expected)
216
217			action_4 = MagicMock()
218			action_4.action_type = 17
219			action_4.ethertype = 0x88a8
220			return_4 = self.napp._action_to_dict(action_4)
221			expected = {'action_type': 'push_vlan', 'tag_type': 's'}
222			self.assertEqual(return_4, expected)
223
224			action_5 = MagicMock()
225			action_5.action_type = 17
226			action_5.ethertype = 'any'
227			return_5 = self.napp._action_to_dict(action_5)
228			expected = {'action_type': 'push_vlan', 'tag_type': 'c'}
229			self.assertEqual(return_5, expected)
230
231			action_6 = MagicMock()
232			action_6.action_type = 18
233			return_6 = self.napp._action_to_dict(action_6)
234			expected = {'action_type': 'pop_vlan'}
235			self.assertEqual(return_6, expected)
236
237			action_7 = MagicMock()
238			action_7.action_type = 'any'
239			return_7 = self.napp._action_to_dict(action_7)
240			self.assertEqual(return_7, {})
241
242			def test_filter_actions(self):
243			"""Test _filter_actions method."""
244			action = MagicMock()
245			action.action_type = 0
246			action.port = 0xfffffffd
247			instruction = MagicMock()
248			instruction.instruction_type = 4
249			instruction.actions = [action]
250			flow_stats = MagicMock()
251			flow_stats.instructions = [instruction]
252
253			action_list = self.napp._filter_actions(flow_stats)
254
255			self.assertEqual(list(action_list), instruction.actions)
256

kytos / flow_manager

Push — master ( 151596...743f3c )

TestFlowSerializer13.setUp() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like