salt/tests/unit/modules/test_archive.py

# -*- coding: utf-8 -*-
'''
    :codeauthor: :email:`Pedro Algarvio (pedro@algarvio.me)`


    tests.unit.modules.archive_test
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
'''

# Import Python libs
from __future__ import absolute_import
import os.path

# Import Salt Testing libs
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.unit import skipIf, TestCase
from tests.support.mock import NO_MOCK, NO_MOCK_REASON, MagicMock, patch

# Import salt libs
import salt.modules.archive as archive
from salt.exceptions import CommandNotFoundError
from salt.utils import which_bin

# Import 3rd-party libs
from salt.ext.six.moves import zip


class ZipFileMock(MagicMock):
    def __init__(self, files=None, **kwargs):  # pylint: disable=W0231
        if files is None:
            files = ['salt']
        MagicMock.__init__(self, **kwargs)
        self._files = files
        self.external_attr = 0o0777 << 16

    def namelist(self):
        return self._files


@skipIf(NO_MOCK, NO_MOCK_REASON)
class ArchiveTestCase(TestCase, LoaderModuleMockMixin):

    def setup_loader_modules(self):
        return {archive: {'__grains__': {'id': 0}}}

    def test_tar(self):
        with patch('glob.glob', lambda pathname: [pathname]):
            with patch('salt.utils.which', lambda exe: exe):
                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.tar(
                        '-zcvf', 'foo.tar',
                        ['/tmp/something-to-compress-1',
                         '/tmp/something-to-compress-2'],
                        template=None
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['tar', '-zcvf', 'foo.tar', '/tmp/something-to-compress-1',
                         '/tmp/something-to-compress-2'],
                        runas=None, python_shell=False, template=None, cwd=None
                    )

                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.tar(
                        '-zcvf', 'foo.tar',
                        '/tmp/something-to-compress-1,/tmp/something-to-compress-2',
                        template=None
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['tar', '-zcvf', 'foo.tar', '/tmp/something-to-compress-1',
                         '/tmp/something-to-compress-2'],
                        runas=None, python_shell=False, template=None, cwd=None
                    )

    def test_tar_raises_exception_if_not_found(self):
        with patch('salt.utils.which', lambda exe: None):
            mock = MagicMock(return_value='salt')
            with patch.dict(archive.__salt__, {'cmd.run': mock}):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.tar,
                    'zxvf',
                    'foo.tar',
                    '/tmp/something-to-compress'
                )
                self.assertFalse(mock.called)

    def test_gzip(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: exe):
                ret = archive.gzip('/tmp/something-to-compress')
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['gzip', '/tmp/something-to-compress'],
                    runas=None, python_shell=False, template=None
                )

    def test_gzip_raises_exception_if_not_found(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: None):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.gzip, '/tmp/something-to-compress'
                )
                self.assertFalse(mock.called)

    def test_gunzip(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: exe):
                ret = archive.gunzip('/tmp/something-to-decompress.tar.gz')
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['gunzip', '/tmp/something-to-decompress.tar.gz'],
                    runas=None, python_shell=False, template=None
                )

    def test_gunzip_raises_exception_if_not_found(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: None):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.gunzip,
                    '/tmp/something-to-decompress.tar.gz'
                )
                self.assertFalse(mock.called)

    def test_cmd_zip(self):
        with patch('glob.glob', lambda pathname: [pathname]):
            with patch('salt.utils.which', lambda exe: exe):
                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.cmd_zip(
                        '/tmp/salt.{{grains.id}}.zip',
                        '/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                        template='jinja'
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['zip', '-r', '/tmp/salt.{{grains.id}}.zip',
                         '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                         runas=None, python_shell=False, template='jinja', cwd=None
                    )

                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.cmd_zip(
                        '/tmp/salt.{{grains.id}}.zip',
                        ['/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                        template='jinja'
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['zip', '-r', '/tmp/salt.{{grains.id}}.zip',
                         '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                        runas=None, python_shell=False, template='jinja', cwd=None
                    )

    def test_zip(self):
        with patch('glob.glob', lambda pathname: [pathname]):
            with patch.multiple(os.path, **{'isdir': MagicMock(return_value=False),
                                            'exists': MagicMock(return_value=True)}):
                with patch('zipfile.ZipFile', MagicMock()):
                    ret = archive.zip_(
                        '/tmp/salt.{{grains.id}}.zip',
                        '/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                        template='jinja'
                    )
                    expected = [os.path.join('tmp', 'tmpePe8yO'),
                                os.path.join('tmp', 'tmpLeSw1A')]
                    self.assertEqual(expected, ret)

    def test_zip_raises_exception_if_not_found(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: None):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.cmd_zip,
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                    template='jinja',
                )
                self.assertFalse(mock.called)

    def test_cmd_unzip(self):
        def _get_mock():
            '''
            Create a new MagicMock for each scenario in this test, so that
            assert_called_once_with doesn't complain that the same mock object
            is called more than once.
            '''
            return MagicMock(return_value={'stdout': 'salt',
                                           'stderr': '',
                                           'pid': 12345,
                                           'retcode': 0})

        with patch('salt.utils.which', lambda exe: exe):

            mock = _get_mock()
            with patch.dict(archive.__salt__, {'cmd.run_all': mock}):
                ret = archive.cmd_unzip(
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/dest',
                    excludes='/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                    template='jinja'
                )
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['unzip', '/tmp/salt.{{grains.id}}.zip', '-d', '/tmp/dest',
                     '-x', '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    output_loglevel='debug',
                    python_shell=False,
                    redirect_stderr=True,
                    runas=None,
                    template='jinja'
                )

            mock = _get_mock()
            with patch.dict(archive.__salt__, {'cmd.run_all': mock}):
                ret = archive.cmd_unzip(
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/dest',
                    excludes=['/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    template='jinja'
                )
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['unzip', '/tmp/salt.{{grains.id}}.zip', '-d', '/tmp/dest',
                     '-x', '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    output_loglevel='debug',
                    python_shell=False,
                    redirect_stderr=True,
                    runas=None,
                    template='jinja'
                )

            mock = _get_mock()
            with patch.dict(archive.__salt__, {'cmd.run_all': mock}):
                ret = archive.cmd_unzip(
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/dest',
                    excludes='/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                    template='jinja',
                    options='-fo'
                )
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['unzip', '-fo', '/tmp/salt.{{grains.id}}.zip', '-d',
                     '/tmp/dest', '-x', '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    output_loglevel='debug',
                    python_shell=False,
                    redirect_stderr=True,
                    runas=None,
                    template='jinja'
                )

            mock = _get_mock()
            with patch.dict(archive.__salt__, {'cmd.run_all': mock}):
                ret = archive.cmd_unzip(
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/dest',
                    excludes=['/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    template='jinja',
                    options='-fo'
                )
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['unzip', '-fo', '/tmp/salt.{{grains.id}}.zip', '-d',
                     '/tmp/dest', '-x', '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    output_loglevel='debug',
                    python_shell=False,
                    redirect_stderr=True,
                    runas=None,
                    template='jinja'
                )

            mock = _get_mock()
            with patch.dict(archive.__salt__, {'cmd.run_all': mock}):
                ret = archive.cmd_unzip(
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/dest',
                    excludes=['/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    template='jinja',
                    options='-fo',
                    password='asdf',
                )
                self.assertEqual(['salt'], ret)
                mock.assert_called_once_with(
                    ['unzip', '-P', 'asdf', '-fo', '/tmp/salt.{{grains.id}}.zip',
                     '-d', '/tmp/dest', '-x', '/tmp/tmpePe8yO', '/tmp/tmpLeSw1A'],
                    output_loglevel='quiet',
                    python_shell=False,
                    redirect_stderr=True,
                    runas=None,
                    template='jinja'
                )

    def test_unzip(self):
        mock = ZipFileMock()
        with patch('zipfile.ZipFile', mock):
            ret = archive.unzip(
                '/tmp/salt.{{grains.id}}.zip',
                '/tmp/dest',
                excludes='/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                template='jinja',
                extract_perms=False
            )
            self.assertEqual(['salt'], ret)

    def test_unzip_raises_exception_if_not_found(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: None):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.cmd_unzip,
                    '/tmp/salt.{{grains.id}}.zip',
                    '/tmp/dest',
                    excludes='/tmp/tmpePe8yO,/tmp/tmpLeSw1A',
                    template='jinja',
                )
                self.assertFalse(mock.called)

    def test_rar(self):
        with patch('glob.glob', lambda pathname: [pathname]):
            with patch('salt.utils.which', lambda exe: exe):
                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.rar(
                        '/tmp/rarfile.rar',
                        '/tmp/sourcefile1,/tmp/sourcefile2'
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['rar', 'a', '-idp', '/tmp/rarfile.rar',
                         '/tmp/sourcefile1', '/tmp/sourcefile2'],
                        runas=None, python_shell=False, template=None, cwd=None
                    )

                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.rar(
                        '/tmp/rarfile.rar',
                        ['/tmp/sourcefile1', '/tmp/sourcefile2']
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['rar', 'a', '-idp', '/tmp/rarfile.rar',
                         '/tmp/sourcefile1', '/tmp/sourcefile2'],
                        runas=None, python_shell=False, template=None, cwd=None
                    )

    def test_rar_raises_exception_if_not_found(self):
        mock = MagicMock(return_value='salt')
        with patch.dict(archive.__salt__, {'cmd.run': mock}):
            with patch('salt.utils.which', lambda exe: None):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.rar,
                    '/tmp/rarfile.rar',
                    '/tmp/sourcefile1,/tmp/sourcefile2'
                )
                self.assertFalse(mock.called)

    @skipIf(which_bin(('unrar', 'rar')) is None, 'unrar not installed')
    def test_unrar(self):
        with patch('salt.utils.which_bin', lambda exe: exe[0] if isinstance(exe, (list, tuple)) else exe):
            with patch('salt.utils.which', lambda exe: exe[0] if isinstance(exe, (list, tuple)) else exe):
                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.unrar(
                        '/tmp/rarfile.rar',
                        '/home/strongbad/',
                        excludes='file_1,file_2'
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['unrar', 'x', '-idp', '/tmp/rarfile.rar',
                         '-x', 'file_1', '-x', 'file_2', '/home/strongbad/'],
                        runas=None, python_shell=False, template=None
                    )

                mock = MagicMock(return_value='salt')
                with patch.dict(archive.__salt__, {'cmd.run': mock}):
                    ret = archive.unrar(
                        '/tmp/rarfile.rar',
                        '/home/strongbad/',
                        excludes=['file_1', 'file_2']
                    )
                    self.assertEqual(['salt'], ret)
                    mock.assert_called_once_with(
                        ['unrar', 'x', '-idp', '/tmp/rarfile.rar',
                         '-x', 'file_1', '-x', 'file_2', '/home/strongbad/'],
                        runas=None, python_shell=False, template=None
                    )

    def test_unrar_raises_exception_if_not_found(self):
        with patch('salt.utils.which_bin', lambda exe: None):
            mock = MagicMock(return_value='salt')
            with patch.dict(archive.__salt__, {'cmd.run': mock}):
                self.assertRaises(
                    CommandNotFoundError,
                    archive.unrar,
                    '/tmp/rarfile.rar',
                    '/home/strongbad/',
                )
                self.assertFalse(mock.called)

    def test_trim_files(self):
        with patch('salt.utils.which_bin', lambda exe: exe):
            source = 'file.tar.gz'
            tmp_dir = 'temp'
            files = [
                '\n'.join(['file1'] * 200),
                '\n'.join(['file2'] * 200),
                'this\nthat\nother',
                'this\nthat\nother',
                'this\nthat\nother'
            ]
            trim_opt = [
                True,
                False,
                3,
                1,
                0
            ]
            trim_100 = (['file1'] * 100)
            trim_100.append("List trimmed after 100 files.")
            expected = [
                trim_100,
                ['file2'] * 200,
                ['this', 'that', 'other'],
                ['this', 'List trimmed after 1 files.'],
                ['List trimmed after 0 files.'],
            ]

            for test_files, test_trim_opts, test_expected in zip(files, trim_opt, expected):
                with patch.dict(archive.__salt__, {'cmd.run': MagicMock(return_value=test_files)}):
                    ret = archive.unrar(source, tmp_dir, trim_output=test_trim_opts)
                    self.assertEqual(ret, test_expected)