Merge "Improve required_fixture typing"

This commit is contained in:
Zuul 2021-12-14 13:59:41 +00:00 committed by Gerrit Code Review
commit 75c12cc995
6 changed files with 27 additions and 17 deletions

View File

@ -288,16 +288,19 @@ def fixture_property(*args, **kwargs):
return FixtureProperty(*args, **kwargs)
def required_fixture(obj, **params):
'''Creates a property that gets fixture identified by given :param obj:
F = typing.TypeVar('F', bound='SharedFixture')
'''
return RequiredFixtureProperty(obj, **params)
def required_fixture(cls: typing.Type[F], **params) \
-> 'RequiredFixtureProperty[F]':
"""Creates a property that gets fixture identified by given :param cls:
"""
return RequiredFixtureProperty[F](cls, **params)
@_deprecation.deprecated(
deprecated_in='0.4.7',
removed_in='0.4.8',
removed_in='0.4.12',
details='use tobiko.required_fixture function instead')
def required_setup_fixture(obj, **params):
'''Creates a property that sets up fixture identified by given :param obj:
@ -460,20 +463,29 @@ class FixtureProperty(property):
return super(FixtureProperty, self).__get__(instance, owner)
class RequiredFixtureProperty(object):
class RequiredFixtureProperty(typing.Generic[F]):
def __init__(self, fixture, setup=True, **params):
def __init__(self, fixture: typing.Any, setup=True, **params):
self.fixture = fixture
self.fixture_params = params
self.setup = setup
@typing.overload
def __get__(self, instance: None, owner: typing.Type[F]) \
-> 'RequiredFixtureProperty[F]':
pass
@typing.overload
def __get__(self, instance: F, owner: typing.Type[F]) -> F:
pass
def __get__(self, instance, _):
if instance is None:
return self
else:
return self.get_fixture(instance)
def get_fixture(self, _instance):
def get_fixture(self, _instance) -> F:
fixture = get_fixture(self.fixture, **self.fixture_params)
if self.setup:
setup_fixture(fixture)

View File

@ -201,7 +201,8 @@ class ServerStackFixture(heat.HeatStackFixture, abc.ABC):
fixed_ips.with_attributes(version=ip_version)
return fixed_ips
def find_fixed_ip(self, ip_version: typing.Optional[int] = None,
def find_fixed_ip(self,
ip_version: int = None,
unique=False) -> netaddr.IPAddress:
fixed_ips = self.list_fixed_ips(ip_version=ip_version)
if unique:

View File

@ -24,8 +24,7 @@ from tobiko.run import _config
class WorkersPoolFixture(tobiko.SharedFixture):
config: _config.RunConfigFixture = tobiko.required_fixture(
_config.RunConfigFixture)
config = tobiko.required_fixture(_config.RunConfigFixture)
pool: multiprocessing.pool.Pool
workers_count: int = 0

View File

@ -31,8 +31,7 @@ from tobiko.tripleo import undercloud
@neutron.skip_if_missing_networking_agents(neutron.L3_AGENT)
class OpenvswitchTest(testtools.TestCase):
stack: stacks.ServerStackFixture = tobiko.required_fixture(
stacks.CirrosServerStackFixture)
stack = tobiko.required_fixture(stacks.CirrosServerStackFixture)
def setUp(self):
super(OpenvswitchTest, self).setUp()

View File

@ -111,8 +111,8 @@ class OvercloudServicesTest(testtools.TestCase):
across the overcloud nodes
"""
services_status: services.OvercloudServicesStatus = (
tobiko.required_fixture(services.OvercloudServicesStatus))
services_status = tobiko.required_fixture(
services.OvercloudServicesStatus)
def test_get_services_resource_table(self):
self.assertIsInstance(self.services_status.oc_services_df,

View File

@ -46,8 +46,7 @@ class RebootTrunkServerStackFixture(stacks.UbuntuServerStackFixture):
class RebootTrunkTest(testtools.TestCase):
"""Tests trunk functionality"""
stack: RebootTrunkServerStackFixture = tobiko.required_fixture(
RebootTrunkServerStackFixture)
stack = tobiko.required_fixture(RebootTrunkServerStackFixture)
def test_0_vlan_ip_addresses(self):
"""Check Nova server VLAN port IP addresses"""