diff --git a/tobiko/__init__.py b/tobiko/__init__.py index 794011832..a30af7439 100644 --- a/tobiko/__init__.py +++ b/tobiko/__init__.py @@ -109,10 +109,12 @@ skip_if = _skip.skip_if skip_unless = _skip.skip_unless BaseTestCase = _testcase.TestCase +assert_test_case_was_skipped = _testcase.assert_test_case_was_skipped discover_test_cases = _testcase.discover_test_cases get_test_case = _testcase.get_test_case pop_test_case = _testcase.pop_test_case push_test_case = _testcase.push_test_case +run_test = _testcase.run_test TestCasesManager = _testcase.TestCasesManager min_seconds = _time.min_seconds diff --git a/tobiko/common/_testcase.py b/tobiko/common/_testcase.py index 7d5b6f510..341ee8f51 100644 --- a/tobiko/common/_testcase.py +++ b/tobiko/common/_testcase.py @@ -237,3 +237,36 @@ class DummyTestCase(TestCase): DUMMY_TEST_CASE = DummyTestCase() + + +def run_test(test_case: testtools.TestCase, + test_result: testtools.TestResult = None) -> testtools.TestResult: + test_result = test_result or testtools.TestResult() + test_case.run(test_result) + return test_result + + +def assert_in(needle, haystack, message: typing.Optional[str] = None, + manager: TestCasesManager = TEST_CASES): + get_test_case(manager=manager).assertIn(needle, haystack, message) + + +def get_skipped_test_cases(test_result: testtools.TestResult, + skip_reason: typing.Optional[str] = None): + if skip_reason is not None: + assert_in(skip_reason, test_result.skip_reasons) + return test_result.skip_reasons[skip_reason] + else: + skipped_test_cases = list() + for cases in test_result.skip_reasons.values(): + skipped_test_cases.extend(cases) + return skipped_test_cases + + +def assert_test_case_was_skipped(test_case: testtools.TestCase, + test_result: testtools.TestResult, + skip_reason: str = None, + manager: TestCasesManager = TEST_CASES): + skipped_tests = get_skipped_test_cases(test_result=test_result, + skip_reason=skip_reason) + assert_in(test_case, skipped_tests, manager=manager) diff --git a/tobiko/openstack/keystone/__init__.py b/tobiko/openstack/keystone/__init__.py index b93c04e3b..402a959af 100644 --- a/tobiko/openstack/keystone/__init__.py +++ b/tobiko/openstack/keystone/__init__.py @@ -31,9 +31,14 @@ KeystoneClientFixture = _client.KeystoneClientFixture CloudsFileKeystoneCredentialsFixture = ( _clouds_file.CloudsFileKeystoneCredentialsFixture) -keystone_credentials = _credentials.keystone_credentials -get_keystone_credentials = _credentials.get_keystone_credentials default_keystone_credentials = _credentials.default_keystone_credentials +get_keystone_credentials = _credentials.get_keystone_credentials +has_keystone_credentials = _credentials.has_keystone_credentials +keystone_credentials = _credentials.keystone_credentials +skip_unless_has_keystone_credentials = ( + _credentials.skip_unless_has_keystone_credentials) +DefaultKeystoneCredentialsFixture = ( + _credentials.DefaultKeystoneCredentialsFixture) KeystoneCredentials = _credentials.KeystoneCredentials KeystoneCredentialsFixture = _credentials.KeystoneCredentialsFixture EnvironKeystoneCredentialsFixture = \ diff --git a/tobiko/openstack/keystone/_credentials.py b/tobiko/openstack/keystone/_credentials.py index 048b53bb2..91f14c771 100644 --- a/tobiko/openstack/keystone/_credentials.py +++ b/tobiko/openstack/keystone/_credentials.py @@ -16,6 +16,7 @@ from __future__ import absolute_import import collections import os import sys +import typing # noqa from oslo_log import log import testtools @@ -27,28 +28,6 @@ import tobiko LOG = log.getLogger(__name__) -def get_keystone_credentials(obj=None): - if not obj: - return default_keystone_credentials() - if tobiko.is_fixture(obj): - obj = tobiko.get_fixture(obj) - if isinstance(obj, KeystoneCredentialsFixture): - obj = tobiko.setup_fixture(obj).credentials - if isinstance(obj, KeystoneCredentials): - return obj - - message = "Can't get {!r} object from {!r}".format( - KeystoneCredentials, obj) - raise TypeError(message) - - -def default_keystone_credentials(): - credentials = tobiko.setup_fixture(DefaultKeystoneCredentialsFixture - ).credentials - tobiko.check_valid_type(credentials, KeystoneCredentials) - return credentials - - class KeystoneCredentials(collections.namedtuple( 'KeystoneCredentials', ['api_version', 'auth_url', @@ -87,6 +66,33 @@ class KeystoneCredentials(collections.namedtuple( raise InvalidKeystoneCredentials(credentials=self, reason=reason) +class NoSuchCredentialsError(tobiko.TobikoException): + message = "No such credentials from any of: {fixtures}" + + +def get_keystone_credentials(obj=None) -> KeystoneCredentials: + if not obj: + return default_keystone_credentials() + if tobiko.is_fixture(obj): + obj = tobiko.get_fixture(obj) + if isinstance(obj, KeystoneCredentialsFixture): + obj = tobiko.setup_fixture(obj).credentials + if isinstance(obj, KeystoneCredentials): + return obj + + message = "Can't get {!r} object from {!r}".format( + KeystoneCredentials, obj) + raise TypeError(message) + + +def default_keystone_credentials() -> KeystoneCredentials: + credentials = tobiko.setup_fixture( + DefaultKeystoneCredentialsFixture).credentials + if credentials: + tobiko.check_valid_type(credentials, KeystoneCredentials) + return credentials + + def keystone_credentials(api_version=None, auth_url=None, username=None, @@ -97,7 +103,7 @@ def keystone_credentials(api_version=None, project_domain_name=None, project_domain_id=None, trust_id=None, - cls=KeystoneCredentials): + cls=KeystoneCredentials) -> KeystoneCredentials: return cls(api_version=api_version, auth_url=auth_url, username=username, @@ -116,9 +122,10 @@ class InvalidKeystoneCredentials(tobiko.TobikoException): class KeystoneCredentialsFixture(tobiko.SharedFixture): - credentials = None + credentials: typing.Optional[KeystoneCredentials] = None - def __init__(self, credentials=None): + def __init__(self, + credentials: typing.Optional[KeystoneCredentials] = None): super(KeystoneCredentialsFixture, self).__init__() if credentials: self.credentials = credentials @@ -143,15 +150,17 @@ class KeystoneCredentialsFixture(tobiko.SharedFixture): def cleanup_credentials(self): del self.credentials - def get_credentials(self): + def get_credentials(self) -> typing.Optional[KeystoneCredentials]: return self.credentials class EnvironKeystoneCredentialsFixture(KeystoneCredentialsFixture): - environ = None + environ: typing.Optional[typing.Dict[str, str]] = None - def __init__(self, credentials=None, environ=None): + def __init__(self, + credentials: typing.Optional[KeystoneCredentials] = None, + environ: typing.Optional[typing.Dict[str, str]] = None): super(EnvironKeystoneCredentialsFixture, self).__init__( credentials=credentials) if environ is not None: @@ -162,10 +171,10 @@ class EnvironKeystoneCredentialsFixture(KeystoneCredentialsFixture): self.environ = self.get_environ() super(EnvironKeystoneCredentialsFixture, self).setup_fixture() - def get_environ(self): - return os.environ + def get_environ(self) -> typing.Optional[typing.Dict[str, str]]: + return dict(os.environ) - def get_credentials(self): + def get_credentials(self) -> typing.Optional[KeystoneCredentials]: auth_url = self.get_env('OS_AUTH_URL') if not auth_url: LOG.debug("OS_AUTH_URL environment variable not defined") @@ -214,19 +223,38 @@ class EnvironKeystoneCredentialsFixture(KeystoneCredentialsFixture): project_domain_id=project_domain_id, trust_id=trust_id) - def get_env(self, name): - return self.environ.get(name, None) + def get_env(self, name) -> typing.Optional[str]: + environ = self.environ + if environ is None: + return None + else: + return environ.get(name) - def get_int_env(self, name): + def get_int_env(self, name) -> typing.Optional[int]: value = self.get_env(name=name) - if value is not None: - value = int(value) - return value + if value is None: + return None + else: + return int(value) + + +def has_keystone_credentials(obj=None) -> bool: + try: + credentials = get_keystone_credentials(obj) + except NoSuchCredentialsError: + return False + else: + return credentials is not None + + +def skip_unless_has_keystone_credentials(*args, **kwargs): + return tobiko.skip_unless('Missing Keystone credentials', + has_keystone_credentials, *args, **kwargs) class ConfigKeystoneCredentialsFixture(KeystoneCredentialsFixture): - def get_credentials(self): + def get_credentials(self) -> typing.Optional[KeystoneCredentials]: from tobiko import config conf = config.CONF.tobiko.keystone auth_url = conf.auth_url @@ -267,7 +295,7 @@ class DefaultKeystoneCredentialsFixture(KeystoneCredentialsFixture): fixtures = DEFAULT_KEYSTONE_CREDENTIALS_FIXTURES - def get_credentials(self): + def get_credentials(self) -> typing.Optional[KeystoneCredentials]: errors = [] for fixture in self.fixtures: try: @@ -290,12 +318,10 @@ class DefaultKeystoneCredentialsFixture(KeystoneCredentialsFixture): elif errors: raise testtools.MultipleExceptions(errors) - raise ValueError("No such credentials from any of: \n " + - '\n '.join(tobiko.get_fixture_name(fixture) - for fixture in self.fixtures)) + raise NoSuchCredentialsError(fixtures=self.fixtures) -def api_version_from_url(auth_url): +def api_version_from_url(auth_url) -> typing.Optional[int]: if auth_url.endswith('/v2.0'): LOG.debug('Got Keystone API version 2 from auth_url: %r', auth_url) return 2 diff --git a/tobiko/tests/functional/shell/test_execute.py b/tobiko/tests/functional/shell/test_execute.py index 3e8e1f52c..7faf1fbe6 100644 --- a/tobiko/tests/functional/shell/test_execute.py +++ b/tobiko/tests/functional/shell/test_execute.py @@ -19,6 +19,7 @@ import testtools import tobiko from tobiko import config +from tobiko.openstack import keystone from tobiko.openstack import stacks from tobiko.shell import sh @@ -170,6 +171,7 @@ class LocalExecuteTest(ExecuteTest): return sh.local_execute(**kwargs) +@keystone.skip_unless_has_keystone_credentials() class SSHExecuteTest(ExecuteTest): server_stack = tobiko.required_setup_fixture( diff --git a/tobiko/tests/unit/openstack/keystone/test_credentials.py b/tobiko/tests/unit/openstack/keystone/test_credentials.py index cc0b2b486..c5e246ea3 100644 --- a/tobiko/tests/unit/openstack/keystone/test_credentials.py +++ b/tobiko/tests/unit/openstack/keystone/test_credentials.py @@ -16,6 +16,8 @@ from __future__ import absolute_import import os +import testtools + import tobiko from tobiko import config from tobiko.openstack import keystone @@ -128,6 +130,11 @@ class EnvironKeystoneCredentialsFixtureTest(openstack.OpenstackTest): fixture = _credentials.EnvironKeystoneCredentialsFixture() self.assertIsNone(fixture.credentials) + def test_setup_with_no_credentials(self): + fixture = _credentials.EnvironKeystoneCredentialsFixture() + fixture.setUp() + self.assertIsNone(fixture.credentials) + def test_setup_v2(self): self.patch(os, 'environ', V2_ENVIRON) fixture = _credentials.EnvironKeystoneCredentialsFixture() @@ -217,19 +224,19 @@ class DefaultKeystoneCredentialsFixtureTest(openstack.OpenstackTest): return self.patch(config.CONF.tobiko, 'keystone', credentials) def test_init(self): - fixture = _credentials.DefaultKeystoneCredentialsFixture() + fixture = keystone.DefaultKeystoneCredentialsFixture() self.assertIsNone(fixture.credentials) def test_setup_from_environ(self): self.patch(os, 'environ', V2_ENVIRON) - fixture = _credentials.DefaultKeystoneCredentialsFixture() + fixture = keystone.DefaultKeystoneCredentialsFixture() fixture.setUp() fixture.credentials.validate() self.assertEqual(V2_PARAMS, fixture.credentials.to_dict()) def test_setup_from_config(self): self.patch_config(V2_PARAMS) - fixture = _credentials.DefaultKeystoneCredentialsFixture() + fixture = keystone.DefaultKeystoneCredentialsFixture() fixture.setUp() fixture.credentials.validate() self.assertEqual(V2_PARAMS, fixture.credentials.to_dict()) @@ -237,7 +244,72 @@ class DefaultKeystoneCredentialsFixtureTest(openstack.OpenstackTest): def test_setup_from_environ_and_confif(self): self.patch(os, 'environ', V3_ENVIRON) self.patch_config(V2_PARAMS) - fixture = _credentials.DefaultKeystoneCredentialsFixture() + fixture = keystone.DefaultKeystoneCredentialsFixture() fixture.setUp() fixture.credentials.validate() self.assertEqual(V3_PARAMS, fixture.credentials.to_dict()) + + +class SkipUnlessHasKeystoneCredentialsTest(openstack.OpenstackTest): + + def setUp(self): + super(SkipUnlessHasKeystoneCredentialsTest, self).setUp() + self.default_credentials = tobiko.setup_fixture( + keystone.DefaultKeystoneCredentialsFixture) + + def test_skip_method_unless_has_keystone_credentials_without_creds(self): + self.patch(self.default_credentials, 'credentials', None) + + @keystone.skip_unless_has_keystone_credentials() + def decorated_func(): + self.fail('Not skipped') + + self.assertFalse(keystone.has_keystone_credentials()) + self.assertRaises(self.skipException, decorated_func) + + def test_skip_class_unless_has_keystone_credentials_without_creds(self): + self.patch(self.default_credentials, 'credentials', None) + + @keystone.skip_unless_has_keystone_credentials() + class SkipTest(testtools.TestCase): + + def test_skip(self): + super(SkipTest, self).setUp() + self.fail('Not skipped') + + self.assertFalse(keystone.has_keystone_credentials()) + + test_case = SkipTest('test_skip') + test_result = tobiko.run_test(test_case) + tobiko.assert_test_case_was_skipped( + test_case, test_result, skip_reason='Missing Keystone credentials') + + def test_skip_method_unless_has_keystone_credentials_with_creds(self): + credentials = make_credentials({}) + self.patch(self.default_credentials, 'credentials', credentials) + + call_args = [] + + @keystone.skip_unless_has_keystone_credentials() + def decorated_func(*args, **kwargs): + call_args.append([args, kwargs]) + + self.assertEqual(credentials, keystone.get_keystone_credentials()) + decorated_func(1, 2, a=1, b=2) + self.assertEqual(call_args, [[(1, 2), {'a': 1, 'b': 2}]]) + + def test_skip_class_unless_has_keystone_credentials_with_creds(self): + credentials = make_credentials({}) + self.patch(self.default_credentials, 'credentials', credentials) + + calls = [] + + @keystone.skip_unless_has_keystone_credentials() + class SkipTest(testtools.TestCase): + + def test_skip(self): + calls.append(True) + + self.assertEqual(credentials, keystone.get_keystone_credentials()) + tobiko.run_test(SkipTest('test_skip')) + self.assertEqual(calls, [True])