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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
import mock
import test_utils
import ceph_hooks as hooks
TO_PATCH = [
'status_set',
'config',
'ceph',
'relation_ids',
'relation_get',
'related_units',
'get_conf',
]
CEPH_MONS = [
'ceph/0',
'ceph/1',
'ceph/2',
]
class ServiceStatusTestCase(test_utils.CharmTestCase):
def setUp(self):
super(ServiceStatusTestCase, self).setUp(hooks, TO_PATCH)
self.config.side_effect = self.test_config.get
def test_assess_status_no_monitor_relation(self):
self.relation_ids.return_value = []
hooks.assess_status()
self.status_set.assert_called_with('blocked', mock.ANY)
def test_assess_status_monitor_relation_incomplete(self):
self.relation_ids.return_value = ['mon:1']
self.related_units.return_value = CEPH_MONS
self.get_conf.return_value = None
hooks.assess_status()
self.status_set.assert_called_with('waiting', mock.ANY)
def test_assess_status_monitor_complete_no_disks(self):
self.relation_ids.return_value = ['mon:1']
self.related_units.return_value = CEPH_MONS
self.get_conf.return_value = 'monitor-bootstrap-key'
self.ceph.get_running_osds.return_value = []
hooks.assess_status()
self.status_set.assert_called_with('blocked', mock.ANY)
def test_assess_status_monitor_complete_disks(self):
self.relation_ids.return_value = ['mon:1']
self.related_units.return_value = CEPH_MONS
self.get_conf.return_value = 'monitor-bootstrap-key'
self.ceph.get_running_osds.return_value = ['12345',
'67890']
hooks.assess_status()
self.status_set.assert_called_with('active', mock.ANY)
|