fixture: Mock out services, regions

Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
Change-Id: Iaf1812dc1bfb26758f758565480ac0391df62390
This commit is contained in:
Stephen Finucane
2025-11-26 12:15:13 +00:00
parent b8824a70c7
commit ebfae62773
2 changed files with 68 additions and 15 deletions

View File

@@ -28,6 +28,7 @@ repos:
hooks:
- id: mypy
additional_dependencies: [
'fixtures',
'keystoneauth',
'oslo.i18n',
'openstacksdk',

View File

@@ -11,16 +11,55 @@
# License for the specific language governing permissions and limitations
# under the License.
from collections.abc import Generator
from typing import Any
from unittest import mock
import fixtures as fixtures
from openstack.identity.v3 import endpoint as _endpoint
from openstack.identity.v3 import limit as _limit
from openstack.identity.v3 import region as _region
from openstack.identity.v3 import registered_limit as _registered_limit
from openstack.identity.v3 import service as _service
from openstack.test import fakes as sdk_fakes
class LimitFixture(fixtures.Fixture): # type: ignore
def _fake_services(
**query: dict[str, Any],
) -> Generator[_service.Service, None, None]:
# we are the only ones calling this, so we know exactly what we should be
# calling it with
assert set(query) == {'type', 'name'}
yield sdk_fakes.generate_fake_resource(
_service.Service, type=query['type'], name=query['name']
)
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_endpoints(
**query: dict[str, Any],
) -> Generator[_endpoint.Endpoint, None, None]:
# we are the only ones calling this, so we know exactly what we should be
# calling it with
assert set(query) == {'service_id', 'region_id', 'interface'}
yield sdk_fakes.generate_fake_resource(
_endpoint.Endpoint,
service_id=query['service_id'],
region_id=query['region_id'],
interface=query['interface'],
)
class LimitFixture(fixtures.Fixture):
def __init__(
self,
reglimits: dict[str, int],
@@ -47,17 +86,19 @@ class LimitFixture(fixtures.Fixture): # type: ignore
region_id: str | None = None,
resource_name: str | None = None,
) -> list[_registered_limit.RegisteredLimit]:
limits = []
registered_limits = []
for name, value in self.reglimits.items():
if resource_name and resource_name != name:
continue
limit = _registered_limit.RegisteredLimit() # type: ignore
limit.resource_name = name
limit.default_limit = value
limits.append(limit)
registered_limit = sdk_fakes.generate_fake_resource(
_registered_limit.RegisteredLimit,
resource_name=name,
default_limit=value,
)
registered_limits.append(registered_limit)
return limits
return registered_limits
def get_projlimit_objects(
self,
@@ -75,10 +116,12 @@ class LimitFixture(fixtures.Fixture): # type: ignore
if resource_name and resource_name != name:
continue
limit = _limit.Limit() # type: ignore
limit.project_id = proj_id
limit.resource_name = name
limit.resource_limit = value
limit = sdk_fakes.generate_fake_resource(
_limit.Limit,
resource_name=name,
resource_limit=value,
project_id=proj_id,
)
limits.append(limit)
return limits
@@ -102,12 +145,21 @@ class LimitFixture(fixtures.Fixture): # type: ignore
).mock
mock_gem.return_value = 'flat'
# Fake keystone endpoint; no per-service limit distinction
fake_endpoint = _endpoint.Endpoint() # type: ignore
fake_endpoint.service_id = "service_id"
fake_endpoint.region_id = "region_id"
# Fake keystone endpoint; this can be requested by ID or by name and we
# need to handle both. First, requests by ID
fake_endpoint = sdk_fakes.generate_fake_resource(
_endpoint.Endpoint,
service_id='service_id',
region_id='region_id',
)
self.mock_conn.get_endpoint.return_value = fake_endpoint
# Then, requests by name
self.mock_conn.services.side_effect = _fake_services
self.mock_conn.regions.side_effect = _fake_regions
self.mock_conn.endpoints.side_effect = _fake_endpoints
# Finally, fake the actual limits and registered limits calls
self.mock_conn.limits.side_effect = self.get_projlimit_objects
self.mock_conn.registered_limits.side_effect = (
self.get_reglimit_objects