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

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