[ceph.git] / ceph / src / ceph-volume / ceph_volume / tests / devices / lvm / test_batch.py

import pytest
import json
import random

from argparse import ArgumentError
from mock import MagicMock, patch

from ceph_volume.devices.lvm import batch
from ceph_volume.util import arg_validators


class TestBatch(object):

    def test_batch_instance(self, is_root):
        b = batch.Batch([])
        b.main()

    def test_invalid_osd_ids_passed(self):
        with pytest.raises(SystemExit):
            batch.Batch(argv=['--osd-ids', '1', 'foo']).main()

    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")
        devices = [device1, device2]
        db_devices = [device2]
        with pytest.raises(Exception) as disjoint_ex:
            batch.ensure_disjoint_device_lists(devices, db_devices)
        assert 'Device lists are not disjoint' in str(disjoint_ex.value)

    @patch('ceph_volume.util.arg_validators.Device')
    def test_reject_partition(self, mocked_device):
        mocked_device.return_value = MagicMock(
            is_partition=True,
            has_gpt_headers=False,
            has_partitions=False,
        )
        with pytest.raises(ArgumentError):
            arg_validators.ValidBatchDevice()('foo')

    @pytest.mark.parametrize('format_', ['pretty', 'json', 'json-pretty'])
    def test_report(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # just ensure reporting works
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = [mock_device_generator() for _ in range(5)]
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=[],
                       wal_devices=[],
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                       data_allocate_fraction=1.0,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        b.report(plan)

    @pytest.mark.parametrize('format_', ['json', 'json-pretty'])
    def test_json_report_valid_empty(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # ensure json reports are valid when empty
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = []
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=[],
                       wal_devices=[],
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                       data_allocate_fraction=1.0,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        report = b._create_report(plan)
        json.loads(report)

    @pytest.mark.parametrize('format_', ['json', 'json-pretty'])
    def test_json_report_valid_empty_unavailable_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # ensure json reports are valid when empty
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = [mock_device_generator() for _ in range(5)]
        fast_devs = [mock_device_generator()]
        fast_devs[0].available_lvm = False
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=fast_devs,
                       wal_devices=[],
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                       data_allocate_fraction=1.0,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        report = b._create_report(plan)
        json.loads(report)


    @pytest.mark.parametrize('format_', ['json', 'json-pretty'])
    def test_json_report_valid_empty_unavailable_very_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
        # ensure json reports are valid when empty
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        devs = [mock_device_generator() for _ in range(5)]
        fast_devs = [mock_device_generator()]
        very_fast_devs = [mock_device_generator()]
        very_fast_devs[0].available_lvm = False
        args = factory(data_slots=1,
                       osds_per_device=1,
                       osd_ids=[],
                       report=True,
                       format=format_,
                       devices=devs,
                       db_devices=fast_devs,
                       wal_devices=very_fast_devs,
                       bluestore=True,
                       block_db_size="1G",
                       dmcrypt=True,
                       data_allocate_fraction=1.0,
                      )
        b = batch.Batch([])
        plan = b.get_plan(args)
        b.args = args
        report = b._create_report(plan)
        json.loads(report)

    @pytest.mark.parametrize('rota', [0, 1])
    def test_batch_sort_full(self, factory, rota):
        device1 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sda")
        device2 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdb")
        device3 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdc")
        devices = [device1, device2, device3]
        args = factory(report=True,
                       devices=devices,
                       filestore=False,
                      )
        b = batch.Batch([])
        b.args = args
        b._sort_rotational_disks()
        assert len(b.args.devices) == 3

    @pytest.mark.parametrize('objectstore', ['bluestore', 'filestore'])
    def test_batch_sort_mixed(self, factory, objectstore):
        device1 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sda")
        device2 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sdb")
        device3 = factory(used_by_ceph=False, available=True, rotational=0, abspath="/dev/sdc")
        devices = [device1, device2, device3]
        args = factory(report=True,
                       devices=devices,
                       filestore=False if objectstore == 'bluestore' else True,
                      )
        b = batch.Batch([])
        b.args = args
        b._sort_rotational_disks()
        assert len(b.args.devices) == 2
        if objectstore == 'bluestore':
            assert len(b.args.db_devices) == 1
        else:
            assert len(b.args.journal_devices) == 1

    def test_get_physical_osds_return_len(self, factory,
                                          mock_devices_available,
                                          conf_ceph_stub,
                                          osds_per_device):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False,
                       data_allocate_fraction=1.0)
        osds = batch.get_physical_osds(mock_devices_available, args)
        assert len(osds) == len(mock_devices_available) * osds_per_device

    def test_get_physical_osds_rel_size(self, factory,
                                          mock_devices_available,
                                          conf_ceph_stub,
                                          osds_per_device,
                                          data_allocate_fraction):
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False,
                       data_allocate_fraction=data_allocate_fraction)
        osds = batch.get_physical_osds(mock_devices_available, args)
        for osd in osds:
            assert osd.data[1] == data_allocate_fraction / osds_per_device

    def test_get_physical_osds_abs_size(self, factory,
                                          mock_devices_available,
                                          conf_ceph_stub,
                                          osds_per_device,
                                          data_allocate_fraction):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False,
                       data_allocate_fraction=data_allocate_fraction)
        osds = batch.get_physical_osds(mock_devices_available, args)
        for osd, dev in zip(osds, mock_devices_available):
            assert osd.data[2] == int(dev.vg_size[0] * (data_allocate_fraction / osds_per_device))

    def test_get_physical_osds_osd_ids(self, factory,
                                          mock_devices_available,
                                          osds_per_device):
        pass

    def test_get_physical_fast_allocs_length(self, factory,
                                             conf_ceph_stub,
                                             mock_devices_available):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(block_db_slots=None, get_block_db_size=None)
        fast = batch.get_physical_fast_allocs(mock_devices_available,
                                              'block_db', 2, 2, args)
        assert len(fast) == 2

    def test_batch_fast_allocations_one_block_db_length(self, factory, conf_ceph_stub,
                                                  mock_lv_device_generator):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')

        b = batch.Batch([])
        db_lv_devices = [mock_lv_device_generator()]
        fast = b.fast_allocations(db_lv_devices, 1, 0, 'block_db')
        assert len(fast) == 1

    @pytest.mark.parametrize('occupied_prior', range(7))
    @pytest.mark.parametrize('slots,num_devs',
                             [l for sub in [list(zip([x]*x, range(1, x + 1))) for x in range(1,7)] for l in sub])
    def test_get_physical_fast_allocs_length_existing(self,
                                                      num_devs,
                                                      slots,
                                                      occupied_prior,
                                                      factory,
                                                      conf_ceph_stub,
                                                      mock_device_generator):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        occupied_prior = min(occupied_prior, slots)
        devs = [mock_device_generator() for _ in range(num_devs)]
        already_assigned = 0
        while already_assigned < occupied_prior:
            dev_i = random.randint(0, num_devs - 1)
            dev = devs[dev_i]
            if len(dev.lvs) < occupied_prior:
                dev.lvs.append('foo')
                dev.path = '/dev/bar'
                already_assigned = sum([len(d.lvs) for d in devs])
        args = factory(block_db_slots=None, get_block_db_size=None)
        expected_num_osds = max(len(devs) * slots - occupied_prior, 0)
        fast = batch.get_physical_fast_allocs(devs,
                                              'block_db', slots,
                                              expected_num_osds, args)
        assert len(fast) == expected_num_osds
        expected_assignment_on_used_devices = sum([slots - len(d.lvs) for d in devs if len(d.lvs) > 0])
        assert len([f for f in fast if f[0] == '/dev/bar']) == expected_assignment_on_used_devices
        assert len([f for f in fast if f[0] != '/dev/bar']) == expected_num_osds - expected_assignment_on_used_devices

    def test_get_lvm_osds_return_len(self, factory,
                                     mock_lv_device_generator,
                                     conf_ceph_stub,
                                     osds_per_device):
        conf_ceph_stub('[global]\nfsid=asdf-lkjh')
        args = factory(data_slots=1, osds_per_device=osds_per_device,
                       osd_ids=[], dmcrypt=False)
        mock_lvs = [mock_lv_device_generator()]
        osds = batch.get_lvm_osds(mock_lvs, args)
        assert len(osds) == 1


