x
/
tobiko
Code Issues Proposed changes

Refactor tripleo skippers for checking RHOSP version 

Change-Id: I9aea1b690f3903220bfc2428d5df92e0dbfc8748
This commit is contained in:
Federico Ressi 2022-07-18 08:33:22 +02:00
parent 45a949a430
commit b46dd592b4
8 changed files with 350 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tobiko/__init__.py
View File

@ -151,11 +151,13 @@ true_seconds = _time.true_seconds
get_short_hostname = _utils.get_short_hostname
InvalidVersion = _version.InvalidVersion
VersionMismatch = _version.VersionMismatch
VersionType = _version.VersionType
Version = _version.Version
check_version = _version.check_version
get_version = _version.get_version
match_version = _version.match_version
parse_version = _version.parse_version
get_version = _version.get_version
dump_yaml = _yaml.dump_yaml
load_yaml = _yaml.load_yaml

40
tobiko/common/_version.py
View File

@ -28,9 +28,13 @@ class InvalidVersion(_exception.TobikoException):
message = "invalid version: {text}"
class VersionMismatch(_exception.TobikoException):
message = "version mismatch {version} {cause}"
cause = ''
VERSION_CLASSES = _version.LegacyVersion, _version.Version
Version = typing.Union[_version.LegacyVersion,
_version.Version]
Version = typing.Union[_version.Version]
VersionType = typing.Union[Version, str]
@ -45,21 +49,43 @@ def get_version(obj: VersionType):
def parse_version(text: str) -> VersionType:
match = VERSION_PATTERN.search(text.strip())
if match is None:
raise InvalidVersion(text=text)
return _version.parse(match.group())
if match is not None:
text = match.group()
try:
return _version.Version(text)
except _version.InvalidVersion as ex:
raise InvalidVersion(text=text) from ex
def match_version(actual: VersionType,
min_version: VersionType = None,
max_version: VersionType = None) -> bool:
try:
check_version(actual=actual,
min_version=min_version,
max_version=max_version)
except VersionMismatch:
return False
else:
return True
def check_version(actual: VersionType,
min_version: VersionType = None,
max_version: VersionType = None,
mismatch_error=VersionMismatch):
actual = get_version(actual)
if min_version is not None:
min_version = get_version(min_version)
if actual < min_version:
return False
raise mismatch_error(
version=actual,
cause=f">= {max_version}")
if max_version is not None:
max_version = get_version(max_version)
if actual >= max_version:
return False
raise mismatch_error(
version=actual,
cause=f">= {max_version}")
return True

100
tobiko/tests/functional/tripleo/test_overcloud.py
View File

@ -149,9 +149,105 @@ class OvercloudProcessesTest(testtools.TestCase):
self.assertTrue(ops.basic_overcloud_processes_running)
@tripleo.skip_if_missing_undercloud
class OvercloudVersionTest(unittest.TestCase):
@tripleo.skip_if_missing_undercloud
@property
def lower_version(self) -> str:
v = tripleo.overcloud_version()
return f"{v.major}.{v.minor}.{v.micro - 1}"
@property
def same_version(self) -> str:
v = tripleo.overcloud_version()
return f"{v.major}.{v.minor}.{v.micro}"
@property
def higher_version(self) -> str:
v = tripleo.overcloud_version()
return f"{v.major}.{v.minor}.{v.micro + 1}"
def test_overcloud_version(self):
version = tripleo.overcloud_version()
self.assertTrue(tobiko.match_version(version, min_version='13.0.0'))
self.assertTrue(tobiko.match_version(version, min_version='13'))
def test_has_overcloud(self):
self.assertTrue(tripleo.has_overcloud())
def test_has_overcloud_with_min_version(self):
self.assertTrue(
tripleo.has_overcloud(min_version=self.same_version))
def test_has_overcloud_with_min_version_lower(self):
self.assertTrue(
tripleo.has_overcloud(min_version=self.lower_version))
def test_has_overcloud_with_min_version_higher(self):
self.assertFalse(
tripleo.has_overcloud(min_version=self.higher_version))
def test_has_overcloud_with_max_version(self):
self.assertFalse(
tripleo.has_overcloud(max_version=self.same_version))
def test_has_overcloud_with_max_version_lower(self):
self.assertFalse(
tripleo.has_overcloud(max_version=self.lower_version))
def test_has_overcloud_with_max_version_higher(self):
self.assertTrue(
tripleo.has_overcloud(max_version=self.higher_version))
def test_skip_unless_has_overcloud(self):
self._assert_test_skip_unless_has_overcloud_dont_skip()
def test_skip_unless_has_overcloud_with_min_version(self):
self._assert_test_skip_unless_has_overcloud_dont_skip(
min_version=self.same_version)
def test_skip_unless_has_overcloud_with_min_version_lower(self):
self._assert_test_skip_unless_has_overcloud_dont_skip(
min_version=self.lower_version)
def test_skip_unless_has_overcloud_with_min_version_higher(self):
self._assert_test_skip_unless_has_overcloud_skip(
min_version=self.higher_version)
def test_skip_unless_has_overcloud_with_max_version(self):
self._assert_test_skip_unless_has_overcloud_skip(
max_version=self.same_version)
def test_skip_unless_has_overcloud_with_max_version_lower(self):
self._assert_test_skip_unless_has_overcloud_skip(
max_version=self.lower_version)
def test_skip_unless_has_overcloud_with_max_version_higher(self):
self._assert_test_skip_unless_has_overcloud_dont_skip(
max_version=self.higher_version)
def _assert_test_skip_unless_has_overcloud_dont_skip(
self,
min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None):
executed = []
@tripleo.skip_unless_has_overcloud(min_version=min_version,
max_version=max_version)
def decorated_function():
executed.append(True)
self.assertFalse(executed)
decorated_function()
self.assertTrue(executed)
def _assert_test_skip_unless_has_overcloud_skip(
self,
min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None):
@tripleo.skip_unless_has_overcloud(min_version=min_version,
max_version=max_version)
def decorated_function():
raise self.fail('Not skipped')
with self.assertRaises(unittest.SkipTest):
decorated_function()

