import mock
from oslo.config import cfg
from oslo.messaging import conffixture as messaging_conffixture
+from oslo.utils import strutils
import testtools
from neutron.common import config
CONF = cfg.CONF
CONF.import_opt('state_path', 'neutron.common.config')
-TRUE_STRING = ['True', '1']
LOG_FORMAT = "%(asctime)s %(levelname)8s [%(name)s] %(message)s"
ROOTDIR = os.path.dirname(__file__)
return []
+def bool_from_env(key, strict=False, default=False):
+ value = os.environ.get(key)
+ return strutils.bool_from_string(value, strict=strict, default=default)
+
+
class BaseTestCase(testtools.TestCase):
@staticmethod
self.addOnException(post_mortem_debug.get_exception_handler(
debugger))
- if os.environ.get('OS_DEBUG') in TRUE_STRING:
+ if bool_from_env('OS_DEBUG'):
_level = std_logging.DEBUG
else:
_level = std_logging.INFO
- capture_logs = os.environ.get('OS_LOG_CAPTURE') in TRUE_STRING
+ capture_logs = bool_from_env('OS_LOG_CAPTURE')
if not capture_logs:
std_logging.basicConfig(format=LOG_FORMAT, level=_level)
self.log_fixture = self.useFixture(
self.addCleanup(mock.patch.stopall)
self.addCleanup(CONF.reset)
- if os.environ.get('OS_STDOUT_CAPTURE') in TRUE_STRING:
+ if bool_from_env('OS_STDOUT_CAPTURE'):
stdout = self.useFixture(fixtures.StringStream('stdout')).stream
self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
- if os.environ.get('OS_STDERR_CAPTURE') in TRUE_STRING:
+ if bool_from_env('OS_STDERR_CAPTURE'):
stderr = self.useFixture(fixtures.StringStream('stderr')).stream
self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
self.useFixture(fixtures.MonkeyPatch(
def setUp(self):
super(BaseSudoTestCase, self).setUp()
- env = os.environ
- self.sudo_enabled = env.get('OS_SUDO_TESTING') in base.TRUE_STRING
- self.root_helper = env.get('OS_ROOTWRAP_CMD', SUDO_CMD)
+ self.sudo_enabled = base.bool_from_env('OS_SUDO_TESTING')
+ self.root_helper = os.environ.get('OS_ROOTWRAP_CMD', SUDO_CMD)
self.fail_on_missing_deps = (
- env.get('OS_FAIL_ON_MISSING_DEPS') in base.TRUE_STRING)
+ base.bool_from_env('OS_FAIL_ON_MISSING_DEPS'))
def check_sudo_enabled(self):
if not self.sudo_enabled:
# under the License.
import gc
-import os
import weakref
import mock
# configured to do so since calling gc.collect() after every
# test increases test suite execution time by ~50%.
check_plugin_deallocation = (
- os.environ.get('OS_CHECK_PLUGIN_DEALLOCATION') in base.TRUE_STRING)
+ base.bool_from_env('OS_CHECK_PLUGIN_DEALLOCATION'))
if check_plugin_deallocation:
plugin = weakref.ref(nm._instance.plugin)