openstack/neutron
Code Issues Proposed changes

Merge "Revert "Drop `AddressScope.shared` field""

Browse Source
This commit is contained in:
Zuul authored and Gerrit Code Review committed 2026-10-02 23:51:10 +00:00
2 parents a6c792b9f6 fa8671a014
commit 2947b28f1a
7 files changed
+63 -97
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

neutron/db/address_scope_db.py
+7 -2
View File
@@ -22,6 +22,7 @@ from neutron_lib.exceptions import address_scope as api_err
from oslo_log import log as logging
from oslo_utils import uuidutils
from neutron._i18n import _
from neutron.extensions import address_scope as ext_address_scope
from neutron.objects import address_scope as obj_addr_scope
from neutron.objects import base as base_obj
@@ -78,10 +79,10 @@ class AddressScopeDbMixin(ext_address_scope.AddressScopePluginBase):
'dictionary, using tenant_id instead. This support '
'has been deprecated and will be removed in a '
'future release.')
a_s.pop('shared', None)
pool_args = {'project_id': a_s['project_id'],
'id': address_scope_id,
'name': a_s['name'],
'shared': a_s['shared'],
'ip_version': a_s['ip_version']}
address_scope = obj_addr_scope.AddressScope(context, **pool_args)
address_scope.create()
@@ -89,8 +90,12 @@ class AddressScopeDbMixin(ext_address_scope.AddressScopePluginBase):
def update_address_scope(self, context, id, address_scope):
a_s = address_scope['address_scope']
a_s.pop('shared', None)
address_scope = self._get_address_scope(context, id)
if address_scope.shared and not a_s.get('shared', True):
reason = _("Shared address scope can't be unshared")
raise api_err.AddressScopeUpdateError(
address_scope_id=id, reason=reason)
address_scope.update_fields(a_s)
address_scope.update()
return self._make_address_scope_dict(address_scope)
neutron/objects/address_scope.py
+2 -17
View File
@@ -14,7 +14,6 @@
from neutron_lib.db import api as db_api
from neutron_lib.objects import common_types
from oslo_utils import versionutils
from oslo_versionedobjects import fields as obj_fields
from neutron.db.models import address_scope as models
@@ -39,12 +38,10 @@ class AddressScopeRBAC(rbac.RBACBaseObject):
class AddressScope(rbac_db.NeutronRbacObject):
# Version 1.0: Initial version
# Version 1.1: Add RBAC support
# Version 1.2: Remove "shared" field
VERSION = '1.2'
VERSION = '1.1'
# required by RbacNeutronMetaclass
rbac_db_cls = AddressScopeRBAC
_rbac_shared_field = False
db_model = models.AddressScope
@@ -52,6 +49,7 @@ class AddressScope(rbac_db.NeutronRbacObject):
'id': common_types.UUIDField(),
'project_id': obj_fields.StringField(nullable=True),
'name': obj_fields.StringField(),
'shared': obj_fields.BooleanField(),
'ip_version': common_types.IPVersionEnumField(),
}
@@ -78,16 +76,3 @@ class AddressScope(rbac_db.NeutronRbacObject):
snp_objs = subnetpool.SubnetPool.get_objects(
context, address_scope_id=obj_id, fields=['project_id'])
return {snp['project_id'] for snp in snp_objs}
def obj_make_compatible(self, primitive, target_version):
_target_version = versionutils.convert_version_to_tuple(target_version)
if _target_version >= (1, 2):
primitive.pop('shared', None)
elif 'shared' not in primitive:
if self.obj_context and self.id:
primitive['shared'] = self.get_shared_with_project(
self.obj_context.elevated(), self.rbac_db_cls,
self.id, self.project_id)
else:
primitive['shared'] = False
return super().obj_make_compatible(primitive, target_version)
neutron/objects/rbac_db.py
+11 -30
View File
@@ -248,13 +248,11 @@ class RbacNeutronDbObjectMixin(rbac_db_mixin.RbacPluginMixin,
return self.obj_context.session.delete(shared_prev)
def from_db_object(self, db_obj):
if getattr(self.__class__, '_rbac_shared_field', True):
self._load_shared(db_obj)
self._load_shared(db_obj)
super().from_db_object(db_obj)
def obj_load_attr(self, attrname):
if (attrname == 'shared' and
getattr(self.__class__, '_rbac_shared_field', True)):
if attrname == 'shared':
return self._load_shared()
super().obj_load_attr(attrname)
@@ -292,9 +290,8 @@ def _update_hook(self, update_orig):
# _update_post method because update() will reset all those changes
obj_changes = self.obj_get_changes()
update_orig(self)
if getattr(self.__class__, '_rbac_shared_field', True):
_update_post(self, obj_changes)
self._load_shared(db_obj=self.db_obj)
_update_post(self, obj_changes)
self._load_shared(db_obj=self.db_obj)
def _create_post(self):
@@ -305,9 +302,8 @@ def _create_post(self):
def _create_hook(self, orig_create):
with self.db_context_writer(self.obj_context):
orig_create(self)
if getattr(self.__class__, '_rbac_shared_field', True):
_create_post(self)
self._load_shared(db_obj=self.db_obj)
_create_post(self)
self._load_shared(db_obj=self.db_obj)
def _to_dict_hook(self, to_dict_orig):
@@ -351,12 +347,6 @@ class RbacNeutronMetaclass(type):
reason=_('already a synthetic attribute'))
dct['synthetic_fields'].append('shared')
@classmethod
def init_synthetic_fields(cls, bases, dct):
if not dct.get('synthetic_fields', None):
synthetic_attr = cls.get_attribute('synthetic_fields', bases, dct)
dct['synthetic_fields'] = synthetic_attr or []
@staticmethod
def subscribe_to_rbac_events(class_instance):
for e in (events.BEFORE_CREATE, events.BEFORE_UPDATE,
@@ -365,8 +355,8 @@ class RbacNeutronMetaclass(type):
resources.RBAC_POLICY, e)
@staticmethod
def validate_existing_attrs(cls_name, dct, uses_shared_field=True):
if uses_shared_field and 'shared' not in dct['fields']:
def validate_existing_attrs(cls_name, dct):
if 'shared' not in dct['fields']:
raise KeyError(_('No shared key in %s fields') % cls_name)
if 'rbac_db_cls' not in dct:
raise AttributeError(_('rbac_db_cls not found in %s') % cls_name)
@@ -389,20 +379,11 @@ class RbacNeutronMetaclass(type):
dct[orig_method_name] = hook_method
def __new__(cls, name, bases, dct):
uses_shared_field = dct.get('_rbac_shared_field')
if uses_shared_field is None:
uses_shared_field = cls._get_attribute('_rbac_shared_field', bases)
if uses_shared_field is None:
uses_shared_field = True
cls.validate_existing_attrs(name, dct, uses_shared_field)
if uses_shared_field:
cls.update_synthetic_fields(bases, dct)
else:
cls.init_synthetic_fields(bases, dct)
cls.validate_existing_attrs(name, dct)
cls.update_synthetic_fields(bases, dct)
cls.replace_class_methods_with_hooks(bases, dct)
klass = type(name, (RbacNeutronDbObjectMixin,) + bases, dct)
if uses_shared_field:
klass.add_extra_filter_name('shared')
klass.add_extra_filter_name('shared')
cls.subscribe_to_rbac_events(klass)
return klass
neutron/tests/unit/extensions/test_address_scope.py
+42 -18
View File
@@ -27,9 +27,7 @@ import webob.exc
from neutron.db import address_scope_db
from neutron.db import db_base_plugin_v2
from neutron.db import rbac_db_models
from neutron.extensions import address_scope as ext_address_scope
from neutron.objects import address_scope as obj_addr_scope
from neutron.tests.common import test_db_base_plugin_v2
DB_PLUGIN_KLASS = ('neutron.tests.unit.extensions.test_address_scope.'
@@ -98,16 +96,6 @@ class AddressScopeTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
self._compare_resource(addr_scope, expected, 'address_scope')
return addr_scope
def _share_address_scope(self, addr_scope, target_project='*'):
rbac = obj_addr_scope.AddressScopeRBAC(
context.get_admin_context(),
project_id=addr_scope['address_scope']['project_id'],
object_id=addr_scope['address_scope']['id'],
action=rbac_db_models.ACCESS_SHARED,
target_project=target_project,
)
rbac.create()
def _test_update_address_scope(self, addr_scope_id, data, admin=False,
expected=None, project_id=None):
update_req = self.new_update_request(
@@ -142,6 +130,7 @@ class TestAddressScope(AddressScopeTestCase):
def test_create_address_scope_ipv4(self):
expected_addr_scope = {'name': 'foo-address-scope',
'project_id': self._project_id,
'shared': False,
'ip_version': constants.IP_VERSION_4}
self._test_create_address_scope(name='foo-address-scope',
expected=expected_addr_scope)
@@ -149,6 +138,7 @@ class TestAddressScope(AddressScopeTestCase):
def test_create_address_scope_ipv6(self):
expected_addr_scope = {'name': 'foo-address-scope',
'project_id': self._project_id,
'shared': False,
'ip_version': constants.IP_VERSION_6}
self._test_create_address_scope(constants.IP_VERSION_6,
name='foo-address-scope',
@@ -156,12 +146,25 @@ class TestAddressScope(AddressScopeTestCase):
def test_create_address_scope_empty_name(self):
expected_addr_scope = {'name': '',
'project_id': self._project_id}
'project_id': self._project_id,
'shared': False}
self._test_create_address_scope(name='', expected=expected_addr_scope)
# no name specified
self._test_create_address_scope(expected=expected_addr_scope)
def test_create_address_scope_shared_admin(self):
expected_addr_scope = {'name': 'foo-address-scope', 'shared': True}
self._test_create_address_scope(name='foo-address-scope', admin=True,
shared=True,
expected=expected_addr_scope)
def test_created_address_scope_shared_non_admin(self):
res = self._create_address_scope(self.fmt, name='foo-address-scope',
project_id=self._project_id,
admin=False, shared=True)
self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
def test_created_address_scope_specify_id(self):
res = self._create_address_scope(self.fmt, name='foo-address-scope',
id='foo-id')
@@ -179,6 +182,28 @@ class TestAddressScope(AddressScopeTestCase):
self._test_update_address_scope(addr_scope['address_scope']['id'],
data, expected=data['address_scope'])
def test_update_address_scope_shared_true_admin(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
data = {'address_scope': {'shared': True}}
self._test_update_address_scope(addr_scope['address_scope']['id'],
data, admin=True,
expected=data['address_scope'])
def test_update_address_scope_shared_true_non_admin(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
data = {'address_scope': {'shared': True}}
res = self._test_update_address_scope(
addr_scope['address_scope']['id'], data, admin=False)
self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
def test_update_address_scope_shared_false_admin(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope',
admin=True, shared=True)
data = {'address_scope': {'shared': False}}
res = self._test_update_address_scope(
addr_scope['address_scope']['id'], data, admin=True)
self.assertEqual(webob.exc.HTTPClientError.code, res.status_int)
def test_get_address_scope(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
req = self.new_show_request('address-scopes',
@@ -197,8 +222,8 @@ class TestAddressScope(AddressScopeTestCase):
self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)
def test_get_address_scope_different_projects_shared(self):
addr_scope = self._test_create_address_scope(name='foo-address-scope')
self._share_address_scope(addr_scope)
addr_scope = self._test_create_address_scope(name='foo-address-scope',
shared=True, admin=True)
req = self.new_show_request('address-scopes',
addr_scope['address_scope']['id'])
neutron_context = context.Context('', 'test-project-2')
@@ -215,9 +240,8 @@ class TestAddressScope(AddressScopeTestCase):
self.assertEqual(2, len(res['address_scopes']))
def test_list_address_scopes_different_projects_shared(self):
addr_scope = self._test_create_address_scope(
name='foo-address-scope')
self._share_address_scope(addr_scope)
self._test_create_address_scope(name='foo-address-scope', shared=True,
admin=True)
admin_res = self._list('address-scopes')
mortal_res = self._list(
'address-scopes', project_id='not-the-owner')
neutron/tests/unit/objects/test_address_scope.py
-23
View File
@@ -14,7 +14,6 @@
from neutron_lib import constants as lib_constants
from neutron.db import rbac_db_models
from neutron.objects import address_scope
from neutron.tests.unit.objects import test_base
from neutron.tests.unit.objects import test_rbac
@@ -31,28 +30,6 @@ class AddressScopeDbObjectTestCase(test_base.BaseDbObjectTestCase,
_test_class = address_scope.AddressScope
def test_object_version_degradation_1_2_to_1_1_drop_shared_field(self):
as_obj = address_scope.AddressScope()
as_obj_1_2 = as_obj.obj_to_primitive(target_version='1.2')
self.assertNotIn('shared', as_obj_1_2['versioned_object.data'])
as_obj_1_1 = as_obj.obj_to_primitive(target_version='1.1')
self.assertIn('shared', as_obj_1_1['versioned_object.data'])
self.assertFalse(as_obj_1_1['versioned_object.data']['shared'])
def test_object_version_degradation_1_2_to_1_1_shared_from_rbac(self):
as_obj = self._make_object(self.obj_fields[0])
as_obj.create()
rbac = address_scope.AddressScopeRBAC(
self.context, project_id=as_obj.project_id,
object_id=as_obj.id,
action=rbac_db_models.ACCESS_SHARED,
target_project='*')
rbac.create()
as_obj_1_1 = as_obj.obj_to_primitive(target_version='1.1')
self.assertTrue(as_obj_1_1['versioned_object.data']['shared'])
class AddressScopeRBACDbObjectTestCase(test_rbac.TestRBACObjectMixin,
test_base.BaseDbObjectTestCase,
neutron/tests/unit/objects/test_objects.py
+1 -1
View File
@@ -29,7 +29,7 @@ object_data = {
'AddressAssociation': '1.0-b92160a3dd2fb7b951adcd2e6ae1665a',
'AddressGroup': '1.2-1ddbf0a9f61785033ce31818ac62687e',
'AddressGroupRBAC': '1.1-be82ed54376b85ee4f963d479ac48c91',
'AddressScope': '1.2-2019fecf9701570128fba5d8a9f2f9e3',
'AddressScope': '1.1-dd0dfdb67775892d3adc090e28e43bd8',
'AddressScopeRBAC': '1.1-be82ed54376b85ee4f963d479ac48c91',
'Agent': '1.1-4feb0abf8732a17c16dee962091d9203',
'AllowedAddressPair': '1.0-9f9186b6f952fbf31d257b0458b852c0',
releasenotes/notes/drop-address-scope-shared-field-2026547.yaml
-6
View File
@@ -1,6 +0,0 @@
---
upgrade:
- |
The ``shared`` attribute is no longer accepted when creating or updating
address scopes. Use RBAC policies to share address scopes across
projects instead.