99
tobiko/tests/functional/tripleo/test_undercloud.py
View File

@ -74,9 +74,106 @@ class UndercloudKeystoneClientTest(testtools.TestCase):
self.assertTrue(services)
@tripleo.skip_if_missing_undercloud
class UndercloudVersionTest(unittest.TestCase):
@property
def lower_version(self) -> str:
v = tripleo.undercloud_version()
return f"{v.major}.{v.minor}.{v.micro - 1}"
@property
def same_version(self) -> str:
v = tripleo.undercloud_version()
return f"{v.major}.{v.minor}.{v.micro}"
@property
def higher_version(self) -> str:
v = tripleo.undercloud_version()
return f"{v.major}.{v.minor}.{v.micro + 1}"
@tripleo.skip_if_missing_undercloud
def test_undercloud_version(self):
version = tripleo.undercloud_version()
self.assertTrue(tobiko.match_version(version, min_version='13.0.0'))
self.assertTrue(tobiko.match_version(version, min_version='13'))
def test_has_undercloud(self):
self.assertTrue(tripleo.has_undercloud())
def test_has_undercloud_with_min_version(self):
self.assertTrue(
tripleo.has_undercloud(min_version=self.same_version))
def test_has_undercloud_with_min_version_lower(self):
self.assertTrue(
tripleo.has_undercloud(min_version=self.lower_version))
def test_has_undercloud_with_min_version_higher(self):
self.assertFalse(
tripleo.has_undercloud(min_version=self.higher_version))
def test_has_undercloud_with_max_version(self):
self.assertFalse(
tripleo.has_undercloud(max_version=self.same_version))
def test_has_undercloud_with_max_version_lower(self):
self.assertFalse(
tripleo.has_undercloud(max_version=self.lower_version))
def test_has_undercloud_with_max_version_higher(self):
self.assertTrue(
tripleo.has_undercloud(max_version=self.higher_version))
def test_skip_unless_has_undercloud(self):
self._assert_test_skip_unless_has_undercloud_dont_skip()
def test_skip_unless_has_undercloud_with_min_version(self):
self._assert_test_skip_unless_has_undercloud_dont_skip(
min_version=self.same_version)
def test_skip_unless_has_undercloud_with_min_version_lower(self):
self._assert_test_skip_unless_has_undercloud_dont_skip(
min_version=self.lower_version)
def test_skip_unless_has_undercloud_with_min_version_higher(self):
self._assert_test_skip_unless_has_undercloud_skip(
min_version=self.higher_version)
def test_skip_unless_has_undercloud_with_max_version(self):
self._assert_test_skip_unless_has_undercloud_skip(
max_version=self.same_version)
def test_skip_unless_has_undercloud_with_max_version_lower(self):
self._assert_test_skip_unless_has_undercloud_skip(
max_version=self.lower_version)
def test_skip_unless_has_undercloud_with_max_version_higher(self):
self._assert_test_skip_unless_has_undercloud_dont_skip(
max_version=self.higher_version)
def _assert_test_skip_unless_has_undercloud_dont_skip(
self,
min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None):
executed = []
@tripleo.skip_unless_has_undercloud(min_version=min_version,
max_version=max_version)
def decorated_function():
executed.append(True)
self.assertFalse(executed)
decorated_function()
self.assertTrue(executed)
def _assert_test_skip_unless_has_undercloud_skip(
self,
min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None):
@tripleo.skip_unless_has_undercloud(min_version=min_version,
max_version=max_version)
def decorated_function():
raise self.fail('Not skipped')
with self.assertRaises(unittest.SkipTest):
decorated_function()

