Merge "Add new use_fixture function"

This commit is contained in:
Zuul 2022-01-03 12:42:00 +00:00 committed by Gerrit Code Review
commit 1d25a77a33
4 changed files with 89 additions and 27 deletions

View File

@ -75,6 +75,7 @@ remove_fixture = _fixture.remove_fixture
reset_fixture = _fixture.reset_fixture reset_fixture = _fixture.reset_fixture
setup_fixture = _fixture.setup_fixture setup_fixture = _fixture.setup_fixture
cleanup_fixture = _fixture.cleanup_fixture cleanup_fixture = _fixture.cleanup_fixture
use_fixture = _fixture.use_fixture
list_required_fixtures = _fixture.list_required_fixtures list_required_fixtures = _fixture.list_required_fixtures
SharedFixture = _fixture.SharedFixture SharedFixture = _fixture.SharedFixture
FixtureManager = _fixture.FixtureManager FixtureManager = _fixture.FixtureManager

View File

@ -27,6 +27,7 @@ import tobiko
from tobiko.common import _detail from tobiko.common import _detail
from tobiko.common import _deprecation from tobiko.common import _deprecation
from tobiko.common import _exception from tobiko.common import _exception
from tobiko.common import _testcase
LOG = log.getLogger(__name__) LOG = log.getLogger(__name__)
@ -41,7 +42,7 @@ def is_fixture(obj):
def get_fixture(obj: typing.Any, def get_fixture(obj: typing.Any,
fixture_id: typing.Any = None, fixture_id: typing.Any = None,
manager: 'FixtureManager' = None): manager: 'FixtureManager' = None) -> fixtures.Fixture:
'''Returns a fixture identified by given :param obj: '''Returns a fixture identified by given :param obj:
It returns registered fixture for given :param obj:. If none has been It returns registered fixture for given :param obj:. If none has been
@ -73,7 +74,7 @@ def get_fixture(obj: typing.Any,
return manager.get_fixture(obj, fixture_id=fixture_id) return manager.get_fixture(obj, fixture_id=fixture_id)
def get_fixture_name(obj): def get_fixture_name(obj) -> str:
'''Get unique fixture name''' '''Get unique fixture name'''
name = getattr(obj, '__tobiko_fixture_name__', None) name = getattr(obj, '__tobiko_fixture_name__', None)
if name is None: if name is None:
@ -85,7 +86,7 @@ def get_fixture_name(obj):
return name return name
def get_fixture_class(obj): def get_fixture_class(obj) -> typing.Type[fixtures.Fixture]:
'''Get fixture class''' '''Get fixture class'''
if isinstance(obj, str): if isinstance(obj, str):
obj = tobiko.load_object(obj) obj = tobiko.load_object(obj)
@ -102,13 +103,15 @@ def get_fixture_dir(obj):
return os.path.dirname(inspect.getfile(get_fixture_class(obj))) return os.path.dirname(inspect.getfile(get_fixture_class(obj)))
def remove_fixture(obj, fixture_id=None, manager=None): def remove_fixture(obj, fixture_id=None, manager=None) \
-> typing.Optional[fixtures.Fixture]:
'''Unregister fixture identified by given :param obj: if any''' '''Unregister fixture identified by given :param obj: if any'''
manager = manager or FIXTURES manager = manager or FIXTURES
return manager.remove_fixture(obj, fixture_id=fixture_id) return manager.remove_fixture(obj, fixture_id=fixture_id)
def setup_fixture(obj, fixture_id=None, manager=None, alternative=None): def setup_fixture(obj, fixture_id=None, manager=None, alternative=None) \
-> fixtures.Fixture:
'''Get registered fixture and setup it up''' '''Get registered fixture and setup it up'''
if alternative is None: if alternative is None:
objs = [obj] objs = [obj]
@ -142,7 +145,8 @@ def handle_setup_error(ex_type, ex_value, ex_tb):
exc_info=(ex_type, ex_value, ex_tb)) exc_info=(ex_type, ex_value, ex_tb))
def reset_fixture(obj, fixture_id=None, manager=None): def reset_fixture(obj, fixture_id=None, manager=None) \
-> fixtures.Fixture:
'''Get registered fixture and reset it''' '''Get registered fixture and reset it'''
fixture = get_fixture(obj, fixture_id=fixture_id, manager=manager) fixture = get_fixture(obj, fixture_id=fixture_id, manager=manager)
with _exception.handle_multiple_exceptions(): with _exception.handle_multiple_exceptions():
@ -150,7 +154,8 @@ def reset_fixture(obj, fixture_id=None, manager=None):
return fixture return fixture
def cleanup_fixture(obj, fixture_id=None, manager=None): def cleanup_fixture(obj, fixture_id=None, manager=None) \
-> fixtures.Fixture:
'''Get registered fixture and clean it up''' '''Get registered fixture and clean it up'''
fixture = get_fixture(obj, fixture_id=fixture_id, manager=manager) fixture = get_fixture(obj, fixture_id=fixture_id, manager=manager)
with _exception.handle_multiple_exceptions(): with _exception.handle_multiple_exceptions():
@ -158,7 +163,15 @@ def cleanup_fixture(obj, fixture_id=None, manager=None):
return fixture return fixture
def get_name_and_object(obj): def use_fixture(obj, fixture_id=None, manager=None) \
-> fixtures.Fixture:
fixture = setup_fixture(obj, fixture_id=fixture_id, manager=manager)
_testcase.add_cleanup(tobiko.cleanup_fixture, fixture)
return fixture
def get_name_and_object(obj: typing.Any) \
-> typing.Tuple[str, typing.Any]:
'''Get (name, obj) tuple identified by given :param obj:''' '''Get (name, obj) tuple identified by given :param obj:'''
if isinstance(obj, str): if isinstance(obj, str):
return obj, tobiko.load_object(obj) return obj, tobiko.load_object(obj)
@ -206,7 +219,7 @@ def list_required_fixtures(objects):
return result return result
def is_test_method(obj): def is_test_method(obj) -> bool:
'''Returns whenever given object is a test method''' '''Returns whenever given object is a test method'''
return ((inspect.isfunction(obj) or inspect.ismethod(obj)) and return ((inspect.isfunction(obj) or inspect.ismethod(obj)) and
obj.__name__.startswith('test_')) obj.__name__.startswith('test_'))
@ -313,38 +326,40 @@ def get_fixture_id(obj: typing.Any) -> typing.Any:
return getattr(obj, '__tobiko_fixture_id__', None) return getattr(obj, '__tobiko_fixture_id__', None)
def get_object_name(obj): def get_object_name(obj) -> str:
'''Gets a fully qualified name for given :param obj:''' '''Gets a fully qualified name for given :param obj:'''
if isinstance(obj, str): if isinstance(obj, str):
return obj return obj
name = getattr(obj, '__tobiko_fixture_name__', None) name = getattr(obj, '__tobiko_fixture_name__', None)
if name: if isinstance(name, str):
assert isinstance(name, str)
return name return name
assert name is None, f"{name} is not None"
if (not inspect.isfunction(obj) and if (not inspect.isfunction(obj) and
not inspect.ismethod(obj) and not inspect.ismethod(obj) and
not inspect.isclass(obj)): not inspect.isclass(obj)):
obj = type(obj) obj = type(obj)
module = inspect.getmodule(obj).__name__ module = inspect.getmodule(obj)
if module is not None:
name = getattr(obj, '__qualname__', None) name = getattr(obj, '__qualname__', None)
if name: if isinstance(name, str):
return module + '.' + name return module.__name__ + '.' + name
msg = "Unable to get fixture name from object {!r}".format(obj) raise TypeError(f"Unable to get fixture name from object {obj!r}")
raise TypeError(msg)
class FixtureManager(object): class FixtureManager(object):
def __init__(self): def __init__(self):
self.fixtures = {} self.fixtures: typing.Dict[str, fixtures.Fixture] = {}
def get_fixture(self, def get_fixture(self,
obj: typing.Any, obj: typing.Any,
fixture_id: typing.Any = None): fixture_id: typing.Any = None) \
-> fixtures.Fixture:
name, obj = get_name_and_object(obj) name, obj = get_name_and_object(obj)
if fixture_id: if fixture_id:
name += f'-{fixture_id}' name += f'-{fixture_id}'
@ -358,12 +373,14 @@ class FixtureManager(object):
@staticmethod @staticmethod
def init_fixture(obj: typing.Any, def init_fixture(obj: typing.Any,
name: str, name: str,
fixture_id: typing.Any): fixture_id: typing.Any) \
-> fixtures.Fixture:
return init_fixture(obj=obj, return init_fixture(obj=obj,
name=name, name=name,
fixture_id=fixture_id) fixture_id=fixture_id)
def remove_fixture(self, obj, fixture_id=None): def remove_fixture(self, obj, fixture_id=None) \
-> typing.Optional[fixtures.Fixture]:
name = get_object_name(obj) name = get_object_name(obj)
if fixture_id: if fixture_id:
name += '-' + str(fixture_id) name += '-' + str(fixture_id)