class TestBatchOsd(object):

    def test_osd_class_ctor(self):
        osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
        assert osd.data == batch.Batch.OSD.VolSpec('/dev/data',
                                                   1,
                                                   '5G',
                                                   1,
                                                   'data')
    def test_add_fast(self):
        osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
        osd.add_fast_device('/dev/db', 1, '5G', 1, 'block_db')
        assert osd.fast == batch.Batch.OSD.VolSpec('/dev/db',
                                                   1,
                                                   '5G',
                                                   1,
                                                   'block_db')

    def test_add_very_fast(self):
        osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
        osd.add_very_fast_device('/dev/wal', 1, '5G', 1)
        assert osd.very_fast == batch.Batch.OSD.VolSpec('/dev/wal',
                                                        1,
                                                        '5G',
                                                        1,
                                                        'block_wal')
Commit	Line	Data
494da23a	1	import pytest
f6b5b4d7	2	import json
f91f0fd5 TL	3	import random
	4
	5	from argparse import ArgumentError
	6	from mock import MagicMock, patch
	7
91327a77	8	from ceph_volume.devices.lvm import batch
f91f0fd5	9	from ceph_volume.util import arg_validators
91327a77 AA	10
91327a77 AA	11
11fdf7f2	12	class TestBatch(object):
f64942e4 AA	13
	14	def test_batch_instance(self, is_root):
	15	b = batch.Batch([])
	16	b.main()
	17
