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.helpers import destructiveTest, slowTest
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


@destructiveTest
@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")
    @slowTest
    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")
    @slowTest
    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")
    @slowTest
    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")
    @slowTest
    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")
    @slowTest
    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")
    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")
    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")
    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")
    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")
    @slowTest
    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")
    @slowTest
    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()