src/ceph/src/ceph-disk/tests/test_prepare.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright (C) 2015, 2016 Red Hat <contact@redhat.com>
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU Library Public License as published by
   7 # the Free Software Foundation; either version 2, or (at your option)
   8 # any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU Library Public License for more details.
  14 #
  15 import argparse
  16 import configobj
  17 import mock
  18 import os
  19 import platform
  20 import pytest
  21 import shutil
  22 import tempfile
  23
  24 from ceph_disk import main
  25
  26
  27 class Base(object):
  28
  29     def setup_class(self):
  30         main.setup_logging(True, False)
  31         os.environ['PATH'] = "..:" + os.environ['PATH']
  32
  33     def setup(self):
  34         _, self.conf_file = tempfile.mkstemp()
  35         os.environ['CEPH_CONF'] = self.conf_file
  36         self.conf = configobj.ConfigObj(self.conf_file)
  37         self.conf['global'] = {}
  38
  39     def teardown(self):
  40         os.unlink(self.conf_file)
  41
  42     def save_conf(self):
  43         self.conf.write(open(self.conf_file, 'wb'))
  44
  45
  46 class TestPrepare(Base):
  47
  48     def test_init_filestore_dir(self):
  49         parser = argparse.ArgumentParser('ceph-disk')
  50         subparsers = parser.add_subparsers()
  51         main.Prepare.set_subparser(subparsers)
  52
  53         data = tempfile.mkdtemp()
  54         main.setup_statedir(data)
  55         args = parser.parse_args([
  56             'prepare',
  57             data,
  58             '--filestore',
  59         ])
  60
  61         def set_type(self):
  62             self.type = self.FILE
  63         with mock.patch.multiple(main.PrepareData,
  64                                  set_type=set_type):
  65             prepare = main.Prepare.factory(args)
  66         assert isinstance(prepare.data, main.PrepareFilestoreData)
  67         assert prepare.data.is_file()
  68         assert isinstance(prepare.journal, main.PrepareJournal)
  69         assert prepare.journal.is_none()
  70         prepare.prepare()
  71         assert os.path.exists(os.path.join(data, 'fsid'))
  72         shutil.rmtree(data)
  73
  74     @mock.patch('stat.S_ISBLK')
  75     @mock.patch('ceph_disk.main.is_partition')
  76     def test_init_filestore_dev(self, m_is_partition, m_s_isblk):
  77         m_s_isblk.return_value = True
  78
  79         parser = argparse.ArgumentParser('ceph-disk')
  80         subparsers = parser.add_subparsers()
  81         main.Prepare.set_subparser(subparsers)
  82
  83         m_is_partition.return_value = False
  84         _, data = tempfile.mkstemp()
  85
  86         args = parser.parse_args([
  87             'prepare',
  88             data,
  89             '--filestore',
  90         ])
  91         prepare = main.Prepare.factory(args)
  92         assert isinstance(prepare.data, main.PrepareData)
  93         assert prepare.data.is_device()
  94         assert isinstance(prepare.journal, main.PrepareJournal)
  95         assert prepare.journal.is_device()
  96
  97     def test_init_default_dir(self):
  98         parser = argparse.ArgumentParser('ceph-disk')
  99         subparsers = parser.add_subparsers()
 100         main.Prepare.set_subparser(subparsers)
 101
 102         data = tempfile.mkdtemp()
 103         main.setup_statedir(data)
 104         args = parser.parse_args([
 105             'prepare',
 106             data,
 107         ])
 108
 109         def set_type(self):
 110             self.type = self.FILE
 111         with mock.patch.multiple(main.PrepareData,
 112                                  set_type=set_type):
 113             prepare = main.Prepare.factory(args)
 114         assert isinstance(prepare.data, main.PrepareBluestoreData)
 115         assert prepare.data.is_file()
 116         prepare.prepare()
 117         assert os.path.exists(os.path.join(data, 'fsid'))
 118         shutil.rmtree(data)
 119
 120     def test_set_subparser(self):
 121         parser = argparse.ArgumentParser('ceph-disk')
 122         subparsers = parser.add_subparsers()
 123         main.Prepare.set_subparser(subparsers)
 124         osd_uuid = 'OSD_UUID'
 125         journal_uuid = 'JOURNAL_UUID'
 126         fs_type = 'xfs'
 127         data = 'DATA'
 128         args = parser.parse_args([
 129             'prepare',
 130             '--osd-uuid', osd_uuid,
 131             '--journal-uuid', journal_uuid,
 132             '--fs-type', fs_type,
 133             data,
 134         ])
 135         assert args.journal_uuid == journal_uuid
 136         assert args.osd_uuid == osd_uuid
 137         assert args.fs_type == fs_type
 138         assert args.data == data
 139
 140
 141 class TestDevice(Base):
 142
 143     @mock.patch('ceph_disk.main.is_partition')
 144     def test_init(self, m_is_partition):
 145         m_is_partition.return_value = False
 146         device = main.Device('/dev/wholedisk', argparse.Namespace())
 147         assert device.dev_size is None
 148
 149     @mock.patch('ceph_disk.main.is_partition')
 150     @mock.patch('ceph_disk.main.get_free_partition_index')
 151     @mock.patch('ceph_disk.main.update_partition')
 152     @mock.patch('ceph_disk.main.get_dm_uuid')
 153     @mock.patch('ceph_disk.main.get_dev_size')
 154     @mock.patch('ceph_disk.main.command_check_call')
 155     def test_create_partition(self,
 156                               m_command_check_call,
 157                               m_get_dev_size,
 158                               m_get_dm_uuid,
 159                               m_update_partition,
 160                               m_get_free_partition_index,
 161                               m_is_partition):
 162         if platform.system() == 'FreeBSD':
 163             return
 164         m_is_partition.return_value = False
 165         partition_number = 1
 166         m_get_free_partition_index.return_value = partition_number
 167         path = '/dev/wholedisk'
 168         device = main.Device(path, argparse.Namespace(dmcrypt=False))
 169         uuid = 'UUID'
 170         m_get_dm_uuid.return_value = uuid
 171         size = 200
 172         m_get_dev_size.return_value = size + 100
 173         name = 'journal'
 174         actual_partition_number = device.create_partition(
 175             uuid=uuid, name=name, size=size)
 176         assert actual_partition_number == partition_number
 177         command = ['sgdisk',
 178                    '--new=%d:0:+%dM' % (partition_number, size),
 179                    '--change-name=%d:ceph %s' % (partition_number, name),
 180                    '--partition-guid=%d:%s' % (partition_number, uuid),
 181                    '--typecode=%d:%s' % (
 182                        partition_number,
 183                        main.PTYPE['regular']['journal']['ready']),
 184                    '--mbrtogpt', '--', path]
 185         m_command_check_call.assert_called_with(command, exit=True)
 186         m_update_partition.assert_called_with(path, 'created')
 187
 188         actual_partition_number = device.create_partition(
 189             uuid=uuid, name=name)
 190         command = ['sgdisk',
 191                    '--largest-new=%d' % partition_number,
 192                    '--change-name=%d:ceph %s' % (partition_number, name),
 193                    '--partition-guid=%d:%s' % (partition_number, uuid),
 194                    '--typecode=%d:%s' % (
 195                        partition_number,
 196                        main.PTYPE['regular']['journal']['ready']),
 197                    '--mbrtogpt', '--', path]
 198         m_command_check_call.assert_called_with(command, exit=True)
 199
 200
 201 class TestDevicePartition(Base):
 202
 203     def test_init(self):
 204         partition = main.DevicePartition(argparse.Namespace())
 205         for name in ('osd', 'journal'):
 206             assert (main.PTYPE['regular'][name]['ready'] ==
 207                     partition.ptype_for_name(name))
 208
 209     def test_get_uuid(self):
 210         partition = main.DevicePartition(argparse.Namespace())
 211         uuid = 'UUID'
 212         with mock.patch.multiple(main,
 213                                  get_partition_uuid=lambda path: uuid):
 214             assert uuid == partition.get_uuid()
 215         assert uuid == partition.get_uuid()
 216
 217     def test_get_ptype(self):
 218         partition = main.DevicePartition(argparse.Namespace())
 219         ptype = main.PTYPE['regular']['osd']['tobe']
 220         with mock.patch.multiple(main,
 221                                  get_partition_type=lambda path: ptype):
 222             assert ptype == partition.get_ptype()
 223         assert ptype == partition.get_ptype()
 224
 225     def test_partition_number(self):
 226         partition = main.DevicePartition(argparse.Namespace())
 227         num = 123
 228         assert num != partition.get_partition_number()
 229         partition.set_partition_number(num)
 230         assert num == partition.get_partition_number()
 231
 232     def test_dev(self):
 233         partition = main.DevicePartition(argparse.Namespace())
 234         dev = '/dev/sdbFOo'
 235         assert dev != partition.get_dev()
 236         assert dev != partition.get_rawdev()
 237         partition.set_dev(dev)
 238         assert dev == partition.get_dev()
 239         assert dev == partition.get_rawdev()
 240
 241     @mock.patch('ceph_disk.main.is_mpath')
 242     def test_factory(self, m_is_mpath):
 243         parser = argparse.ArgumentParser('ceph-disk')
 244         subparsers = parser.add_subparsers()
 245         main.Prepare.set_subparser(subparsers)
 246
 247         path = 'DATA'
 248         m_is_mpath.return_value = False
 249
 250         #
 251         # Device partition
 252         #
 253         args = parser.parse_args([
 254             'prepare',
 255             path,
 256         ])
 257         partition = main.DevicePartition.factory(
 258             path=path, dev=None, args=args)
 259         assert isinstance(partition, main.DevicePartition)
 260
 261         #
 262         # Multipath device partition
 263         #
 264         m_is_mpath.return_value = True
 265         args = parser.parse_args([
 266             'prepare',
 267             path,
 268         ])
 269         partition = main.DevicePartition.factory(
 270             path=path, dev=None, args=args)
 271         assert isinstance(partition, main.DevicePartitionMultipath)
 272         m_is_mpath.return_value = False
 273
 274         #
 275         # Device partition encrypted via dmcrypt luks
 276         #
 277         args = parser.parse_args([
 278             'prepare',
 279             '--dmcrypt',
 280             path,
 281         ])
 282         partition = main.DevicePartition.factory(
 283             path=path, dev=None, args=args)
 284         assert isinstance(partition, main.DevicePartitionCryptLuks)
 285
 286         #
 287         # Device partition encrypted via dmcrypt plain
 288         #
 289         self.conf['global']['osd dmcrypt type'] = 'plain'
 290         self.save_conf()
 291         args = parser.parse_args([
 292             'prepare',
 293             '--dmcrypt',
 294             path,
 295         ])
 296         partition = main.DevicePartition.factory(
 297             path=path, dev=None, args=args)
 298         assert isinstance(partition, main.DevicePartitionCryptPlain)
 299
 300
 301 class TestDevicePartitionMultipath(Base):
 302
 303     def test_init(self):
 304         partition = main.DevicePartitionMultipath(argparse.Namespace())
 305         for name in ('osd', 'journal'):
 306             assert (main.PTYPE['mpath'][name]['ready'] ==
 307                     partition.ptype_for_name(name))
 308
 309
 310 class TestDevicePartitionCrypt(Base):
 311
 312     @mock.patch('ceph_disk.main.get_conf')
 313     def test_luks(self, m_get_conf):
 314         parser = argparse.ArgumentParser('ceph-disk')
 315         subparsers = parser.add_subparsers()
 316         main.Prepare.set_subparser(subparsers)
 317         key_size = 256
 318
 319         def get_conf(**kwargs):
 320             if kwargs['variable'] == 'osd_dmcrypt_key_size':
 321                 return key_size
 322             elif kwargs['variable'] == 'osd_dmcrypt_type':
 323                 return 'luks'
 324             elif kwargs['variable'] == 'osd_cryptsetup_parameters':
 325                 return 'PARAMETERS'
 326             else:
 327                 assert 0
 328
 329         m_get_conf.side_effect = get_conf
 330         data = 'DATA'
 331         args = parser.parse_args([
 332             'prepare',
 333             data,
 334             '--dmcrypt',
 335         ])
 336         partition = main.DevicePartitionCryptLuks(args)
 337         assert partition.luks()
 338         assert partition.osd_dm_key is None
 339         uuid = 'UUID'
 340         with mock.patch.multiple(main,
 341                                  _dmcrypt_map=mock.DEFAULT,
 342                                  get_dmcrypt_key=mock.DEFAULT,
 343                                  get_partition_uuid=lambda path: uuid) as m:
 344             partition.map()
 345             assert m['_dmcrypt_map'].called
 346             m['get_dmcrypt_key'].assert_called_with(
 347                 uuid, '/etc/ceph/dmcrypt-keys', True)
 348
 349
 350 class TestCryptHelpers(Base):
 351
 352     @mock.patch('ceph_disk.main.get_conf')
 353     def test_get_dmcrypt_type(self, m_get_conf):
 354         args = argparse.Namespace(dmcrypt=False)
 355         assert main.CryptHelpers.get_dmcrypt_type(args) is None
 356
 357         m_get_conf.return_value = 'luks'
 358         args = argparse.Namespace(dmcrypt=True, cluster='ceph')
 359         assert main.CryptHelpers.get_dmcrypt_type(args) is 'luks'
 360
 361         m_get_conf.return_value = None
 362         args = argparse.Namespace(dmcrypt=True, cluster='ceph')
 363         assert main.CryptHelpers.get_dmcrypt_type(args) is 'luks'
 364
 365         m_get_conf.return_value = 'plain'
 366         args = argparse.Namespace(dmcrypt=True, cluster='ceph')
 367         assert main.CryptHelpers.get_dmcrypt_type(args) is 'plain'
 368
 369         invalid = 'INVALID'
 370         m_get_conf.return_value = invalid
 371         args = argparse.Namespace(dmcrypt=True, cluster='ceph')
 372         with pytest.raises(main.Error) as err:
 373             main.CryptHelpers.get_dmcrypt_type(args)
 374         assert invalid in str(err)
 375
 376
 377 class TestPrepareData(Base):
 378
 379     def test_set_variables(self):
 380         parser = argparse.ArgumentParser('ceph-disk')
 381         subparsers = parser.add_subparsers()
 382         main.Prepare.set_subparser(subparsers)
 383
 384         osd_uuid = 'OSD_UUID'
 385         cluster_uuid = '571bb920-6d85-44d7-9eca-1bc114d1cd75'
 386         data = 'data'
 387         args = parser.parse_args([
 388             'prepare',
 389             '--osd-uuid', osd_uuid,
 390             '--cluster-uuid', cluster_uuid,
 391             data,
 392         ])
 393
 394         def set_type(self):
 395             self.type = self.FILE
 396         with mock.patch.multiple(main.PrepareData,
 397                                  set_type=set_type):
 398             data = main.PrepareData(args)
 399         assert data.args.osd_uuid == osd_uuid
 400         assert data.args.cluster_uuid == cluster_uuid
 401
 402         data = 'data'
 403         args = parser.parse_args([
 404             'prepare',
 405             data,
 406         ])
 407
 408         with mock.patch.multiple(main.PrepareData,
 409                                  set_type=set_type):
 410             data = main.PrepareData(args)
 411         assert 36 == len(data.args.osd_uuid)
 412         assert 36 == len(data.args.cluster_uuid)
 413
 414         self.conf['global']['fsid'] = cluster_uuid
 415         self.save_conf()
 416         data = 'data'
 417         args = parser.parse_args([
 418             'prepare',
 419             data,
 420         ])
 421
 422         with mock.patch.multiple(main.PrepareData,
 423                                  set_type=set_type):
 424             data = main.PrepareData(args)
 425         assert data.args.cluster_uuid == cluster_uuid
 426
 427
 428 class TestSecrets(Base):
 429
 430     @mock.patch('ceph_disk.main.command')
 431     def test_secrets(self, m_command):
 432         key = "KEY"
 433         m_command.side_effect = lambda cmd: (key + "\n", '', 0)
 434         s = main.Secrets()
 435         assert {"cephx_secret": key} == s.keys
 436         assert '{"cephx_secret": "' + key + '"}' == s.get_json()
 437
 438     @mock.patch('ceph_disk.main.open')
 439     @mock.patch('ceph_disk.main.CryptHelpers.get_dmcrypt_keysize')
 440     @mock.patch('ceph_disk.main.command')
 441     def test_lockbox_secrets(self,
 442                              m_command,
 443                              m_get_dmcrypt_keysize,
 444                              m_open):
 445         key = "KEY"
 446         m_command.side_effect = lambda cmd: (key + "\n", '', 0)
 447         m_get_dmcrypt_keysize.side_effect = lambda args: 32
 448
 449         class File:
 450             def read(self, size):
 451                 return b'O' * size
 452
 453         m_open.side_effect = lambda path, mode: File()
 454         s = main.LockboxSecrets({})
 455         assert {
 456             "dmcrypt_key": 'T09PTw==',
 457             "cephx_secret": key,
 458             "cephx_lockbox_secret": key,
 459         } == s.keys