a4b75251 TL	18	def test_invalid_osd_ids_passed(self):
	19	with pytest.raises(SystemExit):
	20	batch.Batch(argv=['--osd-ids', '1', 'foo']).main()
	21
494da23a TL	22	def test_disjoint_device_lists(self, factory):
	23	device1 = factory(used_by_ceph=False, available=True, abspath="/dev/sda")
	24	device2 = factory(used_by_ceph=False, available=True, abspath="/dev/sdb")
f91f0fd5 TL	25	devices = [device1, device2]
f91f0fd5 TL	26	db_devices = [device2]
494da23a	27	with pytest.raises(Exception) as disjoint_ex:
f91f0fd5	28	batch.ensure_disjoint_device_lists(devices, db_devices)
494da23a TL	29	assert 'Device lists are not disjoint' in str(disjoint_ex.value)
494da23a TL	30
f91f0fd5 TL	31	@patch('ceph_volume.util.arg_validators.Device')
	32	def test_reject_partition(self, mocked_device):
	33	mocked_device.return_value = MagicMock(
	34	is_partition=True,
	35	has_gpt_headers=False,
1d09f67e	36	has_partitions=False,
f91f0fd5 TL	37	)
	38	with pytest.raises(ArgumentError):
	39	arg_validators.ValidBatchDevice()('foo')
f64942e4	40
f91f0fd5 TL	41	@pytest.mark.parametrize('format_', ['pretty', 'json', 'json-pretty'])
	42	def test_report(self, format_, factory, conf_ceph_stub, mock_device_generator):
	43	# just ensure reporting works
	44	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	45	devs = [mock_device_generator() for _ in range(5)]
	46	args = factory(data_slots=1,
	47	osds_per_device=1,
	48	osd_ids=[],
	49	report=True,
	50	format=format_,
	51	devices=devs,
	52	db_devices=[],
	53	wal_devices=[],
	54	bluestore=True,
	55	block_db_size="1G",
	56	dmcrypt=True,
20effc67	57	data_allocate_fraction=1.0,
f91f0fd5 TL	58	)
	59	b = batch.Batch([])
	60	plan = b.get_plan(args)
	61	b.args = args
	62	b.report(plan)
91327a77	63
f91f0fd5 TL	64	@pytest.mark.parametrize('format_', ['json', 'json-pretty'])
	65	def test_json_report_valid_empty(self, format_, factory, conf_ceph_stub, mock_device_generator):
	66	# ensure json reports are valid when empty
	67	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	68	devs = []
	69	args = factory(data_slots=1,
	70	osds_per_device=1,
	71	osd_ids=[],
	72	report=True,
	73	format=format_,
	74	devices=devs,
	75	db_devices=[],
	76	wal_devices=[],
	77	bluestore=True,
	78	block_db_size="1G",
	79	dmcrypt=True,
20effc67	80	data_allocate_fraction=1.0,
f91f0fd5 TL	81	)
	82	b = batch.Batch([])
	83	plan = b.get_plan(args)
	84	b.args = args
	85	report = b._create_report(plan)
	86	json.loads(report)
