return strutils.bool_from_string(value, strict=strict, default=default)
+def get_test_timeout(default=0):
+ return int(os.environ.get('OS_TEST_TIMEOUT', 0))
+
+
class AttributeDict(dict):
"""
nuke_handlers=capture_logs,
))
- test_timeout = int(os.environ.get('OS_TEST_TIMEOUT', 0))
+ test_timeout = get_test_timeout()
if test_timeout == -1:
test_timeout = 0
if test_timeout > 0:
from neutron.agent.linux import ovsdb_monitor
from neutron.agent.linux import utils
+from neutron.tests import base as tests_base
from neutron.tests.common import net_helpers
from neutron.tests.functional.agent.linux import base as linux_base
from neutron.tests.functional import base as functional_base
self.monitor = ovsdb_monitor.SimpleInterfaceMonitor()
self.addCleanup(self.monitor.stop)
- self.monitor.start(block=True, timeout=60)
+ # In case a global test timeout isn't set or disabled, use a
+ # value that will ensure the monitor has time to start.
+ timeout = max(tests_base.get_test_timeout(), 60)
+ self.monitor.start(block=True, timeout=timeout)
def test_has_updates(self):
self.assertTrue(self.monitor.has_updates,