mirror of
https://github.com/ceph/ceph
synced 2025-02-19 08:57:27 +00:00
This is to make it easier to see what actually timed out when scanning error logs Signed-off-by: Zack Cerza <zack.cerza@inktank.com>
64 lines
1.7 KiB
Python
64 lines
1.7 KiB
Python
from pytest import raises
|
|
from teuthology import contextutil
|
|
|
|
|
|
class TestSafeWhile(object):
|
|
|
|
def setup(self):
|
|
self.fake_sleep = lambda s: True
|
|
self.s_while = contextutil.safe_while
|
|
|
|
def test_6_5_10_deal(self):
|
|
with raises(contextutil.MaxWhileTries):
|
|
with self.s_while(_sleeper=self.fake_sleep) as bomb:
|
|
while 1:
|
|
bomb()
|
|
|
|
def test_6_0_1_deal(self):
|
|
with raises(contextutil.MaxWhileTries) as error:
|
|
with self.s_while(
|
|
tries=1,
|
|
_sleeper=self.fake_sleep
|
|
) as bomb:
|
|
while 1:
|
|
bomb()
|
|
|
|
msg = error.value[0]
|
|
assert 'waiting for 6 seconds' in msg
|
|
|
|
def test_1_0_10_deal(self):
|
|
with raises(contextutil.MaxWhileTries) as error:
|
|
with self.s_while(
|
|
sleep=1,
|
|
_sleeper=self.fake_sleep
|
|
) as bomb:
|
|
while 1:
|
|
bomb()
|
|
|
|
msg = error.value[0]
|
|
assert 'waiting for 10 seconds' in msg
|
|
|
|
def test_6_1_10_deal(self):
|
|
with raises(contextutil.MaxWhileTries) as error:
|
|
with self.s_while(
|
|
increment=1,
|
|
_sleeper=self.fake_sleep
|
|
) as bomb:
|
|
while 1:
|
|
bomb()
|
|
|
|
msg = error.value[0]
|
|
assert 'waiting for 105 seconds' in msg
|
|
|
|
def test_action(self):
|
|
with raises(contextutil.MaxWhileTries) as error:
|
|
with self.s_while(
|
|
action='doing the thing',
|
|
_sleeper=self.fake_sleep
|
|
) as bomb:
|
|
while 1:
|
|
bomb()
|
|
|
|
msg = error.value[0]
|
|
assert "'doing the thing'" in msg
|