/
test_object_life_cycle.py
41 lines (32 loc) · 1.79 KB
/
test_object_life_cycle.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from cloudbridge.cloud.base import helpers as cb_helpers
from cloudbridge.cloud.interfaces import VolumeState
from cloudbridge.cloud.interfaces.exceptions import WaitStateException
from test import helpers
from test.helpers import ProviderTestBase
class CloudObjectLifeCycleTestCase(ProviderTestBase):
_multiprocess_can_split_ = True
@helpers.skipIfNoService(['storage.volumes'])
def test_object_life_cycle(self):
# Test object life cycle methods by using a volume.
label = "cb-objlifecycle-{0}".format(helpers.get_uuid())
test_vol = None
with cb_helpers.cleanup_action(lambda: test_vol.delete()):
test_vol = self.provider.storage.volumes.create(
label, 1,
helpers.get_provider_test_data(self.provider, "placement"))
# Waiting for an invalid timeout should raise an exception
with self.assertRaises(AssertionError):
test_vol.wait_for([VolumeState.ERROR], timeout=-1, interval=1)
with self.assertRaises(AssertionError):
test_vol.wait_for([VolumeState.ERROR], timeout=1, interval=-1)
# If interval < timeout, an exception should be raised
with self.assertRaises(AssertionError):
test_vol.wait_for([VolumeState.ERROR], timeout=10, interval=20)
test_vol.wait_till_ready()
# Hitting a terminal state should raise an exception
with self.assertRaises(WaitStateException):
test_vol.wait_for([VolumeState.ERROR],
terminal_states=[VolumeState.AVAILABLE])
# Hitting the timeout should raise an exception
with self.assertRaises(WaitStateException):
test_vol.wait_for([VolumeState.ERROR], timeout=0, interval=0)