openstack
/
neutron
Code Issues Proposed changes

Fix ML2 test extension driver API test cases 

The test cases were asserting bad behavior by the way the
test driver was designed. Rather than ensuring that the extension
was extending the resource in each case with the correct value,
it was asserting that different values came back from
update/create/show. In other words, it was asserting API
inconsistency which interferred with the dependent patch.

This patch fixes that by having the extension store the values requested
in the API and then use those same values for extending the result
whether it be for updates, creates, or shows.

Change-Id: Ib1517102e3fb776fca818acfddd9e0f733563fb3
Partial-Bug: #1625981
This commit is contained in:
Kevin Benton 2016-09-20 01:45:36 -07:00
parent 9019ff2f74
commit 7e82296042
2 changed files with 30 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
neutron/tests/unit/plugins/ml2/drivers/ext_test.py
View File

@ -17,6 +17,7 @@
# under the License.
from neutron_lib.api import validators
from neutron_lib import constants
from neutron_lib.db import model_base
import oslo_db.sqlalchemy.session
import sqlalchemy as sa
@ -42,9 +43,8 @@ class TestExtensionDriverBase(driver_api.ExtensionDriver):
class TestExtensionDriver(TestExtensionDriverBase):
def initialize(self):
super(TestExtensionDriver, self).initialize()
self.network_extension = 'Test_Network_Extension'
self.subnet_extension = 'Test_Subnet_Extension'
self.port_extension = 'Test_Port_Extension'
# keep track of values
self.val_by_id = {}
def _check_create(self, session, data, result):
assert(isinstance(session, oslo_db.sqlalchemy.session.Session))
@ -67,50 +67,59 @@ class TestExtensionDriver(TestExtensionDriverBase):
assert(isinstance(db_entry, expected_db_entry_class))
assert(db_entry.id == result['id'])
def _store_change(self, result, data, field):
if field in data and data[field] != constants.ATTR_NOT_SPECIFIED:
self.val_by_id[result['id']] = data[field]
elif result['id'] not in self.val_by_id:
self.val_by_id[result['id']] = 'default_%s' % field
def process_create_network(self, plugin_context, data, result):
session = plugin_context.session
self._check_create(session, data, result)
result['network_extension'] = self.network_extension + '_create'
self._store_change(result, data, 'network_extension')
result['network_extension'] = self.val_by_id[result['id']]
def process_update_network(self, plugin_context, data, result):
session = plugin_context.session
self._check_update(session, data, result)
self.network_extension = data['network_extension']
result['network_extension'] = self.network_extension + '_update'
self._store_change(result, data, 'network_extension')
result['network_extension'] = self.val_by_id[result['id']]
def extend_network_dict(self, session, net_db, result):
self._check_extend(session, result, net_db, models_v2.Network)
result['network_extension'] = self.network_extension + '_extend'
result['network_extension'] = self.val_by_id.get(result['id'])
def process_create_subnet(self, plugin_context, data, result):
session = plugin_context.session
self._check_create(session, data, result)
result['subnet_extension'] = self.subnet_extension + '_create'
self._store_change(result, data, 'subnet_extension')
result['subnet_extension'] = self.val_by_id[result['id']]
def process_update_subnet(self, plugin_context, data, result):
session = plugin_context.session
self._check_update(session, data, result)
self.subnet_extension = data['subnet_extension']
result['subnet_extension'] = self.subnet_extension + '_update'
self._store_change(result, data, 'subnet_extension')
result['subnet_extension'] = self.val_by_id[result['id']]
def extend_subnet_dict(self, session, subnet_db, result):
self._check_extend(session, result, subnet_db, models_v2.Subnet)
result['subnet_extension'] = self.subnet_extension + '_extend'
result['subnet_extension'] = self.val_by_id.get(result['id'])
def process_create_port(self, plugin_context, data, result):
session = plugin_context.session
self._check_create(session, data, result)
result['port_extension'] = self.port_extension + '_create'
self._store_change(result, data, 'port_extension')
result['port_extension'] = self.val_by_id[result['id']]
def process_update_port(self, plugin_context, data, result):
session = plugin_context.session
self._check_update(session, data, result)
self.port_extension = data['port_extension']
result['port_extension'] = self.port_extension + '_update'
self._store_change(result, data, 'port_extension')
result['port_extension'] = self.val_by_id[result['id']]
def extend_port_dict(self, session, port_db, result):
self._check_extend(session, result, port_db, models_v2.Port)
result['port_extension'] = self.port_extension + '_extend'
result['port_extension'] = self.val_by_id.get(result['id'])
class TestNetworkExtension(model_base.BASEV2):

12
neutron/tests/unit/plugins/ml2/test_extension_driver_api.py
View File

@ -94,14 +94,14 @@ class ExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase):
# Test list networks
res = self._list('networks')
val = res['networks'][0].get('network_extension')
self.assertEqual('Test_Network_Extension_extend', val)
self.assertEqual('default_network_extension', val)
# Test network update
data = {'network':
{'network_extension': 'Test_Network_Extension_Update'}}
res = self._update('networks', network['network']['id'], data)
val = res['network'].get('network_extension')
self.assertEqual('Test_Network_Extension_Update_update', val)
self.assertEqual('Test_Network_Extension_Update', val)
def test_subnet_attr(self):
with self.subnet() as subnet:
@ -112,14 +112,14 @@ class ExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase):
# Test list subnets
res = self._list('subnets')
val = res['subnets'][0].get('subnet_extension')
self.assertEqual('Test_Subnet_Extension_extend', val)
self.assertEqual('default_subnet_extension', val)
# Test subnet update
data = {'subnet':
{'subnet_extension': 'Test_Subnet_Extension_Update'}}
res = self._update('subnets', subnet['subnet']['id'], data)
val = res['subnet'].get('subnet_extension')
self.assertEqual('Test_Subnet_Extension_Update_update', val)
self.assertEqual('Test_Subnet_Extension_Update', val)
def test_port_attr(self):
with self.port() as port:
@ -130,13 +130,13 @@ class ExtensionDriverTestCase(test_plugin.Ml2PluginV2TestCase):
# Test list ports
res = self._list('ports')
val = res['ports'][0].get('port_extension')
self.assertEqual('Test_Port_Extension_extend', val)
self.assertEqual('default_port_extension', val)
# Test port update
data = {'port': {'port_extension': 'Test_Port_Extension_Update'}}
res = self._update('ports', port['port']['id'], data)
val = res['port'].get('port_extension')
self.assertEqual('Test_Port_Extension_Update_update', val)
self.assertEqual('Test_Port_Extension_Update', val)
def test_extend_network_dict(self):
with mock.patch.object(ext_test.TestExtensionDriver,