5 from argparse
import ArgumentError
6 from mock
import MagicMock
, patch
8 from ceph_volume
.devices
.lvm
import batch
9 from ceph_volume
.util
import arg_validators
12 class TestBatch(object):
14 def test_batch_instance(self
, is_root
):
18 def test_disjoint_device_lists(self
, factory
):
19 device1
= factory(used_by_ceph
=False, available
=True, abspath
="/dev/sda")
20 device2
= factory(used_by_ceph
=False, available
=True, abspath
="/dev/sdb")
21 devices
= [device1
, device2
]
22 db_devices
= [device2
]
23 with pytest
.raises(Exception) as disjoint_ex
:
24 batch
.ensure_disjoint_device_lists(devices
, db_devices
)
25 assert 'Device lists are not disjoint' in str(disjoint_ex
.value
)
27 @patch('ceph_volume.util.arg_validators.Device')
28 def test_reject_partition(self
, mocked_device
):
29 mocked_device
.return_value
= MagicMock(
31 has_gpt_headers
=False,
33 with pytest
.raises(ArgumentError
):
34 arg_validators
.ValidBatchDevice()('foo')
36 @pytest.mark
.parametrize('format_', ['pretty', 'json', 'json-pretty'])
37 def test_report(self
, format_
, factory
, conf_ceph_stub
, mock_device_generator
):
38 # just ensure reporting works
39 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
40 devs
= [mock_device_generator() for _
in range(5)]
41 args
= factory(data_slots
=1,
54 plan
= b
.get_plan(args
)
58 @pytest.mark
.parametrize('format_', ['json', 'json-pretty'])
59 def test_json_report_valid_empty(self
, format_
, factory
, conf_ceph_stub
, mock_device_generator
):
60 # ensure json reports are valid when empty
61 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
63 args
= factory(data_slots
=1,
76 plan
= b
.get_plan(args
)
78 report
= b
._create
_report
(plan
)
81 @pytest.mark
.parametrize('format_', ['json', 'json-pretty'])
82 def test_json_report_valid_empty_unavailable_fast(self
, format_
, factory
, conf_ceph_stub
, mock_device_generator
):
83 # ensure json reports are valid when empty
84 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
85 devs
= [mock_device_generator() for _
in range(5)]
86 fast_devs
= [mock_device_generator()]
87 fast_devs
[0].available_lvm
= False
88 args
= factory(data_slots
=1,
101 plan
= b
.get_plan(args
)
103 report
= b
._create
_report
(plan
)
107 @pytest.mark
.parametrize('format_', ['json', 'json-pretty'])
108 def test_json_report_valid_empty_unavailable_very_fast(self
, format_
, factory
, conf_ceph_stub
, mock_device_generator
):
109 # ensure json reports are valid when empty
110 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
111 devs
= [mock_device_generator() for _
in range(5)]
112 fast_devs
= [mock_device_generator()]
113 very_fast_devs
= [mock_device_generator()]
114 very_fast_devs
[0].available_lvm
= False
115 args
= factory(data_slots
=1,
121 db_devices
=fast_devs
,
122 wal_devices
=very_fast_devs
,
128 plan
= b
.get_plan(args
)
130 report
= b
._create
_report
(plan
)
133 @pytest.mark
.parametrize('rota', [0, 1])
134 def test_batch_sort_full(self
, factory
, rota
):
135 device1
= factory(used_by_ceph
=False, available
=True, rotational
=rota
, abspath
="/dev/sda")
136 device2
= factory(used_by_ceph
=False, available
=True, rotational
=rota
, abspath
="/dev/sdb")
137 device3
= factory(used_by_ceph
=False, available
=True, rotational
=rota
, abspath
="/dev/sdc")
138 devices
= [device1
, device2
, device3
]
139 args
= factory(report
=True,
145 b
._sort
_rotational
_disks
()
146 assert len(b
.args
.devices
) == 3
148 @pytest.mark
.parametrize('objectstore', ['bluestore', 'filestore'])
149 def test_batch_sort_mixed(self
, factory
, objectstore
):
150 device1
= factory(used_by_ceph
=False, available
=True, rotational
=1, abspath
="/dev/sda")
151 device2
= factory(used_by_ceph
=False, available
=True, rotational
=1, abspath
="/dev/sdb")
152 device3
= factory(used_by_ceph
=False, available
=True, rotational
=0, abspath
="/dev/sdc")
153 devices
= [device1
, device2
, device3
]
154 args
= factory(report
=True,
156 filestore
=False if objectstore
== 'bluestore' else True,
160 b
._sort
_rotational
_disks
()
161 assert len(b
.args
.devices
) == 2
162 if objectstore
== 'bluestore':
163 assert len(b
.args
.db_devices
) == 1
165 assert len(b
.args
.journal_devices
) == 1
167 def test_get_physical_osds_return_len(self
, factory
,
168 mock_devices_available
,
171 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
172 args
= factory(data_slots
=1, osds_per_device
=osds_per_device
,
173 osd_ids
=[], dmcrypt
=False)
174 osds
= batch
.get_physical_osds(mock_devices_available
, args
)
175 assert len(osds
) == len(mock_devices_available
) * osds_per_device
177 def test_get_physical_osds_rel_size(self
, factory
,
178 mock_devices_available
,
181 args
= factory(data_slots
=1, osds_per_device
=osds_per_device
,
182 osd_ids
=[], dmcrypt
=False)
183 osds
= batch
.get_physical_osds(mock_devices_available
, args
)
185 assert osd
.data
[1] == 1.0 / osds_per_device
187 def test_get_physical_osds_abs_size(self
, factory
,
188 mock_devices_available
,
191 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
192 args
= factory(data_slots
=1, osds_per_device
=osds_per_device
,
193 osd_ids
=[], dmcrypt
=False)
194 osds
= batch
.get_physical_osds(mock_devices_available
, args
)
195 for osd
, dev
in zip(osds
, mock_devices_available
):
196 assert osd
.data
[2] == int(dev
.vg_size
[0] / osds_per_device
)
198 def test_get_physical_osds_osd_ids(self
, factory
,
199 mock_devices_available
,
203 def test_get_physical_fast_allocs_length(self
, factory
,
205 mock_devices_available
):
206 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
207 args
= factory(block_db_slots
=None, get_block_db_size
=None)
208 fast
= batch
.get_physical_fast_allocs(mock_devices_available
,
209 'block_db', 2, 2, args
)
210 assert len(fast
) == 2
212 @pytest.mark
.parametrize('occupied_prior', range(7))
213 @pytest.mark
.parametrize('slots,num_devs',
214 [l
for sub
in [list(zip([x
]*x
, range(1, x
+ 1))) for x
in range(1,7)] for l
in sub
])
215 def test_get_physical_fast_allocs_length_existing(self
,
221 mock_device_generator
):
222 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
223 occupied_prior
= min(occupied_prior
, slots
)
224 devs
= [mock_device_generator() for _
in range(num_devs
)]
226 while already_assigned
< occupied_prior
:
227 dev_i
= random
.randint(0, num_devs
- 1)
229 if len(dev
.lvs
) < occupied_prior
:
230 dev
.lvs
.append('foo')
231 dev
.path
= '/dev/bar'
232 already_assigned
= sum([len(d
.lvs
) for d
in devs
])
233 args
= factory(block_db_slots
=None, get_block_db_size
=None)
234 expected_num_osds
= max(len(devs
) * slots
- occupied_prior
, 0)
235 fast
= batch
.get_physical_fast_allocs(devs
,
237 expected_num_osds
, args
)
238 assert len(fast
) == expected_num_osds
239 expected_assignment_on_used_devices
= sum([slots
- len(d
.lvs
) for d
in devs
if len(d
.lvs
) > 0])
240 assert len([f
for f
in fast
if f
[0] == '/dev/bar']) == expected_assignment_on_used_devices
241 assert len([f
for f
in fast
if f
[0] != '/dev/bar']) == expected_num_osds
- expected_assignment_on_used_devices
243 def test_get_lvm_osds_return_len(self
, factory
,
244 mock_lv_device_generator
,
247 conf_ceph_stub('[global]\nfsid=asdf-lkjh')
248 args
= factory(data_slots
=1, osds_per_device
=osds_per_device
,
249 osd_ids
=[], dmcrypt
=False)
250 mock_lvs
= [mock_lv_device_generator()]
251 osds
= batch
.get_lvm_osds(mock_lvs
, args
)
252 assert len(osds
) == 1
255 class TestBatchOsd(object):
257 def test_osd_class_ctor(self
):
258 osd
= batch
.Batch
.OSD('/dev/data', 1, '5G', 1, 1, None)
259 assert osd
.data
== batch
.Batch
.OSD
.VolSpec('/dev/data',
264 def test_add_fast(self
):
265 osd
= batch
.Batch
.OSD('/dev/data', 1, '5G', 1, 1, None)
266 osd
.add_fast_device('/dev/db', 1, '5G', 1, 'block_db')
267 assert osd
.fast
== batch
.Batch
.OSD
.VolSpec('/dev/db',
273 def test_add_very_fast(self
):
274 osd
= batch
.Batch
.OSD('/dev/data', 1, '5G', 1, 1, None)
275 osd
.add_very_fast_device('/dev/wal', 1, '5G', 1)
276 assert osd
.very_fast
== batch
.Batch
.OSD
.VolSpec('/dev/wal',