ceph-ansible/tests/library/test_ceph_volume.py

import sys
import mock
import os
import pytest
import ca_test_common
sys.path.append('./library')
import ceph_volume  # noqa: E402


# Python 3
try:
    from unittest.mock import MagicMock, patch
except ImportError:
    # Python 2
    try:
        from mock import MagicMock, patch
    except ImportError:
        print('You need the mock library installed on python2.x to run tests')


def get_mounts(mounts=None):
    volumes = {}
    volumes['/run/lock/lvm'] = '/run/lock/lvm:z'
    volumes['/var/run/udev'] = '/var/run/udev:z'
    volumes['/dev'] = '/dev'
    volumes['/etc/ceph'] = '/etc/ceph:z'
    volumes['/run/lvm'] = '/run/lvm'
    volumes['/var/lib/ceph'] = '/var/lib/ceph:z'
    volumes['/var/log/ceph'] = '/var/log/ceph:z'
    if mounts is not None:
        volumes.update(mounts)

    return sum([['-v', '{}:{}'.format(src_dir, dst_dir)] for src_dir, dst_dir in volumes.items()], [])


def get_container_cmd(mounts=None):

    return ['docker', 'run', '--rm', '--privileged',
            '--net=host', '--ipc=host'] + \
            get_mounts(mounts) + ['--entrypoint=ceph-volume']