91327a77	87
f91f0fd5 TL	88	@pytest.mark.parametrize('format_', ['json', 'json-pretty'])
	89	def test_json_report_valid_empty_unavailable_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
	90	# ensure json reports are valid when empty
	91	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	92	devs = [mock_device_generator() for _ in range(5)]
	93	fast_devs = [mock_device_generator()]
	94	fast_devs[0].available_lvm = False
	95	args = factory(data_slots=1,
	96	osds_per_device=1,
	97	osd_ids=[],
	98	report=True,
	99	format=format_,
	100	devices=devs,
	101	db_devices=fast_devs,
	102	wal_devices=[],
	103	bluestore=True,
	104	block_db_size="1G",
	105	dmcrypt=True,
20effc67	106	data_allocate_fraction=1.0,
f91f0fd5	107	)
92f5a8d4	108	b = batch.Batch([])
f91f0fd5	109	plan = b.get_plan(args)
92f5a8d4	110	b.args = args
f91f0fd5 TL	111	report = b._create_report(plan)
	112	json.loads(report)
	113
	114
	115	@pytest.mark.parametrize('format_', ['json', 'json-pretty'])
	116	def test_json_report_valid_empty_unavailable_very_fast(self, format_, factory, conf_ceph_stub, mock_device_generator):
	117	# ensure json reports are valid when empty
	118	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	119	devs = [mock_device_generator() for _ in range(5)]
	120	fast_devs = [mock_device_generator()]
	121	very_fast_devs = [mock_device_generator()]
	122	very_fast_devs[0].available_lvm = False
	123	args = factory(data_slots=1,
	124	osds_per_device=1,
	125	osd_ids=[],
	126	report=True,
	127	format=format_,
	128	devices=devs,
	129	db_devices=fast_devs,
	130	wal_devices=very_fast_devs,
	131	bluestore=True,
	132	block_db_size="1G",
	133	dmcrypt=True,
20effc67	134	data_allocate_fraction=1.0,
f91f0fd5 TL	135	)
	136	b = batch.Batch([])
	137	plan = b.get_plan(args)
	138	b.args = args
	139	report = b._create_report(plan)
	140	json.loads(report)
	141
	142	@pytest.mark.parametrize('rota', [0, 1])
	143	def test_batch_sort_full(self, factory, rota):
	144	device1 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sda")
	145	device2 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdb")
	146	device3 = factory(used_by_ceph=False, available=True, rotational=rota, abspath="/dev/sdc")
	147	devices = [device1, device2, device3]
	148	args = factory(report=True,
	149	devices=devices,
	150	filestore=False,
	151	)
f6b5b4d7 TL	152	b = batch.Batch([])
f6b5b4d7 TL	153	b.args = args
f91f0fd5 TL	154	b._sort_rotational_disks()
	155	assert len(b.args.devices) == 3
	156
	157	@pytest.mark.parametrize('objectstore', ['bluestore', 'filestore'])
	158	def test_batch_sort_mixed(self, factory, objectstore):
	159	device1 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sda")
	160	device2 = factory(used_by_ceph=False, available=True, rotational=1, abspath="/dev/sdb")
	161	device3 = factory(used_by_ceph=False, available=True, rotational=0, abspath="/dev/sdc")
	162	devices = [device1, device2, device3]
	163	args = factory(report=True,
	164	devices=devices,
	165	filestore=False if objectstore == 'bluestore' else True,
	166	)
	167	b = batch.Batch([])
	168	b.args = args
	169	b._sort_rotational_disks()
	170	assert len(b.args.devices) == 2
	171	if objectstore == 'bluestore':
	172	assert len(b.args.db_devices) == 1
	173	else:
	174	assert len(b.args.journal_devices) == 1
	175
	176	def test_get_physical_osds_return_len(self, factory,
	177	mock_devices_available,
	178	conf_ceph_stub,
	179	osds_per_device):
	180	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	181	args = factory(data_slots=1, osds_per_device=osds_per_device,
20effc67 TL	182	osd_ids=[], dmcrypt=False,
20effc67 TL	183	data_allocate_fraction=1.0)
f91f0fd5 TL	184	osds = batch.get_physical_osds(mock_devices_available, args)
	185	assert len(osds) == len(mock_devices_available) * osds_per_device
	186
	187	def test_get_physical_osds_rel_size(self, factory,
	188	mock_devices_available,
	189	conf_ceph_stub,
20effc67 TL	190	osds_per_device,
20effc67 TL	191	data_allocate_fraction):
f91f0fd5	192	args = factory(data_slots=1, osds_per_device=osds_per_device,
20effc67 TL	193	osd_ids=[], dmcrypt=False,
20effc67 TL	194	data_allocate_fraction=data_allocate_fraction)
f91f0fd5 TL	195	osds = batch.get_physical_osds(mock_devices_available, args)
f91f0fd5 TL	196	for osd in osds:
20effc67	197	assert osd.data[1] == data_allocate_fraction / osds_per_device
f91f0fd5 TL	198
	199	def test_get_physical_osds_abs_size(self, factory,
	200	mock_devices_available,
	201	conf_ceph_stub,
20effc67 TL	202	osds_per_device,
20effc67 TL	203	data_allocate_fraction):
f91f0fd5 TL	204	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
f91f0fd5 TL	205	args = factory(data_slots=1, osds_per_device=osds_per_device,
20effc67 TL	206	osd_ids=[], dmcrypt=False,
20effc67 TL	207	data_allocate_fraction=data_allocate_fraction)
f91f0fd5 TL	208	osds = batch.get_physical_osds(mock_devices_available, args)
f91f0fd5 TL	209	for osd, dev in zip(osds, mock_devices_available):
20effc67	210	assert osd.data[2] == int(dev.vg_size[0] * (data_allocate_fraction / osds_per_device))
f91f0fd5 TL	211
	212	def test_get_physical_osds_osd_ids(self, factory,
	213	mock_devices_available,
	214	osds_per_device):
	215	pass
	216
	217	def test_get_physical_fast_allocs_length(self, factory,
	218	conf_ceph_stub,
	219	mock_devices_available):
	220	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	221	args = factory(block_db_slots=None, get_block_db_size=None)
	222	fast = batch.get_physical_fast_allocs(mock_devices_available,
	223	'block_db', 2, 2, args)
	224	assert len(fast) == 2
	225
