self.args = parser.parse_args(argv)
self.parser = parser
for dev_list in ['', 'db_', 'wal_', 'journal_']:
- setattr(self, '{}usable'.format(dev_list), [])
+ setattr(self, '{}usable'.format(dev_list), set())
def get_devices(self):
# remove devices with partitions
def _get_explicit_strategy(self):
# TODO assert that none of the device lists overlap?
self._filter_devices()
+ self._ensure_disjoint_device_lists()
if self.args.bluestore:
if self.db_usable or self.wal_usable:
self.strategy = strategies.bluestore.MixedType(
dev_list_prop = '{}devices'.format(dev_list)
if hasattr(self.args, dev_list_prop):
usable_dev_list_prop = '{}usable'.format(dev_list)
- usable = [d for d in getattr(self.args, dev_list_prop) if d.available]
+ usable = set([d for d in getattr(self.args, dev_list_prop) if
+ d.available])
setattr(self, usable_dev_list_prop, usable)
self.filtered_devices.update({d: used_reason for d in
getattr(self.args, dev_list_prop)
if d.used_by_ceph})
+
+ def _ensure_disjoint_device_lists(self):
+ # check that all device lists are disjoint with each other
+ if not(self.usable.isdisjoint(self.db_usable) and
+ self.usable.isdisjoint(self.wal_usable) and
+ self.usable.isdisjoint(self.journal_usable) and
+ self.db_usable.isdisjoint(self.wal_usable)):
+ raise Exception('Device lists are not disjoint')
+import pytest
from ceph_volume.devices.lvm import batch
result = b.get_devices().strip()
assert result == '* /dev/vdf 20.00 GB rotational'
+ def test_disjoint_device_lists(self, factory):
+ device1 = factory(used_by_ceph=False, available=True, abspath="/dev/sda")
+ device2 = factory(used_by_ceph=False, available=True, abspath="/dev/sdb")
+ b = batch.Batch([])
+ b.args.devices = [device1, device2]
+ b.args.db_devices = [device2]
+ b._filter_devices()
+ with pytest.raises(Exception):
+ b._ensure_disjoint_device_lists()
+
class TestFilterDevices(object):