build.tests.unit.test_flow_serializer_v0x04.TestFlowSerializer13.test_from_dict() - Code Metrics - Inspection of "Create unit tests" - kytos/flow_manager - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#81)

by Gleyberson

created 2020-06-02 18:42 UTC

TestFlowSerializer13.test_from_dict() A

↳ Parent: build.tests.unit.test_flow_serializer_v0x04

Complexity

Conditions

Size

Total Lines	29
Code Lines	25

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	25
dl	0
loc	29
rs	9.28
c	0
b	0
f	0
cc	1
nop	2

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

kytos / flow_manager

Pull Request — master (#81)

TestFlowSerializer13.test_from_dict() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like