522d829b TL	226	def test_batch_fast_allocations_one_block_db_length(self, factory, conf_ceph_stub,
	227	mock_lv_device_generator):
	228	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	229
	230	b = batch.Batch([])
	231	db_lv_devices = [mock_lv_device_generator()]
	232	fast = b.fast_allocations(db_lv_devices, 1, 0, 'block_db')
	233	assert len(fast) == 1
	234
f91f0fd5 TL	235	@pytest.mark.parametrize('occupied_prior', range(7))
	236	@pytest.mark.parametrize('slots,num_devs',
	237	[l for sub in [list(zip([x]*x, range(1, x + 1))) for x in range(1,7)] for l in sub])
	238	def test_get_physical_fast_allocs_length_existing(self,
	239	num_devs,
	240	slots,
	241	occupied_prior,
	242	factory,
	243	conf_ceph_stub,
	244	mock_device_generator):
	245	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	246	occupied_prior = min(occupied_prior, slots)
	247	devs = [mock_device_generator() for _ in range(num_devs)]
	248	already_assigned = 0
	249	while already_assigned < occupied_prior:
	250	dev_i = random.randint(0, num_devs - 1)
	251	dev = devs[dev_i]
	252	if len(dev.lvs) < occupied_prior:
	253	dev.lvs.append('foo')
	254	dev.path = '/dev/bar'
	255	already_assigned = sum([len(d.lvs) for d in devs])
	256	args = factory(block_db_slots=None, get_block_db_size=None)
	257	expected_num_osds = max(len(devs) * slots - occupied_prior, 0)
	258	fast = batch.get_physical_fast_allocs(devs,
	259	'block_db', slots,
	260	expected_num_osds, args)
	261	assert len(fast) == expected_num_osds
	262	expected_assignment_on_used_devices = sum([slots - len(d.lvs) for d in devs if len(d.lvs) > 0])
	263	assert len([f for f in fast if f[0] == '/dev/bar']) == expected_assignment_on_used_devices
	264	assert len([f for f in fast if f[0] != '/dev/bar']) == expected_num_osds - expected_assignment_on_used_devices
	265
	266	def test_get_lvm_osds_return_len(self, factory,
	267	mock_lv_device_generator,
	268	conf_ceph_stub,
	269	osds_per_device):
	270	conf_ceph_stub('[global]\nfsid=asdf-lkjh')
	271	args = factory(data_slots=1, osds_per_device=osds_per_device,
	272	osd_ids=[], dmcrypt=False)
	273	mock_lvs = [mock_lv_device_generator()]
	274	osds = batch.get_lvm_osds(mock_lvs, args)
	275	assert len(osds) == 1
	276
	277
	278	class TestBatchOsd(object):
	279
	280	def test_osd_class_ctor(self):
	281	osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
	282	assert osd.data == batch.Batch.OSD.VolSpec('/dev/data',
	283	1,
	284	'5G',
	285	1,
	286	'data')
	287	def test_add_fast(self):
	288	osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
	289	osd.add_fast_device('/dev/db', 1, '5G', 1, 'block_db')
	290	assert osd.fast == batch.Batch.OSD.VolSpec('/dev/db',
	291	1,
	292	'5G',
	293	1,
	294	'block_db')
	295
	296	def test_add_very_fast(self):
	297	osd = batch.Batch.OSD('/dev/data', 1, '5G', 1, 1, None)
	298	osd.add_very_fast_device('/dev/wal', 1, '5G', 1)
299	assert osd.very_fast == batch.Batch.OSD.VolSpec('/dev/wal',
300	1,
301	'5G',
302	1,
303	'block_wal')