View File

@ -19,9 +19,6 @@ import os
import typing # noqa import typing # noqa
import time import time
# import testtools
import testtools
import tobiko import tobiko
from tobiko.shell import ping from tobiko.shell import ping
from tobiko.shell import sh from tobiko.shell import sh
@ -82,8 +79,7 @@ def test_servers_creation(stack=TestServerCreationStack,
# Create all servers stacks # Create all servers stacks
for fixture in fixtures: for fixture in fixtures:
assert isinstance(test_case, testtools.TestCase) tobiko.use_fixture(fixture)
test_case.useFixture(fixture)
# Check every server ID is unique and new # Check every server ID is unique and new
server_ids = {fixture.server_id for fixture in fixtures} server_ids = {fixture.server_id for fixture in fixtures}

View File

@ -15,6 +15,7 @@ from __future__ import absolute_import
import os import os
import sys import sys
import unittest
import fixtures import fixtures
import mock import mock
@ -180,6 +181,53 @@ class SetupFixtureTest(unit.TobikoUnitTest):
result.cleanup_fixture.assert_not_called() result.cleanup_fixture.assert_not_called()
class UseFixtureTest(unit.TobikoUnitTest):
def test_with_name(self):
self._test_use_fixture(canonical_name(MyFixture))
def test_with_type(self):
self._test_use_fixture(MyFixture)
def test_with_instance(self):
self._test_use_fixture(MyFixture2())
def test_with_name_and_fixture_id(self):
self._test_use_fixture(canonical_name(MyFixture), fixture_id=5)
def test_with_type_and_fixture_id(self):
self._test_use_fixture(MyFixture, fixture_id=6)
def test_with_instance_and_fixture_id(self):
self._test_use_fixture(MyFixture2(), fixture_id=7)
def _test_use_fixture(self, obj, fixture_id=None):
fixture: MyFixture = tobiko.get_fixture(
obj=obj, fixture_id=fixture_id) # type: ignore
class InnerTest(unittest.TestCase):
def runTest(self):
fixture.setup_fixture.assert_not_called()
fixture.cleanup_fixture.assert_not_called()
result = tobiko.use_fixture(obj, fixture_id=fixture_id)
fixture.setup_fixture.assert_called_once_with()
fixture.cleanup_fixture.assert_not_called()
self.assertIs(fixture, result)
fixture.setup_fixture.assert_not_called()
fixture.cleanup_fixture.assert_not_called()
result = tobiko.run_test(InnerTest())
fixture.setup_fixture.assert_called_once_with()
fixture.cleanup_fixture.assert_called_once_with()
self.assertEqual(1, result.testsRun)
self.assertEqual([], result.errors)
self.assertEqual([], result.failures)
class ResetFixtureTest(unit.TobikoUnitTest): class ResetFixtureTest(unit.TobikoUnitTest):
def test_with_name(self): def test_with_name(self):