Fix region query

The list region API in keystone has no "name" concept. The region name
is actually presented by the "id" field and is unique. Use get region
API instead.

Closes-Bug: #2131974
Change-Id: I42200f7ef56352ecfb8dbc22b6e1709333ccf274
Signed-off-by: Takashi Kajinami <kajinamit@oss.nttdata.com>
This commit is contained in:
Takashi Kajinami
2025-12-03 22:07:23 +09:00
parent f1ce4022d6
commit cde63ed003
3 changed files with 20 additions and 51 deletions

View File

@@ -36,13 +36,15 @@ def _fake_services(
)
def _fake_regions(
**query: dict[str, Any],
) -> Generator[_region.Region, None, None]:
# we are the only ones calling this, so we know exactly what we should be
# calling it with
assert set(query) == {'name'}
yield sdk_fakes.generate_fake_resource(_region.Region, name=query['name'])
def _fake_region(
region: str | _region.Region,
) -> _region.Region:
if isinstance(region, str):
region_id = region
else:
region_id = region.id
return sdk_fakes.generate_fake_resource(_region.Region, id=region_id)
def _fake_endpoints(
@@ -156,7 +158,7 @@ class LimitFixture(fixtures.Fixture):
# Then, requests by name
self.mock_conn.services.side_effect = _fake_services
self.mock_conn.regions.side_effect = _fake_regions
self.mock_conn.get_region.side_effect = _fake_region
self.mock_conn.endpoints.side_effect = _fake_endpoints
# Finally, fake the actual limits and registered limits calls

View File

@@ -396,18 +396,14 @@ class _EnforcerUtils:
# find region (if any)
if CONF.oslo_limit.endpoint_region_name is not None:
regions = self.connection.regions( # type: ignore
name=CONF.oslo_limit.endpoint_region_name
)
regions = list(regions)
if len(regions) > 1:
raise ValueError("Multiple regions found")
if len(regions) == 0:
try:
region = self.connection.get_region( # type: ignore
CONF.oslo_limit.endpoint_region_name
)
except os_exceptions.ResourceNotFound:
raise ValueError("Region not found")
region_id = regions[0].id
region_id = region.id
else:
region_id = None

View File

@@ -535,7 +535,7 @@ class TestEnforcerUtils(base.BaseTestCase):
fake_endpoint = endpoint.Endpoint()
self.mock_conn.endpoints.return_value = [fake_endpoint]
fake_region = region.Region(id='REGION_ID')
self.mock_conn.regions.return_value = [fake_region]
self.mock_conn.get_region.return_value = fake_region
utils = limit._EnforcerUtils()
@@ -544,7 +544,7 @@ class TestEnforcerUtils(base.BaseTestCase):
self.mock_conn.services.assert_called_once_with(
type='SERVICE_TYPE', name='SERVICE_NAME'
)
self.mock_conn.regions.assert_called_once_with(name='regionOne')
self.mock_conn.get_region.assert_called_once_with('regionOne')
self.mock_conn.endpoints.assert_called_once_with(
service_id='SERVICE_ID', region_id='REGION_ID', interface='public'
)
@@ -564,7 +564,7 @@ class TestEnforcerUtils(base.BaseTestCase):
self.mock_conn.services.return_value = [fake_service]
fake_endpoint = endpoint.Endpoint()
self.mock_conn.endpoints.return_value = [fake_endpoint]
self.mock_conn.regions.return_value = []
self.mock_conn.get_region.side_effect = os_exceptions.ResourceNotFound
self.assertRaises(ValueError, limit._EnforcerUtils)
@@ -572,36 +572,7 @@ class TestEnforcerUtils(base.BaseTestCase):
self.mock_conn.services.assert_called_once_with(
type='SERVICE_TYPE', name='SERVICE_NAME'
)
self.mock_conn.regions.assert_called_once_with(name='regionOne')
self.mock_conn.endpoints.assert_not_called()
def test_get_endpoint_lookup_with_mutliple_regions(self):
self.config_fixture.config(group='oslo_limit', endpoint_id=None)
self.config_fixture.config(
group='oslo_limit', endpoint_service_type='SERVICE_TYPE'
)
self.config_fixture.config(
group='oslo_limit', endpoint_service_name='SERVICE_NAME'
)
self.config_fixture.config(
group='oslo_limit', endpoint_region_name='regionOne'
)
fake_service = service.Service(id='SERVICE_ID')
self.mock_conn.services.return_value = [fake_service]
fake_endpoint = endpoint.Endpoint()
self.mock_conn.endpoints.return_value = [fake_endpoint]
self.mock_conn.regions.return_value = [
region.Region(id='REGION_ID1'),
region.Region(id='REGION_ID2'),
]
self.assertRaises(ValueError, limit._EnforcerUtils)
self.mock_conn.get_endpoint.assert_not_called()
self.mock_conn.services.assert_called_once_with(
type='SERVICE_TYPE', name='SERVICE_NAME'
)
self.mock_conn.regions.assert_called_once_with(name='regionOne')
self.mock_conn.get_region.assert_called_once_with('regionOne')
self.mock_conn.endpoints.assert_not_called()
def test_get_registered_limit_empty(self):