--- /dev/null
+#!/usr/bin/env python
+#
+# Copyright (C) 2015, 2016 Red Hat <contact@redhat.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Library Public License as published by
+# the Free Software Foundation; either version 2, or (at your option)
+# any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Library Public License for more details.
+#
+import argparse
+import configobj
+import mock
+import os
+import platform
+import pytest
+import shutil
+import tempfile
+
+from ceph_disk import main
+
+
+class Base(object):
+
+ def setup_class(self):
+ main.setup_logging(True, False)
+ os.environ['PATH'] = "..:" + os.environ['PATH']
+
+ def setup(self):
+ _, self.conf_file = tempfile.mkstemp()
+ os.environ['CEPH_CONF'] = self.conf_file
+ self.conf = configobj.ConfigObj(self.conf_file)
+ self.conf['global'] = {}
+
+ def teardown(self):
+ os.unlink(self.conf_file)
+
+ def save_conf(self):
+ self.conf.write(open(self.conf_file, 'wb'))
+
+
+class TestPrepare(Base):
+
+ def test_init_filestore_dir(self):
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+
+ data = tempfile.mkdtemp()
+ main.setup_statedir(data)
+ args = parser.parse_args([
+ 'prepare',
+ data,
+ '--filestore',
+ ])
+
+ def set_type(self):
+ self.type = self.FILE
+ with mock.patch.multiple(main.PrepareData,
+ set_type=set_type):
+ prepare = main.Prepare.factory(args)
+ assert isinstance(prepare.data, main.PrepareFilestoreData)
+ assert prepare.data.is_file()
+ assert isinstance(prepare.journal, main.PrepareJournal)
+ assert prepare.journal.is_none()
+ prepare.prepare()
+ assert os.path.exists(os.path.join(data, 'fsid'))
+ shutil.rmtree(data)
+
+ @mock.patch('stat.S_ISBLK')
+ @mock.patch('ceph_disk.main.is_partition')
+ def test_init_filestore_dev(self, m_is_partition, m_s_isblk):
+ m_s_isblk.return_value = True
+
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+
+ m_is_partition.return_value = False
+ _, data = tempfile.mkstemp()
+
+ args = parser.parse_args([
+ 'prepare',
+ data,
+ '--filestore',
+ ])
+ prepare = main.Prepare.factory(args)
+ assert isinstance(prepare.data, main.PrepareData)
+ assert prepare.data.is_device()
+ assert isinstance(prepare.journal, main.PrepareJournal)
+ assert prepare.journal.is_device()
+
+ def test_init_default_dir(self):
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+
+ data = tempfile.mkdtemp()
+ main.setup_statedir(data)
+ args = parser.parse_args([
+ 'prepare',
+ data,
+ ])
+
+ def set_type(self):
+ self.type = self.FILE
+ with mock.patch.multiple(main.PrepareData,
+ set_type=set_type):
+ prepare = main.Prepare.factory(args)
+ assert isinstance(prepare.data, main.PrepareBluestoreData)
+ assert prepare.data.is_file()
+ prepare.prepare()
+ assert os.path.exists(os.path.join(data, 'fsid'))
+ shutil.rmtree(data)
+
+ def test_set_subparser(self):
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+ osd_uuid = 'OSD_UUID'
+ journal_uuid = 'JOURNAL_UUID'
+ fs_type = 'xfs'
+ data = 'DATA'
+ args = parser.parse_args([
+ 'prepare',
+ '--osd-uuid', osd_uuid,
+ '--journal-uuid', journal_uuid,
+ '--fs-type', fs_type,
+ data,
+ ])
+ assert args.journal_uuid == journal_uuid
+ assert args.osd_uuid == osd_uuid
+ assert args.fs_type == fs_type
+ assert args.data == data
+
+
+class TestDevice(Base):
+
+ @mock.patch('ceph_disk.main.is_partition')
+ def test_init(self, m_is_partition):
+ m_is_partition.return_value = False
+ device = main.Device('/dev/wholedisk', argparse.Namespace())
+ assert device.dev_size is None
+
+ @mock.patch('ceph_disk.main.is_partition')
+ @mock.patch('ceph_disk.main.get_free_partition_index')
+ @mock.patch('ceph_disk.main.update_partition')
+ @mock.patch('ceph_disk.main.get_dm_uuid')
+ @mock.patch('ceph_disk.main.get_dev_size')
+ @mock.patch('ceph_disk.main.command_check_call')
+ def test_create_partition(self,
+ m_command_check_call,
+ m_get_dev_size,
+ m_get_dm_uuid,
+ m_update_partition,
+ m_get_free_partition_index,
+ m_is_partition):
+ if platform.system() == 'FreeBSD':
+ return
+ m_is_partition.return_value = False
+ partition_number = 1
+ m_get_free_partition_index.return_value = partition_number
+ path = '/dev/wholedisk'
+ device = main.Device(path, argparse.Namespace(dmcrypt=False))
+ uuid = 'UUID'
+ m_get_dm_uuid.return_value = uuid
+ size = 200
+ m_get_dev_size.return_value = size + 100
+ name = 'journal'
+ actual_partition_number = device.create_partition(
+ uuid=uuid, name=name, size=size)
+ assert actual_partition_number == partition_number
+ command = ['sgdisk',
+ '--new=%d:0:+%dM' % (partition_number, size),
+ '--change-name=%d:ceph %s' % (partition_number, name),
+ '--partition-guid=%d:%s' % (partition_number, uuid),
+ '--typecode=%d:%s' % (
+ partition_number,
+ main.PTYPE['regular']['journal']['ready']),
+ '--mbrtogpt', '--', path]
+ m_command_check_call.assert_called_with(command, exit=True)
+ m_update_partition.assert_called_with(path, 'created')
+
+ actual_partition_number = device.create_partition(
+ uuid=uuid, name=name)
+ command = ['sgdisk',
+ '--largest-new=%d' % partition_number,
+ '--change-name=%d:ceph %s' % (partition_number, name),
+ '--partition-guid=%d:%s' % (partition_number, uuid),
+ '--typecode=%d:%s' % (
+ partition_number,
+ main.PTYPE['regular']['journal']['ready']),
+ '--mbrtogpt', '--', path]
+ m_command_check_call.assert_called_with(command, exit=True)
+
+
+class TestDevicePartition(Base):
+
+ def test_init(self):
+ partition = main.DevicePartition(argparse.Namespace())
+ for name in ('osd', 'journal'):
+ assert (main.PTYPE['regular'][name]['ready'] ==
+ partition.ptype_for_name(name))
+
+ def test_get_uuid(self):
+ partition = main.DevicePartition(argparse.Namespace())
+ uuid = 'UUID'
+ with mock.patch.multiple(main,
+ get_partition_uuid=lambda path: uuid):
+ assert uuid == partition.get_uuid()
+ assert uuid == partition.get_uuid()
+
+ def test_get_ptype(self):
+ partition = main.DevicePartition(argparse.Namespace())
+ ptype = main.PTYPE['regular']['osd']['tobe']
+ with mock.patch.multiple(main,
+ get_partition_type=lambda path: ptype):
+ assert ptype == partition.get_ptype()
+ assert ptype == partition.get_ptype()
+
+ def test_partition_number(self):
+ partition = main.DevicePartition(argparse.Namespace())
+ num = 123
+ assert num != partition.get_partition_number()
+ partition.set_partition_number(num)
+ assert num == partition.get_partition_number()
+
+ def test_dev(self):
+ partition = main.DevicePartition(argparse.Namespace())
+ dev = '/dev/sdbFOo'
+ assert dev != partition.get_dev()
+ assert dev != partition.get_rawdev()
+ partition.set_dev(dev)
+ assert dev == partition.get_dev()
+ assert dev == partition.get_rawdev()
+
+ @mock.patch('ceph_disk.main.is_mpath')
+ def test_factory(self, m_is_mpath):
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+
+ path = 'DATA'
+ m_is_mpath.return_value = False
+
+ #
+ # Device partition
+ #
+ args = parser.parse_args([
+ 'prepare',
+ path,
+ ])
+ partition = main.DevicePartition.factory(
+ path=path, dev=None, args=args)
+ assert isinstance(partition, main.DevicePartition)
+
+ #
+ # Multipath device partition
+ #
+ m_is_mpath.return_value = True
+ args = parser.parse_args([
+ 'prepare',
+ path,
+ ])
+ partition = main.DevicePartition.factory(
+ path=path, dev=None, args=args)
+ assert isinstance(partition, main.DevicePartitionMultipath)
+ m_is_mpath.return_value = False
+
+ #
+ # Device partition encrypted via dmcrypt luks
+ #
+ args = parser.parse_args([
+ 'prepare',
+ '--dmcrypt',
+ path,
+ ])
+ partition = main.DevicePartition.factory(
+ path=path, dev=None, args=args)
+ assert isinstance(partition, main.DevicePartitionCryptLuks)
+
+ #
+ # Device partition encrypted via dmcrypt plain
+ #
+ self.conf['global']['osd dmcrypt type'] = 'plain'
+ self.save_conf()
+ args = parser.parse_args([
+ 'prepare',
+ '--dmcrypt',
+ path,
+ ])
+ partition = main.DevicePartition.factory(
+ path=path, dev=None, args=args)
+ assert isinstance(partition, main.DevicePartitionCryptPlain)
+
+
+class TestDevicePartitionMultipath(Base):
+
+ def test_init(self):
+ partition = main.DevicePartitionMultipath(argparse.Namespace())
+ for name in ('osd', 'journal'):
+ assert (main.PTYPE['mpath'][name]['ready'] ==
+ partition.ptype_for_name(name))
+
+
+class TestDevicePartitionCrypt(Base):
+
+ @mock.patch('ceph_disk.main.get_conf')
+ def test_luks(self, m_get_conf):
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+ key_size = 256
+
+ def get_conf(**kwargs):
+ if kwargs['variable'] == 'osd_dmcrypt_key_size':
+ return key_size
+ elif kwargs['variable'] == 'osd_dmcrypt_type':
+ return 'luks'
+ elif kwargs['variable'] == 'osd_cryptsetup_parameters':
+ return 'PARAMETERS'
+ else:
+ assert 0
+
+ m_get_conf.side_effect = get_conf
+ data = 'DATA'
+ args = parser.parse_args([
+ 'prepare',
+ data,
+ '--dmcrypt',
+ ])
+ partition = main.DevicePartitionCryptLuks(args)
+ assert partition.luks()
+ assert partition.osd_dm_key is None
+ uuid = 'UUID'
+ with mock.patch.multiple(main,
+ _dmcrypt_map=mock.DEFAULT,
+ get_dmcrypt_key=mock.DEFAULT,
+ get_partition_uuid=lambda path: uuid) as m:
+ partition.map()
+ assert m['_dmcrypt_map'].called
+ m['get_dmcrypt_key'].assert_called_with(
+ uuid, '/etc/ceph/dmcrypt-keys', True)
+
+
+class TestCryptHelpers(Base):
+
+ @mock.patch('ceph_disk.main.get_conf')
+ def test_get_dmcrypt_type(self, m_get_conf):
+ args = argparse.Namespace(dmcrypt=False)
+ assert main.CryptHelpers.get_dmcrypt_type(args) is None
+
+ m_get_conf.return_value = 'luks'
+ args = argparse.Namespace(dmcrypt=True, cluster='ceph')
+ assert main.CryptHelpers.get_dmcrypt_type(args) is 'luks'
+
+ m_get_conf.return_value = None
+ args = argparse.Namespace(dmcrypt=True, cluster='ceph')
+ assert main.CryptHelpers.get_dmcrypt_type(args) is 'luks'
+
+ m_get_conf.return_value = 'plain'
+ args = argparse.Namespace(dmcrypt=True, cluster='ceph')
+ assert main.CryptHelpers.get_dmcrypt_type(args) is 'plain'
+
+ invalid = 'INVALID'
+ m_get_conf.return_value = invalid
+ args = argparse.Namespace(dmcrypt=True, cluster='ceph')
+ with pytest.raises(main.Error) as err:
+ main.CryptHelpers.get_dmcrypt_type(args)
+ assert invalid in str(err)
+
+
+class TestPrepareData(Base):
+
+ def test_set_variables(self):
+ parser = argparse.ArgumentParser('ceph-disk')
+ subparsers = parser.add_subparsers()
+ main.Prepare.set_subparser(subparsers)
+
+ osd_uuid = 'OSD_UUID'
+ cluster_uuid = '571bb920-6d85-44d7-9eca-1bc114d1cd75'
+ data = 'data'
+ args = parser.parse_args([
+ 'prepare',
+ '--osd-uuid', osd_uuid,
+ '--cluster-uuid', cluster_uuid,
+ data,
+ ])
+
+ def set_type(self):
+ self.type = self.FILE
+ with mock.patch.multiple(main.PrepareData,
+ set_type=set_type):
+ data = main.PrepareData(args)
+ assert data.args.osd_uuid == osd_uuid
+ assert data.args.cluster_uuid == cluster_uuid
+
+ data = 'data'
+ args = parser.parse_args([
+ 'prepare',
+ data,
+ ])
+
+ with mock.patch.multiple(main.PrepareData,
+ set_type=set_type):
+ data = main.PrepareData(args)
+ assert 36 == len(data.args.osd_uuid)
+ assert 36 == len(data.args.cluster_uuid)
+
+ self.conf['global']['fsid'] = cluster_uuid
+ self.save_conf()
+ data = 'data'
+ args = parser.parse_args([
+ 'prepare',
+ data,
+ ])
+
+ with mock.patch.multiple(main.PrepareData,
+ set_type=set_type):
+ data = main.PrepareData(args)
+ assert data.args.cluster_uuid == cluster_uuid
+
+
+class TestSecrets(Base):
+
+ @mock.patch('ceph_disk.main.command')
+ def test_secrets(self, m_command):
+ key = "KEY"
+ m_command.side_effect = lambda cmd: (key + "\n", '', 0)
+ s = main.Secrets()
+ assert {"cephx_secret": key} == s.keys
+ assert '{"cephx_secret": "' + key + '"}' == s.get_json()
+
+ @mock.patch('ceph_disk.main.open')
+ @mock.patch('ceph_disk.main.CryptHelpers.get_dmcrypt_keysize')
+ @mock.patch('ceph_disk.main.command')
+ def test_lockbox_secrets(self,
+ m_command,
+ m_get_dmcrypt_keysize,
+ m_open):
+ key = "KEY"
+ m_command.side_effect = lambda cmd: (key + "\n", '', 0)
+ m_get_dmcrypt_keysize.side_effect = lambda args: 32
+
+ class File:
+ def read(self, size):
+ return b'O' * size
+
+ m_open.side_effect = lambda path, mode: File()
+ s = main.LockboxSecrets({})
+ assert {
+ "dmcrypt_key": 'T09PTw==',
+ "cephx_secret": key,
+ "cephx_lockbox_secret": key,
+ } == s.keys
Code Review / stor4nfv.git / blobdiff