1
# Copyright (c) 2012 OpenStack Foundation.
3
# Licensed under the Apache License, Version 2.0 (the "License");
4
# you may not use this file except in compliance with the License.
5
# You may obtain a copy of the License at
7
# http://www.apache.org/licenses/LICENSE-2.0
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS,
11
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
13
# See the License for the specific language governing permissions and
14
# limitations under the License.
19
import webob.exc as wexc
21
from neutron.api.v2 import base
22
from neutron import context
23
from neutron.extensions import portbindings
24
from neutron import manager
25
from neutron.openstack.common import log as logging
26
from neutron.plugins.common import constants as p_const
27
from neutron.plugins.ml2 import driver_api as api
28
from neutron.plugins.ml2 import driver_context
29
from neutron.plugins.ml2.drivers.cisco.nexus import config as cisco_config
30
from neutron.plugins.ml2.drivers.cisco.nexus import exceptions as c_exc
31
from neutron.plugins.ml2.drivers.cisco.nexus import mech_cisco_nexus
32
from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2
33
from neutron.plugins.ml2.drivers.cisco.nexus import nexus_network_driver
34
from neutron.tests.unit.ml2 import test_ml2_plugin
37
LOG = logging.getLogger(__name__)
39
COMP_HOST_NAME = 'testhost'
40
COMP_HOST_NAME_2 = 'testhost_2'
43
NEXUS_IP_ADDR = '1.1.1.1'
44
NETWORK_NAME = 'test_network'
45
NETWORK_NAME_2 = 'test_network_2'
46
NEXUS_INTERFACE = '1/1'
47
NEXUS_INTERFACE_2 = '1/2'
48
CIDR_1 = '10.0.0.0/24'
49
CIDR_2 = '10.0.1.0/24'
50
DEVICE_ID_1 = '11111111-1111-1111-1111-111111111111'
51
DEVICE_ID_2 = '22222222-2222-2222-2222-222222222222'
52
DEVICE_OWNER = 'compute:None'
53
BOUND_SEGMENT1 = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
54
api.PHYSICAL_NETWORK: PHYS_NET,
55
api.SEGMENTATION_ID: VLAN_START}
56
BOUND_SEGMENT2 = {api.NETWORK_TYPE: p_const.TYPE_VLAN,
57
api.PHYSICAL_NETWORK: PHYS_NET,
58
api.SEGMENTATION_ID: VLAN_START + 1}
61
class CiscoML2MechanismTestCase(test_ml2_plugin.Ml2PluginV2TestCase):
62
_mechanism_drivers = ['cisco_nexus']
65
"""Configure for end-to-end neutron testing using a mock ncclient.
68
- Configure the ML2 plugin to use VLANs in the range of 1000-1100.
69
- Configure the Cisco mechanism driver to use an imaginary switch
71
- Create a mock NETCONF client (ncclient) for the Cisco mechanism
76
# Configure the Cisco Nexus mechanism driver
78
(NEXUS_IP_ADDR, 'username'): 'admin',
79
(NEXUS_IP_ADDR, 'password'): 'mySecretPassword',
80
(NEXUS_IP_ADDR, 'ssh_port'): 22,
81
(NEXUS_IP_ADDR, COMP_HOST_NAME): NEXUS_INTERFACE,
82
(NEXUS_IP_ADDR, COMP_HOST_NAME_2): NEXUS_INTERFACE_2}
83
nexus_patch = mock.patch.dict(
84
cisco_config.ML2MechCiscoConfig.nexus_dict,
87
self.addCleanup(nexus_patch.stop)
89
# The NETCONF client module is not included in the DevStack
90
# distribution, so mock this module for unit testing.
91
self.mock_ncclient = mock.Mock()
92
mock.patch.object(nexus_network_driver.CiscoNexusDriver,
94
return_value=self.mock_ncclient).start()
96
# Mock port context values for bound_segments and 'status'.
97
self.mock_bound_segment = mock.patch.object(
98
driver_context.PortContext,
99
'bottom_bound_segment',
100
new_callable=mock.PropertyMock).start()
101
self.mock_bound_segment.return_value = BOUND_SEGMENT1
103
self.mock_original_bound_segment = mock.patch.object(
104
driver_context.PortContext,
105
'original_bottom_bound_segment',
106
new_callable=mock.PropertyMock).start()
107
self.mock_original_bound_segment.return_value = None
109
# Use _is_status_active method to determine bind state.
110
def _mock_check_bind_state(port_context):
111
if (port_context[portbindings.VIF_TYPE] !=
112
portbindings.VIF_TYPE_UNBOUND):
117
self.mock_status = mock.patch.object(
118
mech_cisco_nexus.CiscoNexusMechanismDriver,
119
'_is_status_active').start()
120
self.mock_status.side_effect = _mock_check_bind_state
122
super(CiscoML2MechanismTestCase, self).setUp()
124
self.port_create_status = 'DOWN'
126
def _create_deviceowner_mock(self):
127
# Mock deviceowner method for UT's that expect update precommit
128
# failures. This allows control of delete_port_pre/postcommit()
130
mock_deviceowner = mock.patch.object(
131
mech_cisco_nexus.CiscoNexusMechanismDriver,
132
'_is_deviceowner_compute').start()
133
mock_deviceowner.return_value = False
134
self.addCleanup(mock_deviceowner.stop)
136
@contextlib.contextmanager
137
def _patch_ncclient(self, attr, value):
138
"""Configure an attribute on the mock ncclient module.
140
This method can be used to inject errors by setting a side effect
141
or a return value for an ncclient method.
143
:param attr: ncclient attribute (typically method) to be configured.
144
:param value: Value to be configured on the attribute.
147
# Configure attribute.
148
config = {attr: value}
149
self.mock_ncclient.configure_mock(**config)
152
# Unconfigure attribute
153
config = {attr: None}
154
self.mock_ncclient.configure_mock(**config)
157
def _config_dependent_side_effect(match_config, exc):
158
"""Generates a config-dependent side effect for ncclient edit_config.
160
This method generates a mock side-effect function which can be
161
configured on the mock ncclient module for the edit_config method.
162
This side effect will cause a given exception to be raised whenever
163
the XML config string that is passed to edit_config contains all
164
words in a given match config string.
166
:param match_config: String containing keywords to be matched
167
:param exc: Exception to be raised when match is found
168
:return: Side effect function for the mock ncclient module's
172
keywords = match_config.split()
174
def _side_effect_function(target, config):
175
if all(word in config for word in keywords):
177
return _side_effect_function
179
def _is_in_nexus_cfg(self, words):
180
"""Check if any config sent to Nexus contains all words in a list."""
181
for call in (self.mock_ncclient.connect.return_value.
182
edit_config.mock_calls):
183
configlet = call[2]['config']
184
if all(word in configlet for word in words):
188
def _is_in_last_nexus_cfg(self, words):
189
"""Confirm last config sent to Nexus contains specified keywords."""
190
last_cfg = (self.mock_ncclient.connect.return_value.
191
edit_config.mock_calls[-1][2]['config'])
192
return all(word in last_cfg for word in words)
194
def _is_vlan_configured(self, vlan_creation_expected=True,
195
first_vlan_addition=False):
196
"""Confirm if VLAN was configured or not."""
197
vlan_created = self._is_in_nexus_cfg(['vlan', 'vlan-name'])
198
add_appears = self._is_in_last_nexus_cfg(['add'])
199
# The first VLAN being configured should be done without the
200
# ADD keyword. Thereafter additional VLANs to be configured
201
# should be done with the ADD keyword.
202
add_keyword_expected = not first_vlan_addition
203
return (self._is_in_last_nexus_cfg(['allowed', 'vlan']) and
204
vlan_created == vlan_creation_expected and
205
add_appears == add_keyword_expected)
207
def _is_vlan_unconfigured(self, vlan_deletion_expected=True):
208
vlan_deleted = self._is_in_last_nexus_cfg(
209
['no', 'vlan', 'vlan-id-create-delete'])
210
return (self._is_in_nexus_cfg(['allowed', 'vlan', 'remove']) and
211
vlan_deleted == vlan_deletion_expected)
214
class TestCiscoBasicGet(CiscoML2MechanismTestCase,
215
test_ml2_plugin.TestMl2BasicGet):
220
class TestCiscoV2HTTPResponse(CiscoML2MechanismTestCase,
221
test_ml2_plugin.TestMl2V2HTTPResponse):
226
class TestCiscoPortsV2(CiscoML2MechanismTestCase,
227
test_ml2_plugin.TestMl2PortsV2):
229
@contextlib.contextmanager
230
def _create_resources(self, name=NETWORK_NAME, cidr=CIDR_1,
231
device_id=DEVICE_ID_1,
232
host_id=COMP_HOST_NAME,
233
expected_failure=False):
234
"""Create network, subnet, and port resources for test cases.
236
Create a network, subnet, port and then update the port, yield the
237
result, then delete the port, subnet and network.
239
:param name: Name of network to be created.
240
:param cidr: cidr address of subnetwork to be created.
241
:param device_id: Device ID to use for port to be created/updated.
242
:param host_id: Host ID to use for port create/update.
243
:param expected_failure: Set to True when an update_port_precommit
244
failure is expected. Results in no actions being taken in
245
delete_port_pre/postcommit() methods.
247
with self.network(name=name) as network:
248
with self.subnet(network=network, cidr=cidr) as subnet:
249
with self.port(subnet=subnet, cidr=cidr) as port:
251
data = {'port': {portbindings.HOST_ID: host_id,
252
'device_id': device_id,
253
'device_owner': DEVICE_OWNER,
254
'admin_state_up': True}}
255
req = self.new_update_request('ports', data,
257
yield req.get_response(self.api)
259
self._create_deviceowner_mock()
260
self._delete('ports', port['port']['id'])
261
self._delete('networks', network['network']['id'])
263
def _assertExpectedHTTP(self, status, exc):
264
"""Confirm that an HTTP status corresponds to an expected exception.
266
Confirm that an HTTP status which has been returned for an
267
neutron API request matches the HTTP status corresponding
268
to an expected exception.
270
:param status: HTTP status
271
:param exc: Expected exception
274
if exc in base.FAULT_MAP:
275
expected_http = base.FAULT_MAP[exc].code
277
expected_http = wexc.HTTPInternalServerError.code
278
self.assertEqual(status, expected_http)
280
def _mock_config_first_trunk(self):
281
"""Mock the behavior for the first VLAN addition.
283
When the first VLAN is being added to the interface the usage of
284
the ADD keyword should raise an exception specifying that the ADD
285
keyword cannot be used till a VLAN list is created and to create
286
a VLAN list the configuration should not contain the ADD keyword.
289
config = "switchport trunk allowed vlan add"
290
exc = Exception("switchport trunk allowed vlan list is empty, "
291
"please specify a list using "
292
"'switchport trunk allowed vlan X' "
293
"before using the add option")
294
return (self._patch_ncclient(
295
'connect.return_value.edit_config.side_effect',
296
self._config_dependent_side_effect(config, exc)))
298
def test_create_ports_bulk_emulated_plugin_failure(self):
299
real_has_attr = hasattr
301
#ensures the API chooses the emulation code path
302
def fakehasattr(item, attr):
303
if attr.endswith('__native_bulk_support'):
305
return real_has_attr(item, attr)
307
with mock.patch('__builtin__.hasattr',
309
plugin_obj = manager.NeutronManager.get_plugin()
310
orig = plugin_obj.create_port
311
with mock.patch.object(plugin_obj,
312
'_create_port_db') as patched_plugin:
314
def side_effect(*args, **kwargs):
315
return self._fail_second_call(patched_plugin, orig,
318
patched_plugin.side_effect = side_effect
319
with self.network() as net:
320
res = self._create_port_bulk(self.fmt, 2,
321
net['network']['id'],
324
# Expect an internal server error as we injected a fault
325
self._validate_behavior_on_bulk_failure(
328
wexc.HTTPInternalServerError.code)
330
def test_create_ports_bulk_native(self):
331
if self._skip_native_bulk:
332
self.skipTest("Plugin does not support native bulk port create")
334
def test_create_ports_bulk_emulated(self):
335
if self._skip_native_bulk:
336
self.skipTest("Plugin does not support native bulk port create")
338
def test_create_ports_bulk_native_plugin_failure(self):
339
if self._skip_native_bulk:
340
self.skipTest("Plugin does not support native bulk port create")
341
ctx = context.get_admin_context()
342
with self.network() as net:
343
plugin_obj = manager.NeutronManager.get_plugin()
344
orig = plugin_obj.create_port
345
with mock.patch.object(plugin_obj,
346
'_create_port_db') as patched_plugin:
348
def side_effect(*args, **kwargs):
349
return self._fail_second_call(patched_plugin, orig,
352
patched_plugin.side_effect = side_effect
353
res = self._create_port_bulk(self.fmt, 2, net['network']['id'],
354
'test', True, context=ctx)
355
# We expect an internal server error as we injected a fault
356
self._validate_behavior_on_bulk_failure(
359
wexc.HTTPInternalServerError.code)
361
def test_nexus_enable_vlan_cmd_on_same_host(self):
362
"""Verify the syntax of the command to enable a vlan on an intf.
364
Test of the following ml2_conf_cisco_ini config:
365
[ml2_mech_cisco_nexus:1.1.1.1]
366
Resource A on host=COMP_HOST_NAME with vlan_id = 1000
367
Resource B on host=COMP_HOST_NAME with vlan_id = 1001
369
Confirm that when configuring the first VLAN on a Nexus interface,
370
the final command string sent to the switch does not contain the
371
keyword 'add'. The initial attempt will contain 'add' but when
372
the switch rejects it, the re-attempt shouldn't contain the 'add'.
374
Confirm that for the second VLAN configured on a Nexus interface,
375
the command string sent to the switch contains the keyword 'add'
376
since it is on the same host.
379
# First vlan should be configured without 'add' keyword and an
380
# exception should be raised when it is done with the 'add'
381
# thereby triggering a re-attempt without the 'add'.
382
with self._mock_config_first_trunk():
383
with self._create_resources():
384
self.assertTrue(self._is_vlan_configured(
385
vlan_creation_expected=True,
386
first_vlan_addition=True))
387
self.mock_ncclient.reset_mock()
388
self.mock_bound_segment.return_value = BOUND_SEGMENT2
390
# Second vlan should be configured with the 'add' keyword
391
# when on first host.
392
with(self._patch_ncclient(
393
'connect.return_value.edit_config.side_effect',
395
with self._create_resources(name=NETWORK_NAME_2,
396
device_id=DEVICE_ID_2,
398
host_id=COMP_HOST_NAME):
399
self.assertTrue(self._is_vlan_configured(
400
vlan_creation_expected=True,
401
first_vlan_addition=False
404
# Return to first segment for delete port calls.
405
self.mock_bound_segment.return_value = BOUND_SEGMENT1
407
def test_nexus_enable_vlan_cmd_on_different_hosts(self):
408
"""Verify the syntax of the command to enable a vlan on an intf.
410
Test of the following ml2_conf_cisco_ini config:
411
[ml2_mech_cisco_nexus:1.1.1.1]
412
Resource A on host=COMP_HOST_NAME with vlan_id = 1000
413
Resource B on host=COMP_HOST_NAME_2 with vlan_id = 1001
415
Confirm that when configuring the first VLAN on a Nexus interface,
416
the final command string sent to the switch does not contain the
417
keyword 'add'. The initial attempt will contain 'add' but when
418
the switch rejects it, the re-attempt shouldn't contain the 'add'.
420
Confirm that for the second VLAN configured on a Nexus interface,
421
the command string sent to the switch does not contain the
422
keyword 'add' since it is on a different host.
425
# First vlan should be configured without 'add' keyword and an
426
# exception should be raised when it is done with the 'add'
427
# thereby triggering a re-attempt without the 'add'.
428
with self._mock_config_first_trunk():
429
with self._create_resources():
430
self.assertTrue(self._is_vlan_configured(
431
vlan_creation_expected=True,
432
first_vlan_addition=True))
433
self.mock_ncclient.reset_mock()
434
self.mock_bound_segment.return_value = BOUND_SEGMENT2
436
# Second vlan should be configured without the 'add' keyword
437
# when on second host.
438
with self._create_resources(name=NETWORK_NAME_2,
439
device_id=DEVICE_ID_2,
441
host_id=COMP_HOST_NAME_2):
442
self.assertTrue(self._is_vlan_configured(
443
vlan_creation_expected=True,
444
first_vlan_addition=True
447
# Return to first segment for delete port calls.
448
self.mock_bound_segment.return_value = BOUND_SEGMENT1
450
def test_ncclient_version_detect(self):
451
"""Test ability to handle connection to old and new-style ncclient.
453
We used to require a custom version of the ncclient library. However,
454
recent contributions to the ncclient make this unnecessary. Our
455
driver was modified to be able to establish a connection via both
456
the old and new type of ncclient.
458
The new style ncclient.connect() function takes one additional
461
The ML2 driver uses this to detect whether we are dealing with an
462
old or new ncclient installation.
465
# The code we are exercising calls connect() twice, if there is a
466
# TypeError on the first call (if the old ncclient is installed).
467
# The second call should succeed. That's what we are simulating here.
468
orig_connect_return_val = self.mock_ncclient.connect.return_value
469
with self._patch_ncclient('connect.side_effect',
470
[TypeError, orig_connect_return_val]):
471
with self._create_resources() as result:
472
self.assertEqual(result.status_int,
475
def test_ncclient_fail_on_second_connect(self):
476
"""Test that other errors during connect() sequences are still handled.
478
If the old ncclient is installed, we expect to get a TypeError first,
479
but should still handle other errors in the usual way, whether they
480
appear on the first or second call to connect().
483
with self._patch_ncclient('connect.side_effect',
484
[TypeError, IOError]):
485
with self._create_resources() as result:
486
self._assertExpectedHTTP(result.status_int,
487
c_exc.NexusConnectFailed)
489
def test_nexus_connect_fail(self):
490
"""Test failure to connect to a Nexus switch.
492
While creating a network, subnet, and port, simulate a connection
493
failure to a nexus switch. Confirm that the expected HTTP code
494
is returned for the create port operation.
497
with self._patch_ncclient('connect.side_effect',
499
with self._create_resources() as result:
500
self._assertExpectedHTTP(result.status_int,
501
c_exc.NexusConnectFailed)
503
def test_nexus_vlan_config_two_hosts(self):
504
"""Verify config/unconfig of vlan on two compute hosts."""
506
@contextlib.contextmanager
507
def _create_port_check_vlan(comp_host_name, device_id,
508
vlan_creation_expected=True):
509
with self.port(subnet=subnet, fmt=self.fmt) as port:
510
data = {'port': {portbindings.HOST_ID: comp_host_name,
511
'device_id': device_id,
512
'device_owner': DEVICE_OWNER,
513
'admin_state_up': True}}
514
req = self.new_update_request('ports', data,
516
req.get_response(self.api)
517
self.assertTrue(self._is_vlan_configured(
518
vlan_creation_expected=vlan_creation_expected,
519
first_vlan_addition=True))
520
self.mock_ncclient.reset_mock()
522
self._delete('ports', port['port']['id'])
524
# Create network and subnet
525
with self._mock_config_first_trunk():
526
with self.network(name=NETWORK_NAME) as network:
527
with self.subnet(network=network, cidr=CIDR_1) as subnet:
529
# Create an instance on first compute host
530
with _create_port_check_vlan(COMP_HOST_NAME, DEVICE_ID_1,
531
vlan_creation_expected=True):
532
# Create an instance on second compute host
533
with _create_port_check_vlan(
536
vlan_creation_expected=False):
539
# Instance on second host is now terminated.
540
# Vlan should be untrunked from port, but vlan should
541
# still exist on the switch.
542
self.assertTrue(self._is_vlan_unconfigured(
543
vlan_deletion_expected=False))
544
self.mock_ncclient.reset_mock()
546
# Instance on first host is now terminated.
547
# Vlan should be untrunked from port and vlan should have
548
# been deleted from the switch.
549
self.assertTrue(self._is_vlan_unconfigured(
550
vlan_deletion_expected=True))
552
def test_nexus_vm_migration(self):
553
"""Verify VM (live) migration.
555
Simulate the following:
556
Nova informs neutron of live-migration with port-update(new host).
557
This should trigger two update_port_pre/postcommit() calls.
559
The first one should only change the current host_id and remove the
560
binding resulting in the mechanism drivers receiving:
561
PortContext.original['binding:host_id']: previous value
562
PortContext.original_bottom_bound_segment: previous value
563
PortContext.current['binding:host_id']: current (new) value
564
PortContext.bottom_bound_segment: None
566
The second one binds the new host resulting in the mechanism
568
PortContext.original['binding:host_id']: previous value
569
PortContext.original_bottom_bound_segment: None
570
PortContext.current['binding:host_id']: previous value
571
PortContext.bottom_bound_segment: new value
574
# Create network, subnet and port.
575
with self._create_resources() as result:
576
# Verify initial database entry.
577
# Use port_id to verify that 1st host name was used.
578
binding = nexus_db_v2.get_nexusvm_bindings(VLAN_START,
580
intf_type, nexus_port = binding.port_id.split(':')
581
self.assertEqual(nexus_port, NEXUS_INTERFACE)
583
port = self.deserialize(self.fmt, result)
584
port_id = port['port']['id']
586
# Trigger update event to unbind segment.
587
# Results in port being deleted from nexus DB and switch.
588
data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME_2}}
589
self.mock_bound_segment.return_value = None
590
self.mock_original_bound_segment.return_value = BOUND_SEGMENT1
591
self.new_update_request('ports', data,
592
port_id).get_response(self.api)
594
# Verify that port entry has been deleted.
595
self.assertRaises(c_exc.NexusPortBindingNotFound,
596
nexus_db_v2.get_nexusvm_bindings,
597
VLAN_START, DEVICE_ID_1)
599
# Trigger update event to bind segment with new host.
600
self.mock_bound_segment.return_value = BOUND_SEGMENT1
601
self.mock_original_bound_segment.return_value = None
602
self.new_update_request('ports', data,
603
port_id).get_response(self.api)
605
# Verify that port entry has been added using new host name.
606
# Use port_id to verify that 2nd host name was used.
607
binding = nexus_db_v2.get_nexusvm_bindings(VLAN_START,
609
intf_type, nexus_port = binding.port_id.split(':')
610
self.assertEqual(nexus_port, NEXUS_INTERFACE_2)
612
def test_nexus_config_fail(self):
613
"""Test a Nexus switch configuration failure.
615
While creating a network, subnet, and port, simulate a nexus
616
switch configuration error. Confirm that the expected HTTP code
617
is returned for the create port operation.
620
with self._patch_ncclient(
621
'connect.return_value.edit_config.side_effect',
623
with self._create_resources() as result:
624
self._assertExpectedHTTP(result.status_int,
625
c_exc.NexusConfigFailed)
627
def test_nexus_extended_vlan_range_failure(self):
628
"""Test that extended VLAN range config errors are ignored.
630
Some versions of Nexus switch do not allow state changes for
631
the extended VLAN range (1006-4094), but these errors can be
632
ignored (default values are appropriate). Test that such errors
633
are ignored by the Nexus plugin.
636
def mock_edit_config_a(target, config):
637
if all(word in config for word in ['state', 'active']):
638
raise Exception("Can't modify state for extended")
640
with self._patch_ncclient(
641
'connect.return_value.edit_config.side_effect',
643
with self._create_resources() as result:
644
self.assertEqual(result.status_int, wexc.HTTPOk.code)
646
def mock_edit_config_b(target, config):
647
if all(word in config for word in ['no', 'shutdown']):
648
raise Exception("Command is only allowed on VLAN")
650
with self._patch_ncclient(
651
'connect.return_value.edit_config.side_effect',
653
with self._create_resources() as result:
654
self.assertEqual(result.status_int, wexc.HTTPOk.code)
656
def test_nexus_vlan_config_rollback(self):
657
"""Test rollback following Nexus VLAN state config failure.
659
Test that the Cisco Nexus plugin correctly deletes the VLAN
660
on the Nexus switch when the 'state active' command fails (for
661
a reason other than state configuration change is rejected
662
for the extended VLAN range).
665
vlan_state_configs = ['state active', 'no shutdown']
666
for config in vlan_state_configs:
667
with self._patch_ncclient(
668
'connect.return_value.edit_config.side_effect',
669
self._config_dependent_side_effect(config, ValueError)):
670
with self._create_resources() as result:
671
# Confirm that the last configuration sent to the Nexus
672
# switch was deletion of the VLAN.
674
self._is_in_last_nexus_cfg(['<no>', '<vlan>'])
676
self._assertExpectedHTTP(result.status_int,
677
c_exc.NexusConfigFailed)
679
def test_nexus_host_not_configured(self):
680
"""Test handling of a NexusComputeHostNotConfigured exception.
682
Test the Cisco NexusComputeHostNotConfigured exception by using
683
a fictitious host name during port creation.
686
with self._create_resources(host_id='fake_host',
687
expected_failure=True) as result:
688
self._assertExpectedHTTP(result.status_int,
689
c_exc.NexusComputeHostNotConfigured)
691
def test_nexus_missing_fields(self):
692
"""Test handling of a NexusMissingRequiredFields exception.
694
Test the Cisco NexusMissingRequiredFields exception by using
695
empty device_id value during port creation.
698
with self._create_resources(device_id='',
699
expected_failure=True) as result:
700
self._assertExpectedHTTP(result.status_int,
701
c_exc.NexusMissingRequiredFields)
703
def test_update_port_mac(self):
704
# REVISIT: test passes, but is back-end OK?
706
portbindings.HOST_ID: COMP_HOST_NAME,
707
'device_id': DEVICE_ID_1,
709
arg_list = (portbindings.HOST_ID, 'device_id',)
710
self.check_update_port_mac(host_arg=host_arg, arg_list=arg_list)
713
class TestCiscoNetworksV2(CiscoML2MechanismTestCase,
714
test_ml2_plugin.TestMl2NetworksV2):
716
def test_create_networks_bulk_emulated_plugin_failure(self):
717
real_has_attr = hasattr
719
def fakehasattr(item, attr):
720
if attr.endswith('__native_bulk_support'):
722
return real_has_attr(item, attr)
724
plugin_obj = manager.NeutronManager.get_plugin()
725
orig = plugin_obj.create_network
726
#ensures the API choose the emulation code path
727
with mock.patch('__builtin__.hasattr',
729
with mock.patch.object(plugin_obj,
730
'_create_network_db') as patched_plugin:
731
def side_effect(*args, **kwargs):
732
return self._fail_second_call(patched_plugin, orig,
734
patched_plugin.side_effect = side_effect
735
res = self._create_network_bulk(self.fmt, 2, 'test', True)
736
LOG.debug("response is %s" % res)
737
# We expect an internal server error as we injected a fault
738
self._validate_behavior_on_bulk_failure(
741
wexc.HTTPInternalServerError.code)
743
def test_create_networks_bulk_native_plugin_failure(self):
744
if self._skip_native_bulk:
745
self.skipTest("Plugin does not support native bulk network create")
746
plugin_obj = manager.NeutronManager.get_plugin()
747
orig = plugin_obj.create_network
748
with mock.patch.object(plugin_obj,
749
'_create_network_db') as patched_plugin:
751
def side_effect(*args, **kwargs):
752
return self._fail_second_call(patched_plugin, orig,
755
patched_plugin.side_effect = side_effect
756
res = self._create_network_bulk(self.fmt, 2, 'test', True)
757
# We expect an internal server error as we injected a fault
758
self._validate_behavior_on_bulk_failure(
761
wexc.HTTPInternalServerError.code)
764
class TestCiscoSubnetsV2(CiscoML2MechanismTestCase,
765
test_ml2_plugin.TestMl2SubnetsV2):
767
def test_create_subnets_bulk_emulated_plugin_failure(self):
768
real_has_attr = hasattr
770
#ensures the API choose the emulation code path
771
def fakehasattr(item, attr):
772
if attr.endswith('__native_bulk_support'):
774
return real_has_attr(item, attr)
776
with mock.patch('__builtin__.hasattr',
778
plugin_obj = manager.NeutronManager.get_plugin()
779
orig = plugin_obj.create_subnet
780
with mock.patch.object(plugin_obj,
781
'_create_subnet_db') as patched_plugin:
783
def side_effect(*args, **kwargs):
784
self._fail_second_call(patched_plugin, orig,
787
patched_plugin.side_effect = side_effect
788
with self.network() as net:
789
res = self._create_subnet_bulk(self.fmt, 2,
790
net['network']['id'],
792
# We expect an internal server error as we injected a fault
793
self._validate_behavior_on_bulk_failure(
796
wexc.HTTPInternalServerError.code)
798
def test_create_subnets_bulk_native_plugin_failure(self):
799
if self._skip_native_bulk:
800
self.skipTest("Plugin does not support native bulk subnet create")
801
plugin_obj = manager.NeutronManager.get_plugin()
802
orig = plugin_obj.create_subnet
803
with mock.patch.object(plugin_obj,
804
'_create_subnet_db') as patched_plugin:
805
def side_effect(*args, **kwargs):
806
return self._fail_second_call(patched_plugin, orig,
809
patched_plugin.side_effect = side_effect
810
with self.network() as net:
811
res = self._create_subnet_bulk(self.fmt, 2,
812
net['network']['id'],
815
# We expect an internal server error as we injected a fault
816
self._validate_behavior_on_bulk_failure(
819
wexc.HTTPInternalServerError.code)
added
removed
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_mech.py
