'obj', '-name', 'ethernet 1')
self.ixnet_gen.ixnet.commit.assert_called_once()
+ def test_add_vlans_single(self):
+ obj = 'ethernet'
+ self.ixnet_gen.ixnet.getAttribute.return_value = 'attr'
+ self.ixnet_gen.ixnet.getList.return_value = ['vlan1', 'vlan2']
+ vlan1 = ixnet_api.Vlan(vlan_id=100, tp_id='ethertype88a8', prio=2)
+ vlan2 = ixnet_api.Vlan(vlan_id=101, tp_id='ethertype88a8', prio=3)
+ self.ixnet_gen.add_vlans(obj, [vlan1, vlan2])
+ self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('ethernet',
+ '-vlanCount', 2)
+ self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue',
+ '-value', 100)
+ self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue',
+ '-value', 101)
+ self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue',
+ '-value', 2)
+ self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue',
+ '-value', 3)
+ self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call(
+ 'attr/singleValue', '-value', 'ethertype88a8')
+ self.assertEqual(self.ixnet.commit.call_count, 2)
+
+ def test_add_vlans_increment(self):
+ obj = 'ethernet'
+ self.ixnet_gen.ixnet.add.return_value = 'obj'
+ self.ixnet_gen.ixnet.getAttribute.return_value = 'attr'
+ self.ixnet_gen.ixnet.getList.return_value = ['vlan1']
+ vlan = ixnet_api.Vlan(vlan_id=100, vlan_id_step=1, prio=3, prio_step=2)
+ self.ixnet_gen.add_vlans(obj, [vlan])
+ self.ixnet.setMultiAttribute.assert_any_call('obj', '-start', 100,
+ '-step', 1,
+ '-direction', 'increment')
+ self.ixnet.setMultiAttribute.assert_any_call('obj', '-start', 3,
+ '-step', 2,
+ '-direction', 'increment')
+
+ self.assertEqual(self.ixnet.commit.call_count, 2)
+
+ def test_add_vlans_invalid(self):
+ vlans = []
+ self.assertRaises(RuntimeError, self.ixnet_gen.add_vlans, 'obj', vlans)
+
def test_add_ipv4(self):
self.ixnet_gen.ixnet.add.return_value = 'obj'
self.ixnet_gen.add_ipv4('ethernet 1', name='ipv4 1')
self.ixnet_gen.ixnet.setMultiAttribute.assert_not_called()
+ def test_add_bgp(self):
+ self.ixnet_gen.ixnet.add.return_value = 'obj'
+ self.ixnet_gen.ixnet.getAttribute.return_value = 'attr'
+ self.ixnet_gen.add_bgp(ipv4='ipv4 1',
+ dut_ip='10.0.0.1',
+ local_as=65000,
+ bgp_type='external')
+ self.ixnet_gen.ixnet.add.assert_called_once_with('ipv4 1', 'bgpIpv4Peer')
+ self.ixnet_gen.ixnet.setAttribute.assert_any_call(
+ 'attr/singleValue', '-value', '10.0.0.1')
+ self.ixnet_gen.ixnet.setAttribute.assert_any_call(
+ 'attr/singleValue', '-value', 65000)
+ self.ixnet_gen.ixnet.setAttribute.assert_any_call(
+ 'attr/singleValue', '-value', 'external')
+
@mock.patch.object(IxNetwork, 'IxNet')
def test_connect(self, mock_ixnet):
mock_ixnet.return_value = self.ixnet
self.ixnet_gen._cfg = config
self.assertIsNone(self.ixnet_gen.assign_ports())
- self.assertEqual(self.ixnet.execute.call_count, 2)
- self.assertEqual(self.ixnet.commit.call_count, 4)
+ self.assertEqual(self.ixnet.execute.call_count, 1)
+ self.assertEqual(self.ixnet.commit.call_count, 3)
self.assertEqual(self.ixnet.getAttribute.call_count, 2)
@mock.patch.object(ixnet_api, 'log')
'-trackBy', 'trafficGroupId0')
def test__create_flow_groups(self):
+ uplink_endpoints = ['up_endp1', 'up_endp2']
+ downlink_endpoints = ['down_endp1', 'down_endp2']
self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'], ['1', '2']]
- self.ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2']
- self.ixnet_gen._create_flow_groups()
+ self.ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2', 'endp3',
+ 'endp4']
+ self.ixnet_gen._create_flow_groups(uplink_endpoints, downlink_endpoints)
self.ixnet_gen.ixnet.add.assert_has_calls([
mock.call('traffic_item', 'endpointSet'),
mock.call('traffic_item', 'endpointSet')])
self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([
- mock.call('endp1', '-name', '1', '-sources', ['1/protocols'],
- '-destinations', ['2/protocols']),
- mock.call('endp2', '-name', '2', '-sources', ['2/protocols'],
- '-destinations', ['1/protocols'])])
+ mock.call('endp1', '-name', '1', '-sources', ['up_endp1'],
+ '-destinations', ['down_endp1']),
+ mock.call('endp2', '-name', '2', '-sources', ['down_endp1'],
+ '-destinations', ['up_endp1']),
+ mock.call('endp3', '-name', '3', '-sources', ['up_endp2'],
+ '-destinations', ['down_endp2']),
+ mock.call('endp4', '-name', '4', '-sources', ['down_endp2'],
+ '-destinations', ['up_endp2'])])
def test__append_protocol_to_stack(self):
def test_create_traffic_model(self, mock__setup_config_elements,
mock__create_flow_groups,
mock__create_traffic_item):
-
- self.ixnet_gen.create_traffic_model()
- mock__create_traffic_item.assert_called_once()
- mock__create_flow_groups.assert_called_once()
+ uplink_ports = ['port1', 'port3']
+ downlink_ports = ['port2', 'port4']
+ uplink_endpoints = ['port1/protocols', 'port3/protocols']
+ downlink_endpoints = ['port2/protocols', 'port4/protocols']
+ self.ixnet_gen.create_traffic_model(uplink_ports, downlink_ports)
+ mock__create_traffic_item.assert_called_once_with('raw')
+ mock__create_flow_groups.assert_called_once_with(uplink_endpoints,
+ downlink_endpoints)
mock__setup_config_elements.assert_called_once()
+ @mock.patch.object(ixnet_api.IxNextgen, '_create_traffic_item')
+ @mock.patch.object(ixnet_api.IxNextgen, '_create_flow_groups')
+ @mock.patch.object(ixnet_api.IxNextgen, '_setup_config_elements')
+ def test_create_ipv4_traffic_model(self, mock__setup_config_elements,
+ mock__create_flow_groups,
+ mock__create_traffic_item):
+ uplink_topologies = ['up1', 'up3']
+ downlink_topologies = ['down2', 'down4']
+ self.ixnet_gen.create_ipv4_traffic_model(uplink_topologies,
+ downlink_topologies)
+ mock__create_traffic_item.assert_called_once_with('ipv4')
+ mock__create_flow_groups.assert_called_once_with(uplink_topologies,
+ downlink_topologies)
+ mock__setup_config_elements.assert_called_once_with(False)
+
def test__update_frame_mac(self):
with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item') as \
mock_get_field:
mock.call(port_statistics, self.ixnet_gen.PORT_STATS_NAME_MAP),
mock.call(flow_statistics, self.ixnet_gen.LATENCY_NAME_MAP)])
+ def test__set_flow_tracking(self):
+ self.ixnet_gen._ixnet.getList.return_value = ['traffic_item']
+ self.ixnet_gen._set_flow_tracking(track_by=['vlanVlanId0'])
+ self.ixnet_gen.ixnet.setAttribute.assert_called_once_with(
+ 'traffic_item/tracking', '-trackBy', ['vlanVlanId0'])
+ self.assertEqual(self.ixnet.commit.call_count, 1)
+
+ def test__set_egress_flow_tracking(self):
+ self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'],
+ ['encapsulation']]
+ self.ixnet_gen._set_egress_flow_tracking(encapsulation='Ethernet',
+ offset='IPv4 TOS Precedence')
+ self.ixnet_gen.ixnet.setAttribute.assert_any_call(
+ 'traffic_item', '-egressEnabled', True)
+ self.ixnet_gen.ixnet.setAttribute.assert_any_call(
+ 'encapsulation', '-encapsulation', 'Ethernet')
+ self.ixnet_gen.ixnet.setAttribute.assert_any_call(
+ 'encapsulation', '-offset', 'IPv4 TOS Precedence')
+ self.assertEqual(self.ixnet.commit.call_count, 2)
+
def test__update_ipv4_address(self):
with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
return_value='field_desc'):
'outer_l3': {
'proto': 'unsupported',
},
+ 'outer_l4': {
+ 'seed': 1
+ }
},
}
with mock.patch.object(self.ixnet_gen,
self.assertIsNone(result)
self.ixnet.getList.assert_called_once()
self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count)
+
+ def test__get_protocol_status(self):
+ self.ixnet.getAttribute.return_value = ['up']
+ self.ixnet_gen._get_protocol_status('ipv4')
+ self.ixnet.getAttribute.assert_called_once_with('ipv4',
+ '-sessionStatus')
+
+ @mock.patch.object(ixnet_api.IxNextgen, '_get_protocol_status')
+ def test_is_protocols_running(self, mock_ixnextgen_get_protocol_status):
+ mock_ixnextgen_get_protocol_status.return_value = 'up'
+ result = self.ixnet_gen.is_protocols_running(['ethernet', 'ipv4'])
+ self.assertTrue(result)
+
+ @mock.patch.object(ixnet_api.IxNextgen, '_get_protocol_status')
+ def test_is_protocols_stopped(self, mock_ixnextgen_get_protocol_status):
+ mock_ixnextgen_get_protocol_status.return_value = 'down'
+ result = self.ixnet_gen.is_protocols_running(['ethernet', 'ipv4'])
+ self.assertFalse(result)
+
+ def test_start_protocols(self):
+ self.ixnet_gen.start_protocols()
+ self.ixnet.execute.assert_called_once_with('startAllProtocols')
+
+ def test_stop_protocols(self):
+ self.ixnet_gen.stop_protocols()
+ self.ixnet.execute.assert_called_once_with('stopAllProtocols')
+
+ def test_get_vports(self):
+ self.ixnet_gen._ixnet.getRoot.return_value = 'root'
+ self.ixnet_gen.get_vports()
+ self.ixnet.getList.assert_called_once_with('root', 'vport')