@mock.patch.dict(os.environ, {'CEPH_CONTAINER_BINARY': 'docker'})
class TestCephVolumeModule(object):

    def test_data_no_vg(self):
        result = ceph_volume.get_data("/dev/sda", None)
        assert result == "/dev/sda"

    def test_data_with_vg(self):
        result = ceph_volume.get_data("data-lv", "data-vg")
        assert result == "data-vg/data-lv"

    def test_journal_no_vg(self):
        result = ceph_volume.get_journal("/dev/sda1", None)
        assert result == "/dev/sda1"

    def test_journal_with_vg(self):
        result = ceph_volume.get_journal("journal-lv", "journal-vg")
        assert result == "journal-vg/journal-lv"

    def test_db_no_vg(self):
        result = ceph_volume.get_db("/dev/sda1", None)
        assert result == "/dev/sda1"

    def test_db_with_vg(self):
        result = ceph_volume.get_db("db-lv", "db-vg")
        assert result == "db-vg/db-lv"

    def test_wal_no_vg(self):
        result = ceph_volume.get_wal("/dev/sda1", None)
        assert result == "/dev/sda1"

    def test_wal_with_vg(self):
        result = ceph_volume.get_wal("wal-lv", "wal-vg")
        assert result == "wal-vg/wal-lv"

    def test_container_exec(self):
        fake_binary = "ceph-volume"
        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd() + [fake_container_image]
        result = ceph_volume.container_exec(fake_binary, fake_container_image)
        assert result == expected_command_list

    def test_zap_osd_container(self):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda'}
        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd() + \
            [fake_container_image,
                '--cluster',
                'ceph',
                'lvm',
                'zap',
                '--destroy',
                '/dev/sda']
        result = ceph_volume.zap_devices(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_zap_osd(self):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda'}
        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'zap',
                                 '--destroy',
                                 '/dev/sda']
        result = ceph_volume.zap_devices(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_zap_osd_fsid(self):
        fake_module = MagicMock()
        fake_module.params = {'osd_fsid': 'a_uuid'}
        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'zap',
                                 '--destroy',
                                 '--osd-fsid',
                                 'a_uuid']
        result = ceph_volume.zap_devices(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_zap_osd_id(self):
        fake_module = MagicMock()
        fake_module.params = {'osd_id': '123'}
        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'zap',
                                 '--destroy',
                                 '--osd-id',
                                 '123']
        result = ceph_volume.zap_devices(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_activate_osd(self):
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'activate',
                                 '--all']
        result = ceph_volume.activate_osd()
        assert result == expected_command_list

    def test_list_osd(self):
        fake_module = MagicMock()
        fake_module.params = {'cluster': 'ceph', 'data': '/dev/sda'}
        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'list',
                                 '/dev/sda',
                                 '--format=json']
        result = ceph_volume.list_osd(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_list_osd_container(self):
        fake_module = MagicMock()
        fake_module.params = {'cluster': 'ceph', 'data': '/dev/sda'}
        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd(
                                {
                                    '/var/lib/ceph': '/var/lib/ceph:ro'
                                }) + \
            [fake_container_image,
                '--cluster',
                'ceph',
                'lvm',
                'list',
                '/dev/sda',
                '--format=json']
        result = ceph_volume.list_osd(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_list_storage_inventory(self):
        fake_module = MagicMock()
        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'inventory',
                                 '--format=json',
                                 ]
        result = ceph_volume.list_storage_inventory(fake_module, fake_container_image)
        assert result == expected_command_list

    def test_list_storage_inventory_container(self):
        fake_module = MagicMock()
        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd() + \
            [fake_container_image,
                '--cluster',
                'ceph',
                'inventory',
                '--format=json']
        result = ceph_volume.list_storage_inventory(fake_module, fake_container_image)
        assert result == expected_command_list

    @pytest.mark.parametrize('objectstore', ['bluestore'])
    def test_create_osd_container(self, objectstore):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda',
                              'objectstore': objectstore,
                              'cluster': 'ceph', }

        fake_action = "create"
        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd() + \
            [fake_container_image,
                '--cluster',
                'ceph',
                'lvm',
                'create',
                '--%s' % objectstore,
                '--data',
                '/dev/sda']
        result = ceph_volume.prepare_or_create_osd(
            fake_module, fake_action, fake_container_image)
        assert result == expected_command_list

    @pytest.mark.parametrize('objectstore', ['bluestore'])
    def test_create_osd(self, objectstore):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda',
                              'objectstore': objectstore,
                              'cluster': 'ceph', }

        fake_container_image = None
        fake_action = "create"
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'create',
                                 '--%s' % objectstore,
                                 '--data',
                                 '/dev/sda']
        result = ceph_volume.prepare_or_create_osd(
            fake_module, fake_action, fake_container_image)
        assert result == expected_command_list

    @pytest.mark.parametrize('objectstore', ['bluestore'])
    def test_prepare_osd_container(self, objectstore):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda',
                              'objectstore': objectstore,
                              'cluster': 'ceph', }

        fake_action = "prepare"
        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd() + \
            [fake_container_image,
                '--cluster',
                'ceph',
                'lvm',
                'prepare',
                '--%s' % objectstore,
                '--data',
                '/dev/sda']
        result = ceph_volume.prepare_or_create_osd(
            fake_module, fake_action, fake_container_image)
        assert result == expected_command_list

    @pytest.mark.parametrize('objectstore', ['bluestore'])
    def test_prepare_osd(self, objectstore):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda',
                              'objectstore': objectstore,
                              'cluster': 'ceph', }

        fake_container_image = None
        fake_action = "prepare"
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'prepare',
                                 '--%s' % objectstore,
                                 '--data',
                                 '/dev/sda']
        result = ceph_volume.prepare_or_create_osd(
            fake_module, fake_action, fake_container_image)
        assert result == expected_command_list

    @pytest.mark.parametrize('objectstore', ['bluestore'])
    def test_batch_osd_container(self, objectstore):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda',
                              'objectstore': objectstore,
                              'block_db_size': '4096',
                              'journal_size': '4096',
                              'cluster': 'ceph',
                              'batch_devices': ["/dev/sda", "/dev/sdb"]}

        fake_container_image = "quay.io/ceph/daemon:latest"
        expected_command_list = get_container_cmd() + \
            [fake_container_image,
                '--cluster',
                'ceph',
                'lvm',
                'batch',
                '--%s' % objectstore,
                '--yes',
                '--prepare',
                '--block-db-size',
                '4096',
                '/dev/sda',
                '/dev/sdb']
        result = ceph_volume.batch(
            fake_module, fake_container_image)
        assert result == expected_command_list

    @pytest.mark.parametrize('objectstore', ['bluestore'])
    def test_batch_osd(self, objectstore):
        fake_module = MagicMock()
        fake_module.params = {'data': '/dev/sda',
                              'objectstore': objectstore,
                              'block_db_size': '4096',
                              'journal_size': '4096',
                              'cluster': 'ceph',
                              'batch_devices': ["/dev/sda", "/dev/sdb"]}

        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'batch',
                                 '--%s' % objectstore,
                                 '--yes',
                                 '--block-db-size',
                                 '4096',
                                 '/dev/sda',
                                 '/dev/sdb']
        result = ceph_volume.batch(
            fake_module, fake_container_image)
        assert result == expected_command_list

    def test_batch_bluestore_with_dedicated_db(self):
        fake_module = MagicMock()
        fake_module.params = {'objectstore': 'bluestore',
                              'block_db_size': '-1',
                              'cluster': 'ceph',
                              'batch_devices': ["/dev/sda", "/dev/sdb"],
                              'block_db_devices': ["/dev/sdc", "/dev/sdd"]}

        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'batch',
                                 '--bluestore',
                                 '--yes',
                                 '/dev/sda',
                                 '/dev/sdb',
                                 '--db-devices',
                                 '/dev/sdc',
                                 '/dev/sdd']
        result = ceph_volume.batch(
            fake_module, fake_container_image)
        assert result == expected_command_list

    def test_batch_bluestore_with_dedicated_wal(self):
        fake_module = MagicMock()
        fake_module.params = {'objectstore': 'bluestore',
                              'cluster': 'ceph',
                              'block_db_size': '-1',
                              'batch_devices': ["/dev/sda", "/dev/sdb"],
                              'wal_devices': ["/dev/sdc", "/dev/sdd"]}

        fake_container_image = None
        expected_command_list = ['ceph-volume',
                                 '--cluster',
                                 'ceph',
                                 'lvm',
                                 'batch',
                                 '--bluestore',
                                 '--yes',
                                 '/dev/sda',
                                 '/dev/sdb',
                                 '--wal-devices',
                                 '/dev/sdc',
                                 '/dev/sdd']
        result = ceph_volume.batch(
            fake_module, fake_container_image)
        assert result == expected_command_list

    @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
    @patch('ansible.module_utils.basic.AnsibleModule.run_command')
    def test_prepare_no_keyring_in_output(self, m_run_command, m_exit_json):
        ca_test_common.set_module_args({'data': '/dev/sda',
                                        'objectstore': 'bluestore',
                                        'cluster': 'ceph',
                                        'action': 'prepare'})
        keyring = 'AQBqkhNhQDlqEhAAXKxu87L3Mh3mHY+agonKZA=='
        m_exit_json.side_effect = ca_test_common.exit_json
        list_rc = 0
        list_stderr = ''
        list_stdout = '{}'
        prepare_rc = 0
        prepare_stderr = """
    Running command: /usr/bin/ceph-authtool --gen-print-key
    Running command: /usr/bin/mount -t tmpfs tmpfs /var/lib/ceph/osd/ceph-0
    Running command: /usr/bin/chown -h ceph:ceph /dev/test_group/data-lv1
    Running command: /usr/bin/chown -R ceph:ceph /dev/dm-0
    Running command: /usr/bin/ln -s /dev/test_group/data-lv1 /var/lib/ceph/osd/ceph-1/block
     stderr: got monmap epoch 1
    Running command: /usr/bin/ceph-authtool /var/lib/ceph/osd/ceph-1/keyring --create-keyring --name osd.1 --add-key {}
     stdout: creating /var/lib/ceph/osd/ceph-1/keyring
    added entity osd.1 auth(key={})
""".format(keyring, keyring)
        prepare_stdout = ''
        m_run_command.side_effect = [
            (list_rc, list_stdout, list_stderr),
            (prepare_rc, prepare_stdout, prepare_stderr)
        ]

        with pytest.raises(ca_test_common.AnsibleExitJson) as result:
            ceph_volume.main()

        result = result.value.args[0]
        assert result['changed']
        assert result['cmd'] == ['ceph-volume', '--cluster', 'ceph', 'lvm', 'prepare', '--bluestore', '--data', '/dev/sda']
        assert result['rc'] == 0
        assert keyring not in result['stderr']
        assert '*' * 8 in result['stderr']
        assert not result['stdout']

    @patch('ansible.module_utils.basic.AnsibleModule.exit_json')
    @patch('ansible.module_utils.basic.AnsibleModule.run_command')
    def test_batch_no_keyring_in_output(self, m_run_command, m_exit_json):
        ca_test_common.set_module_args({'batch_devices': ['/dev/sda'],
                                        'objectstore': 'bluestore',
                                        'cluster': 'ceph',
                                        'action': 'batch'})
        keyring = 'AQBUixJhnDF1NRAAhl2xrnmOHCCI/T+W6FjqmA=='
        m_exit_json.side_effect = ca_test_common.exit_json
        report_rc = 0
        report_stderr = ''
        report_stdout = '[{"data": "/dev/sda", "data_size": "50.00 GB", "encryption": "None"}]'
        batch_rc = 0
        batch_stderr = """
    Running command: /usr/bin/ceph-authtool --gen-print-key
    Running command: /usr/bin/mount -t tmpfs tmpfs /var/lib/ceph/osd/ceph-0
    Running command: /usr/bin/chown -h ceph:ceph /dev/ceph-863337c4-bef9-4b96-aaac-27cde8c42b8f/osd-block-b1d1036f-0d6e-493b-9d1a-6f6b96df64b1
    Running command: /usr/bin/chown -R ceph:ceph /dev/mapper/ceph--863337c4--bef9--4b96--aaac--27cde8c42b8f-osd--block--b1d1036f--0d6e--493b--9d1a--6f6b96df64b1
    Running command: /usr/bin/ln -s /dev/ceph-863337c4-bef9-4b96-aaac-27cde8c42b8f/osd-block-b1d1036f-0d6e-493b-9d1a-6f6b96df64b1 /var/lib/ceph/osd/ceph-0/block
     stderr: got monmap epoch 1
    Running command: /usr/bin/ceph-authtool /var/lib/ceph/osd/ceph-0/keyring --create-keyring --name osd.0 --add-key {}
     stdout: creating /var/lib/ceph/osd/ceph-0/keyring
    added entity osd.0 auth(key={})
""".format(keyring, keyring)
        batch_stdout = ''
        m_run_command.side_effect = [
            (report_rc, report_stdout, report_stderr),
            (batch_rc, batch_stdout, batch_stderr)
        ]

        with pytest.raises(ca_test_common.AnsibleExitJson) as result:
            ceph_volume.main()

        result = result.value.args[0]
        assert result['changed']
        assert result['cmd'] == ['ceph-volume', '--cluster', 'ceph', 'lvm', 'batch', '--bluestore', '--yes', '/dev/sda']
        assert result['rc'] == 0
        assert keyring not in result['stderr']
        assert '*' * 8 in result['stderr']
        assert not result['stdout']