ceph/src/python-common/ceph/tests/test_drive_group.py

   1 # flake8: noqa
   2 import pytest
   3 import yaml
   4
   5 from ceph.deployment import drive_selection, translate
   6 from ceph.deployment.inventory import Device
   7 from ceph.deployment.service_spec import PlacementSpec, ServiceSpecValidationError
   8 from ceph.tests.utils import _mk_inventory, _mk_device
   9 from ceph.deployment.drive_group import DriveGroupSpec, DeviceSelection, \
  10     DriveGroupValidationError
  11
  12 @pytest.mark.parametrize("test_input",
  13 [
  14     (
  15         [  # new style json
  16             {
  17                 'service_type': 'osd',
  18                 'service_id': 'testing_drivegroup',
  19                 'placement': {'host_pattern': 'hostname'},
  20                 'data_devices': {'paths': ['/dev/sda']}
  21             }
  22         ]
  23     ),
  24 ])
  25 def test_DriveGroup(test_input):
  26     dg = [DriveGroupSpec.from_json(inp) for inp in test_input][0]
  27     assert dg.placement.filter_matching_hosts(lambda label=None, as_hostspec=None: ['hostname']) == ['hostname']
  28     assert dg.service_id == 'testing_drivegroup'
  29     assert all([isinstance(x, Device) for x in dg.data_devices.paths])
  30     assert dg.data_devices.paths[0].path == '/dev/sda'
  31
  32 @pytest.mark.parametrize("test_input",
  33 [
  34     (
  35         {}
  36     ),
  37     (
  38         yaml.safe_load("""
  39 service_type: osd
  40 service_id: mydg
  41 placement:
  42   host_pattern: '*'
  43 data_devices:
  44   limit: 1
  45 """)
  46     )
  47 ])
  48 def test_DriveGroup_fail(test_input):
  49     with pytest.raises(ServiceSpecValidationError):
  50         DriveGroupSpec.from_json(test_input)
  51
  52
  53
  54 def test_drivegroup_pattern():
  55     dg = DriveGroupSpec(PlacementSpec(host_pattern='node[1-3]'), data_devices=DeviceSelection(all=True))
  56     assert dg.placement.filter_matching_hosts(lambda label=None, as_hostspec=None: ['node{}'.format(i) for i in range(10)]) == ['node1', 'node2', 'node3']
  57
  58
  59 def test_drive_selection():
  60     devs = DeviceSelection(paths=['/dev/sda'])
  61     spec = DriveGroupSpec(PlacementSpec('node_name'), data_devices=devs)
  62     assert all([isinstance(x, Device) for x in spec.data_devices.paths])
  63     assert spec.data_devices.paths[0].path == '/dev/sda'
  64
  65     with pytest.raises(DriveGroupValidationError, match='exclusive'):
  66         DeviceSelection(paths=['/dev/sda'], rotational=False)
  67
  68
  69 def test_ceph_volume_command_0():
  70     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
  71                           data_devices=DeviceSelection(all=True)
  72                           )
  73     inventory = _mk_inventory(_mk_device()*2)
  74     sel = drive_selection.DriveSelection(spec, inventory)
  75     cmd = translate.to_ceph_volume(spec, sel, []).run()
  76     assert cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --yes --no-systemd'
  77
  78
  79 def test_ceph_volume_command_1():
  80     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
  81                           data_devices=DeviceSelection(rotational=True),
  82                           db_devices=DeviceSelection(rotational=False)
  83                           )
  84     inventory = _mk_inventory(_mk_device(rotational=True)*2 + _mk_device(rotational=False)*2)
  85     sel = drive_selection.DriveSelection(spec, inventory)
  86     cmd = translate.to_ceph_volume(spec, sel, []).run()
  87     assert cmd == ('lvm batch --no-auto /dev/sda /dev/sdb '
  88                    '--db-devices /dev/sdc /dev/sdd --yes --no-systemd')
  89
  90
  91 def test_ceph_volume_command_2():
  92     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
  93                           data_devices=DeviceSelection(size='200GB:350GB', rotational=True),
  94                           db_devices=DeviceSelection(size='200GB:350GB', rotational=False),
  95                           wal_devices=DeviceSelection(size='10G')
  96                           )
  97     inventory = _mk_inventory(_mk_device(rotational=True, size="300.00 GB")*2 +
  98                               _mk_device(rotational=False, size="300.00 GB")*2 +
  99                               _mk_device(size="10.0 GB", rotational=False)*2
 100                               )
 101     sel = drive_selection.DriveSelection(spec, inventory)
 102     cmd = translate.to_ceph_volume(spec, sel, []).run()
 103     assert cmd == ('lvm batch --no-auto /dev/sda /dev/sdb '
 104                    '--db-devices /dev/sdc /dev/sdd --wal-devices /dev/sde /dev/sdf '
 105                    '--yes --no-systemd')
 106
 107
 108 def test_ceph_volume_command_3():
 109     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
 110                           data_devices=DeviceSelection(size='200GB:350GB', rotational=True),
 111                           db_devices=DeviceSelection(size='200GB:350GB', rotational=False),
 112                           wal_devices=DeviceSelection(size='10G'),
 113                           encrypted=True
 114                           )
 115     inventory = _mk_inventory(_mk_device(rotational=True, size="300.00 GB")*2 +
 116                               _mk_device(rotational=False, size="300.00 GB")*2 +
 117                               _mk_device(size="10.0 GB", rotational=False)*2
 118                               )
 119     sel = drive_selection.DriveSelection(spec, inventory)
 120     cmd = translate.to_ceph_volume(spec, sel, []).run()
 121     assert cmd == ('lvm batch --no-auto /dev/sda /dev/sdb '
 122                    '--db-devices /dev/sdc /dev/sdd '
 123                    '--wal-devices /dev/sde /dev/sdf --dmcrypt '
 124                    '--yes --no-systemd')
 125
 126
 127 def test_ceph_volume_command_4():
 128     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
 129                           data_devices=DeviceSelection(size='200GB:350GB', rotational=True),
 130                           db_devices=DeviceSelection(size='200GB:350GB', rotational=False),
 131                           wal_devices=DeviceSelection(size='10G'),
 132                           block_db_size='500M',
 133                           block_wal_size='500M',
 134                           osds_per_device=3,
 135                           encrypted=True
 136                           )
 137     inventory = _mk_inventory(_mk_device(rotational=True, size="300.00 GB")*2 +
 138                               _mk_device(rotational=False, size="300.00 GB")*2 +
 139                               _mk_device(size="10.0 GB", rotational=False)*2
 140                               )
 141     sel = drive_selection.DriveSelection(spec, inventory)
 142     cmd = translate.to_ceph_volume(spec, sel, []).run()
 143     assert cmd == ('lvm batch --no-auto /dev/sda /dev/sdb '
 144                    '--db-devices /dev/sdc /dev/sdd --wal-devices /dev/sde /dev/sdf '
 145                    '--block-wal-size 500M --block-db-size 500M --dmcrypt '
 146                    '--osds-per-device 3 --yes --no-systemd')
 147
 148
 149 def test_ceph_volume_command_5():
 150     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
 151                           data_devices=DeviceSelection(rotational=True),
 152                           objectstore='filestore'
 153                           )
 154     inventory = _mk_inventory(_mk_device(rotational=True)*2)
 155     sel = drive_selection.DriveSelection(spec, inventory)
 156     cmd = translate.to_ceph_volume(spec, sel, []).run()
 157     assert cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --filestore --yes --no-systemd'
 158
 159
 160 def test_ceph_volume_command_6():
 161     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
 162                           data_devices=DeviceSelection(rotational=False),
 163                           journal_devices=DeviceSelection(rotational=True),
 164                           journal_size='500M',
 165                           objectstore='filestore'
 166                           )
 167     inventory = _mk_inventory(_mk_device(rotational=True)*2 + _mk_device(rotational=False)*2)
 168     sel = drive_selection.DriveSelection(spec, inventory)
 169     cmd = translate.to_ceph_volume(spec, sel, []).run()
 170     assert cmd == ('lvm batch --no-auto /dev/sdc /dev/sdd '
 171                    '--journal-size 500M --journal-devices /dev/sda /dev/sdb '
 172                    '--filestore --yes --no-systemd')
 173
 174
 175 def test_ceph_volume_command_7():
 176     spec = DriveGroupSpec(placement=PlacementSpec(host_pattern='*'),
 177                           data_devices=DeviceSelection(all=True),
 178                           osd_id_claims={'host1': ['0', '1']}
 179                           )
 180     inventory = _mk_inventory(_mk_device(rotational=True)*2)
 181     sel = drive_selection.DriveSelection(spec, inventory)
 182     cmd = translate.to_ceph_volume(spec, sel, ['0', '1']).run()
 183     assert cmd == 'lvm batch --no-auto /dev/sda /dev/sdb --osd-ids 0 1 --yes --no-systemd'