Merge "Remove duplicate test utilities"

This commit is contained in:
Zuul
2025-11-25 14:47:22 +00:00
committed by Gerrit Code Review
3 changed files with 59 additions and 143 deletions

View File

@@ -11,7 +11,6 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
#
"""Test module module""" """Test module module"""
@@ -19,10 +18,18 @@ import sys
from unittest import mock from unittest import mock
from openstackclient.common import module as osc_module from openstackclient.common import module as osc_module
from openstackclient.tests.unit import fakes
from openstackclient.tests.unit import utils from openstackclient.tests.unit import utils
class FakeModule:
def __init__(self, name, version):
self.name = name
self.__version__ = version
# Workaround for openstacksdk case
self.version = mock.Mock()
self.version.__version__ = version
# NOTE(dtroyer): module_1 must match the version list filter (not --all) # NOTE(dtroyer): module_1 must match the version list filter (not --all)
# currently == '*client*' # currently == '*client*'
module_name_1 = 'fakeclient' module_name_1 = 'fakeclient'
@@ -45,11 +52,11 @@ module_version_5 = '0.0.1'
MODULES = { MODULES = {
'sys': sys, 'sys': sys,
module_name_1: fakes.FakeModule(module_name_1, module_version_1), module_name_1: FakeModule(module_name_1, module_version_1),
module_name_2: fakes.FakeModule(module_name_2, module_version_2), module_name_2: FakeModule(module_name_2, module_version_2),
module_name_3: fakes.FakeModule(module_name_3, module_version_3), module_name_3: FakeModule(module_name_3, module_version_3),
module_name_4: fakes.FakeModule(module_name_4, module_version_4), module_name_4: FakeModule(module_name_4, module_version_4),
module_name_5: fakes.FakeModule(module_name_5, module_version_5), module_name_5: FakeModule(module_name_5, module_version_5),
} }

View File