10
tobiko/tests/unit/test_version.py
View File

@ -23,6 +23,10 @@ import tobiko
class TestVersion(unittest.TestCase):
def test_parse_version(self):
self.assertEqual(version.parse('1.0.0'),
tobiko.parse_version('1'))
self.assertEqual(version.parse('1.2.0'),
tobiko.parse_version('1.2'))
self.assertEqual(version.parse('1.2.3'),
tobiko.parse_version('1.2.3'))
self.assertEqual(version.parse('3.2.1'),
@ -30,7 +34,7 @@ class TestVersion(unittest.TestCase):
def test_parse_version_with_invalid(self):
with self.assertRaises(tobiko.InvalidVersion):
tobiko.parse_version('1.2')
tobiko.parse_version('abc')
def test_get_version_with_srt(self):
self.assertEqual(version.parse('3.2.1'),
@ -51,6 +55,10 @@ class TestVersion(unittest.TestCase):
def test_match_version_min_version(self):
self.assertTrue(tobiko.match_version('1.0.0',
min_version='0.9.9'))
self.assertTrue(tobiko.match_version('1.0.0',
min_version='1'))
self.assertTrue(tobiko.match_version('1.0.0',
min_version='1.0'))
self.assertTrue(tobiko.match_version('1.0.0',
min_version='1.0.0'))
self.assertFalse(tobiko.match_version('1.0.0',

2
tobiko/tripleo/__init__.py
View File

@ -43,6 +43,7 @@ overcloud_node_ip_address = overcloud.overcloud_node_ip_address
overcloud_ssh_client = overcloud.overcloud_ssh_client
overcloud_version = overcloud.overcloud_version
skip_if_missing_overcloud = overcloud.skip_if_missing_overcloud
skip_unless_has_overcloud = overcloud.skip_unless_has_overcloud
get_rhosp_release = _rhosp.get_rhosp_release
get_rhosp_version = _rhosp.get_rhosp_version
@ -51,6 +52,7 @@ TripleoTopology = topology.TripleoTopology
load_undercloud_rcfile = undercloud.load_undercloud_rcfile
has_undercloud = undercloud.has_undercloud
skip_unless_has_undercloud = undercloud.skip_unlsess_has_undercloud
skip_if_missing_undercloud = undercloud.skip_if_missing_undercloud
undercloud_host_config = undercloud.undercloud_host_config
undercloud_keystone_client = undercloud.undercloud_keystone_client

75
tobiko/tripleo/_overcloud.py
View File

@ -13,6 +13,7 @@
# under the License.
from __future__ import absolute_import
import functools
import io
import os
import typing
@ -34,34 +35,38 @@ CONF = config.CONF
LOG = log.getLogger(__name__)
def has_overcloud(min_version: str = None,
max_version: str = None) -> bool:
if not _undercloud.has_undercloud():
return False
if min_version or max_version:
if not tobiko.match_version(overcloud_version(),
min_version=min_version,
max_version=max_version):
return False
return True
def overcloud_version() -> tobiko.Version:
from tobiko.tripleo import _topology
node = topology.find_openstack_node(group='overcloud')
assert isinstance(node, _topology.TripleoTopologyNode)
return node.rhosp_version
def load_overcloud_rcfile() -> typing.Dict[str, str]:
return _undercloud.fetch_os_env(*CONF.tobiko.tripleo.overcloud_rcfile)
@functools.lru_cache()
def has_overcloud(min_version: str = None,
max_version: str = None) -> bool:
try:
check_overcloud(min_version=min_version,
max_version=max_version)
except (OvercloudNotFound, OvercloudVersionMismatch) as ex:
LOG.debug(f'Overcloud not found: {ex}')
return False
else:
LOG.debug('Overcloud found')
return True
skip_if_missing_overcloud = tobiko.skip_unless(
'TripleO overcloud not configured', has_overcloud)
def skip_unless_has_overcloud(min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None):
return tobiko.skip_on_error(
'TripleO overcloud not configured',
check_overcloud,
min_version=min_version,
max_version=max_version,
error_type=(OvercloudNotFound, OvercloudVersionMismatch))
class OvercloudKeystoneCredentialsFixture(
keystone.EnvironKeystoneCredentialsFixture):
@ -269,3 +274,33 @@ def is_redis_expected():
if keystone.has_service(name=service):
return True
return False
def overcloud_version() -> tobiko.Version:
from tobiko.tripleo import _topology
node = topology.find_openstack_node(group='overcloud')
assert isinstance(node, _topology.TripleoTopologyNode)
return node.rhosp_version
@functools.lru_cache()
def check_overcloud(min_version: str = None,
max_version: str = None):
try:
_undercloud.check_undercloud()
except _undercloud.UndercloudNotFound as ex:
raise OvercloudNotFound(cause=f'undercloud not found: {ex}') from ex
if min_version or max_version:
tobiko.check_version(overcloud_version(),
min_version=min_version,
max_version=max_version,
mismatch_error=OvercloudVersionMismatch)
class OvercloudNotFound(tobiko.ObjectNotFound):
message = 'overcloud not found: {cause}'
class OvercloudVersionMismatch(tobiko.VersionMismatch):
message = 'overcloud version mismatch: {version} {cause}'

79
tobiko/tripleo/_undercloud.py
View File

@ -103,41 +103,34 @@ class CloudsFileUndercloudKeystoneCredentialsFixture(
clouds_files=clouds_files)
class HasUndercloudFixture(tobiko.SharedFixture):
has_undercloud: typing.Optional[bool] = None
def setup_fixture(self):
self.has_undercloud = check_undercloud()
def check_undercloud() -> bool:
@functools.lru_cache()
def has_undercloud(min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None) -> bool:
try:
ssh_client = undercloud_ssh_client()
except NoSuchUndercloudHostname:
LOG.debug('TripleO undercloud hostname not found')
check_undercloud(min_version=min_version,
max_version=max_version)
except (UndercloudNotFound, UndercloudVersionMismatch) as ex:
LOG.debug(f'TripleO undercloud host not found: {ex.cause}')
return False
try:
ssh_client.connect(retry_count=1,
connection_attempts=1,
timeout=15.)
except Exception as ex:
LOG.debug(f'Unable to connect to TripleO undercloud host: {ex}',
exc_info=1)
return False
LOG.debug('TripleO undercloud host found')
return True
def has_undercloud() -> bool:
return tobiko.setup_fixture(HasUndercloudFixture).has_undercloud
else:
LOG.debug('TripleO undercloud host found')
return True
skip_if_missing_undercloud = tobiko.skip_unless(
'TripleO undercloud hostname not configured', has_undercloud)
def skip_unlsess_has_undercloud(min_version: tobiko.VersionType = None,
max_version: tobiko.VersionType = None):
return tobiko.skip_on_error(
reason='TripleO undercloud not found',
predicate=check_undercloud,
min_version=min_version,
max_version=max_version,
error_type=(UndercloudNotFound, UndercloudVersionMismatch))
class UndecloudHostConfig(tobiko.SharedFixture):
hostname: typing.Optional[str] = None
@ -206,3 +199,35 @@ def undercloud_keystone_credentials():
def undercloud_version() -> tobiko.Version:
ssh_client = undercloud_ssh_client()
return _rhosp.get_rhosp_version(connection=ssh_client)
@functools.lru_cache()
def check_undercloud(min_version: tobiko.Version = None,
max_version: tobiko.Version = None):
try:
ssh_client = undercloud_ssh_client()
except NoSuchUndercloudHostname as ex:
raise UndercloudNotFound(
cause='TripleO undercloud hostname not found') from ex
try:
ssh_client.connect(retry_count=1,
connection_attempts=1,
timeout=15.)
except Exception as ex:
raise UndercloudNotFound(
cause=f'unable to connect to TripleO undercloud host: {ex}'
) from ex
if min_version or max_version:
tobiko.check_version(undercloud_version(),
min_version=min_version,
max_version=max_version,
mismatch_error=UndercloudVersionMismatch)
class UndercloudNotFound(tobiko.ObjectNotFound):
message = 'undercloud not found: {cause}'
class UndercloudVersionMismatch(tobiko.VersionMismatch):
message = 'undercloud version mismatch: {version} {cause}'