Add skip method for missing Keystone credentials

Change-Id: Ic03eb95fa44429d41c3c48766af4bf5d901b7c1d

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

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)

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 = \

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):
+    def get_environ(self) -> typing.Optional[typing.Dict[str, str]]:
-        return os.environ
+        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):
+    def get_env(self, name) -> typing.Optional[str]:
-        return self.environ.get(name, None)
+        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:
+        if value is None:
-            value = int(value)
+            return None
-        return value
+        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    " +
+        raise NoSuchCredentialsError(fixtures=self.fixtures)
-                         '\n    '.join(tobiko.get_fixture_name(fixture)
-                                       for fixture in 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

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(

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])