@@ -12,13 +12,43 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
# TODO(stephenfin): Remove the contents of this module in favour of the osc_lib
# version once our min version is bumped to 4.3.0
import json import json
import sys
from unittest import mock from unittest import mock
from keystoneauth1 import fixture from keystoneauth1 import fixture
from osc_lib.tests.fakes import (
FakeApp,
FakeClientManager as BaseFakeClientManager,
FakeLog,
FakeOptions,
FakeResource as BaseFakeResource,
FakeStdout,
)
import requests import requests
__all__ = [
'AUTH_TOKEN',
'AUTH_URL',
'INTERFACE',
'PASSWORD',
'PROJECT_NAME',
'REGION_NAME',
'TEST_RESPONSE_DICT',
'TEST_RESPONSE_DICT_V3',
'TEST_VERSIONS',
'USERNAME',
'VERSION',
'FakeApp',
'FakeClientManager',
'FakeLog',
'FakeOptions',
'FakeResource',
'FakeResponse',
'FakeStdout',
]
AUTH_TOKEN = "foobar" AUTH_TOKEN = "foobar"
AUTH_URL = "http://0.0.0.0" AUTH_URL = "http://0.0.0.0"
@@ -47,79 +77,15 @@ TEST_RESPONSE_DICT_V3.set_project_scope()
TEST_VERSIONS = fixture.DiscoveryList(href=AUTH_URL) TEST_VERSIONS = fixture.DiscoveryList(href=AUTH_URL)
class FakeStdout: class FakeClientManager(BaseFakeClientManager):
def __init__(self):
self.content = []
def write(self, text):
self.content.append(text)
def make_string(self):
result = ''
for line in self.content:
result = result + line
return result
class FakeLog:
def __init__(self):
self.messages = {}
def debug(self, msg):
self.messages['debug'] = msg
def info(self, msg):
self.messages['info'] = msg
def warning(self, msg):
self.messages['warning'] = msg
def error(self, msg):
self.messages['error'] = msg
def critical(self, msg):
self.messages['critical'] = msg
class FakeApp:
def __init__(self, _stdout, _log):
self.stdout = _stdout
self.client_manager = None
self.api_version = {}
self.stdin = sys.stdin
self.stdout = _stdout or sys.stdout
self.stderr = sys.stderr
self.log = _log
class FakeOptions:
def __init__(self, **kwargs):
self.os_beta_command = False
class FakeClient:
def __init__(self, **kwargs):
self.endpoint = kwargs['endpoint']
self.token = kwargs['token']
class FakeClientManager:
_api_version = { _api_version = {
'image': '2', 'image': '2',
} }
def __init__(self): def __init__(self):
self.compute = None super().__init__()
self.identity = None
self.image = None
self.object_store = None
self.volume = None
self.network = None
self.sdk_connection = mock.Mock()
self.session = None self.sdk_connection = mock.Mock()
self.auth_ref = None
self.auth_plugin_name = None
self.network_endpoint_enabled = True self.network_endpoint_enabled = True
self.compute_endpoint_enabled = True self.compute_endpoint_enabled = True
@@ -158,64 +124,7 @@ class FakeClientManager:
return self.volume_endpoint_enabled return self.volume_endpoint_enabled
class FakeModule: class FakeResource(BaseFakeResource):
def __init__(self, name, version):
self.name = name
self.__version__ = version
# Workaround for openstacksdk case
self.version = mock.Mock()
self.version.__version__ = version
class FakeResource:
def __init__(self, manager=None, info=None, loaded=False, methods=None):
"""Set attributes and methods for a resource.
:param manager:
The resource manager
:param Dictionary info:
A dictionary with all attributes
:param bool loaded:
True if the resource is loaded in memory
:param Dictionary methods:
A dictionary with all methods
"""
info = info or {}
methods = methods or {}
self.__name__ = type(self).__name__
self.manager = manager
self._info = info
self._add_details(info)
self._add_methods(methods)
self._loaded = loaded
def _add_details(self, info):
for k, v in info.items():
setattr(self, k, v)
def _add_methods(self, methods):
"""Fake methods with MagicMock objects.
For each <@key, @value> pairs in methods, add an callable MagicMock
object named @key as an attribute, and set the mock's return_value to
@value. When users access the attribute with (), @value will be
returned, which looks like a function call.
"""
for name, ret in methods.items():
method = mock.Mock(return_value=ret)
setattr(self, name, method)
def __repr__(self):
reprkeys = sorted(
k for k in self.__dict__.keys() if k[0] != '_' and k != 'manager'
)
info = ", ".join(f"{k}={getattr(self, k)}" for k in reprkeys)
return f"<{self.__class__.__name__} {info}>"
def keys(self):
return self._info.keys()
def to_dict(self): def to_dict(self):
return self._info return self._info
@@ -247,11 +156,3 @@ class FakeResponse(requests.Response):
self._content = json.dumps(data) self._content = json.dumps(data)
if not isinstance(self._content, bytes): if not isinstance(self._content, bytes):
self._content = self._content.encode() self._content = self._content.encode()
class FakeModel(dict):
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)

View File

@@ -693,6 +693,14 @@ class TestIdentityv3(
): ... ): ...
class FakeModel(dict):
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)
# We don't use FakeClientMixin since we want a different fake legacy client # We don't use FakeClientMixin since we want a different fake legacy client
class TestFederatedIdentity(utils.TestCommand): class TestFederatedIdentity(utils.TestCommand):
def setUp(self): def setUp(self):
@@ -1075,7 +1083,7 @@ class FakeEndpoint:
# Overwrite default attributes if there are some attributes set # Overwrite default attributes if there are some attributes set
endpoint_filter_info.update(attrs) endpoint_filter_info.update(attrs)
endpoint_filter = fakes.FakeModel(copy.deepcopy(endpoint_filter_info)) endpoint_filter = FakeModel(copy.deepcopy(endpoint_filter_info))
return endpoint_filter return endpoint_filter
@@ -1133,7 +1141,7 @@ class FakeEndpointGroup:
# Overwrite default attributes if there are some attributes set # Overwrite default attributes if there are some attributes set
endpointgroup_filter_info.update(attrs) endpointgroup_filter_info.update(attrs)
endpointgroup_filter = fakes.FakeModel( endpointgroup_filter = FakeModel(
copy.deepcopy(endpointgroup_filter_info) copy.deepcopy(endpointgroup_filter_info)
) )