salt/tests/integration/modules/test_archive.py

# -*- coding: utf-8 -*-
"""
Tests for the archive state
"""
from __future__ import absolute_import, print_function, unicode_literals

import os
import shutil
import textwrap

import pytest
import salt.utils.files
import salt.utils.path
import salt.utils.platform
import salt.utils.stringutils
from salt.ext import six
from tests.support.case import ModuleCase
from tests.support.runtests import RUNTIME_VARS
from tests.support.unit import skipIf

try:
    import zipfile  # pylint: disable=unused-import

    HAS_ZIPFILE = True
except ImportError:
    HAS_ZIPFILE = False


@pytest.mark.destructive_test
@pytest.mark.windows_whitelisted
class ArchiveTest(ModuleCase):
    """
    Validate the archive module
    """

    @classmethod
    def setUpClass(cls):
        # Base path used for test artifacts
        cls.base_path = os.path.join(RUNTIME_VARS.TMP, "modules", "archive")

    @classmethod
    def tearDownClass(cls):
        cls.base_path = None

    def _set_artifact_paths(self, arch_fmt):
        """
        Define the paths for the source, archive, and destination files

        :param str arch_fmt: The archive format used in the test
        """
        self.src = os.path.join(self.base_path, "{0}_src_dir".format(arch_fmt))
        self.src_file = os.path.join(self.src, "file")
        self.arch = os.path.join(self.base_path, "archive.{0}".format(arch_fmt))
        self.dst = os.path.join(self.base_path, "{0}_dst_dir".format(arch_fmt))

    def _set_up(self, arch_fmt, unicode_filename=False):
        """
        Create source file tree and destination directory

        :param str arch_fmt: The archive format used in the test
        """
        self._set_artifact_paths(arch_fmt)

        # Remove the artifacts if any present
        if any([os.path.exists(f) for f in (self.src, self.arch, self.dst)]):
            self._tear_down()
            self._set_artifact_paths(arch_fmt)

        # Create source
        os.makedirs(self.src)
        if unicode_filename:
            filename = "file®"
        else:
            filename = "file"
        with salt.utils.files.fopen(os.path.join(self.src, filename), "wb") as theorem:
            if six.PY3 and salt.utils.platform.is_windows():
                encoding = "utf-8"
            else:
                encoding = None
            theorem.write(
                salt.utils.stringutils.to_bytes(
                    textwrap.dedent(
                        """\
                Compression theorem of computational complexity theory:

                Given a Gödel numbering $φ$ of the computable functions and a
                Blum complexity measure $Φ$ where a complexity class for a
                boundary function $f$ is defined as

                    $\\mathrm C(f) := \\{φ_i ∈ \\mathbb R^{(1)} | (∀^∞ x) Φ_i(x) ≤ f(x)\\}$.

                Then there exists a total computable function $f$ so that for
                all $i$

                    $\\mathrm{Dom}(φ_i) = \\mathrm{Dom}(φ_{f(i)})$

                and

                    $\\mathrm C(φ_i) ⊊ \\mathrm{C}(φ_{f(i)})$.
            """
                    ),
                    encoding=encoding,
                )
            )

        # Create destination
        os.makedirs(self.dst)

    def _tear_down(self):
        """
        Remove source file tree, archive, and destination file tree
        """
        for f in (self.src, self.arch, self.dst):
            if os.path.exists(f):
                if os.path.isdir(f):
                    shutil.rmtree(f, ignore_errors=True)
                else:
                    os.remove(f)
        del self.dst
        del self.src
        del self.arch
        del self.src_file

    def _assert_artifacts_in_ret(self, ret, file_only=False, unix_sep=False):
        """
        Assert that the artifact source files are printed in the source command
        output
        """

        def normdir(path):
            normdir = os.path.normcase(os.path.abspath(path))
            if salt.utils.platform.is_windows():
                # Remove the drive portion of path
                if len(normdir) >= 2 and normdir[1] == ":":
                    normdir = normdir.split(":", 1)[1]
            normdir = normdir.lstrip(os.path.sep)
            # Unzipped paths might have unix line endings
            if unix_sep:
                normdir = normdir.replace(os.path.sep, "/")
            return normdir

        # Try to find source directory and file in output lines
        dir_in_ret = None
        file_in_ret = None
        for line in ret:
            if normdir(self.src) in os.path.normcase(line) and not normdir(
                self.src_file
            ) in os.path.normcase(line):
                dir_in_ret = True
            if normdir(self.src_file) in os.path.normcase(line):
                file_in_ret = True

        # Assert number of lines, reporting of source directory and file
        self.assertTrue(len(ret) >= 1 if file_only else 2)
        if not file_only:
            self.assertTrue(dir_in_ret)
        self.assertTrue(file_in_ret)

    @skipIf(not salt.utils.path.which("tar"), "Cannot find tar executable")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_tar_pack(self):
        """
        Validate using the tar function to create archives
        """
        self._set_up(arch_fmt="tar")

        # Test create archive
        ret = self.run_function("archive.tar", ["-cvf", self.arch], sources=self.src)
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("tar"), "Cannot find tar executable")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_tar_unpack(self):
        """
        Validate using the tar function to extract archives
        """
        self._set_up(arch_fmt="tar")
        self.run_function("archive.tar", ["-cvf", self.arch], sources=self.src)

        # Test extract archive
        ret = self.run_function("archive.tar", ["-xvf", self.arch], dest=self.dst)
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("tar"), "Cannot find tar executable")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_tar_pack_unicode(self):
        """
        Validate using the tar function to create archives
        """
        self._set_up(arch_fmt="tar", unicode_filename=True)

        # Test create archive
        ret = self.run_function("archive.tar", ["-cvf", self.arch], sources=self.src)
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("tar"), "Cannot find tar executable")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_tar_unpack_unicode(self):
        """
        Validate using the tar function to extract archives
        """
        self._set_up(arch_fmt="tar", unicode_filename=True)
        self.run_function("archive.tar", ["-cvf", self.arch], sources=self.src)

        # Test extract archive
        ret = self.run_function("archive.tar", ["-xvf", self.arch], dest=self.dst)
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("tar"), "Cannot find tar executable")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_tar_list_unicode(self):
        """
        Validate using the tar function to extract archives
        """
        self._set_up(arch_fmt="tar", unicode_filename=True)
        self.run_function("archive.tar", ["-cvf", self.arch], sources=self.src)

        # Test list archive
        ret = self.run_function("archive.list", name=self.arch)
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("gzip"), "Cannot find gzip executable")
    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_gzip(self):
        """
        Validate using the gzip function
        """
        self._set_up(arch_fmt="gz")

        # Test create archive
        ret = self.run_function("archive.gzip", [self.src_file], options="-v")
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret, file_only=True)

        self._tear_down()

    @skipIf(not salt.utils.path.which("gzip"), "Cannot find gzip executable")
    @skipIf(not salt.utils.path.which("gunzip"), "Cannot find gunzip executable")
    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_gunzip(self):
        """
        Validate using the gunzip function
        """
        self._set_up(arch_fmt="gz")
        self.run_function("archive.gzip", [self.src_file], options="-v")

        # Test extract archive
        ret = self.run_function("archive.gunzip", [self.src_file + ".gz"], options="-v")
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret, file_only=True)

        self._tear_down()

    @skipIf(not salt.utils.path.which("zip"), "Cannot find zip executable")
    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_cmd_zip(self):
        """
        Validate using the cmd_zip function
        """
        self._set_up(arch_fmt="zip")

        # Test create archive
        ret = self.run_function("archive.cmd_zip", [self.arch, self.src])
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("zip"), "Cannot find zip executable")
    @skipIf(not salt.utils.path.which("unzip"), "Cannot find unzip executable")
    @pytest.mark.slow_test(seconds=1)  # Test takes >0.1 and <=1 seconds
    def test_cmd_unzip(self):
        """
        Validate using the cmd_unzip function
        """
        self._set_up(arch_fmt="zip")
        self.run_function("archive.cmd_zip", [self.arch, self.src])

        # Test create archive
        ret = self.run_function("archive.cmd_unzip", [self.arch, self.dst])
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not HAS_ZIPFILE, "Cannot find zipfile python module")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_zip(self):
        """
        Validate using the zip function
        """
        self._set_up(arch_fmt="zip")

        # Test create archive
        ret = self.run_function("archive.zip", [self.arch, self.src])
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not HAS_ZIPFILE, "Cannot find zipfile python module")
    @pytest.mark.slow_test(seconds=30)  # Test takes >10 and <=30 seconds
    def test_unzip(self):
        """
        Validate using the unzip function
        """
        self._set_up(arch_fmt="zip")
        self.run_function("archive.zip", [self.arch, self.src])

        # Test create archive
        ret = self.run_function("archive.unzip", [self.arch, self.dst])
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret, unix_sep=False)

        self._tear_down()

    @skipIf(not salt.utils.path.which("rar"), "Cannot find rar executable")
    def test_rar(self):
        """
        Validate using the rar function
        """
        self._set_up(arch_fmt="rar")

        # Test create archive
        ret = self.run_function("archive.rar", [self.arch, self.src])
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()

    @skipIf(not salt.utils.path.which("rar"), "Cannot find rar executable")
    @skipIf(not salt.utils.path.which("unrar"), "Cannot find unrar executable")
    def test_unrar(self):
        """
        Validate using the unrar function
        """
        self._set_up(arch_fmt="rar")
        self.run_function("archive.rar", [self.arch, self.src])

        # Test create archive
        ret = self.run_function("archive.unrar", [self.arch, self.dst])
        self.assertTrue(isinstance(ret, list), six.text_type(ret))
        self._assert_artifacts_in_ret(ret)

        self._tear_down()