salt/tests/integration/shell/test_cp.py

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


    tests.integration.shell.cp
    ~~~~~~~~~~~~~~~~~~~~~~~~~~
"""

from __future__ import absolute_import

import logging
import os
import pipes

import pytest
import salt.utils.files
import salt.utils.platform
import salt.utils.yaml
from salt.ext import six
from tests.support.case import ShellCase
from tests.support.helpers import slowTest
from tests.support.mixins import ShellCaseCommonTestsMixin
from tests.support.runtests import RUNTIME_VARS

log = logging.getLogger(__name__)


@pytest.mark.windows_whitelisted
class CopyTest(ShellCase, ShellCaseCommonTestsMixin):

    _call_binary_ = "salt-cp"

    @slowTest
    def test_cp_testfile(self):
        """
        test salt-cp
        """
        minions = []
        for line in self.run_salt('--out yaml "*" test.ping'):
            if not line:
                continue
            data = salt.utils.yaml.safe_load(line)
            minions.extend(data.keys())

        self.assertNotEqual(minions, [])

        testfile = os.path.abspath(
            os.path.join(
                os.path.dirname(os.path.dirname(__file__)),
                "files",
                "file",
                "base",
                "testfile",
            )
        )
        with salt.utils.files.fopen(testfile, "r") as fh_:
            testfile_contents = fh_.read()

        def quote(arg):
            if salt.utils.platform.is_windows():
                return arg
            return pipes.quote(arg)

        for idx, minion in enumerate(minions):
            if "localhost" in minion:
                continue
            ret = self.run_salt(
                "--out yaml {0} file.directory_exists {1}".format(
                    quote(minion), RUNTIME_VARS.TMP
                )
            )
            data = salt.utils.yaml.safe_load("\n".join(ret))
            if data[minion] is False:
                ret = self.run_salt(
                    "--out yaml {0} file.makedirs {1}".format(
                        quote(minion), RUNTIME_VARS.TMP
                    )
                )

                data = salt.utils.yaml.safe_load("\n".join(ret))
                self.assertTrue(data[minion])

            minion_testfile = os.path.join(
                RUNTIME_VARS.TMP, "cp_{0}_testfile".format(idx)
            )

            ret = self.run_cp(
                "--out pprint {0} {1} {2}".format(
                    quote(minion), quote(testfile), quote(minion_testfile),
                )
            )

            data = eval("\n".join(ret), {}, {})  # pylint: disable=eval-used
            for part in six.itervalues(data):
                key = minion_testfile
                self.assertTrue(part[key])

            ret = self.run_salt(
                "--out yaml {0} file.file_exists {1}".format(
                    quote(minion), quote(minion_testfile)
                )
            )
            data = salt.utils.yaml.safe_load("\n".join(ret))
            self.assertTrue(data[minion])

            ret = self.run_salt(
                "--out yaml {0} file.contains {1} {2}".format(
                    quote(minion), quote(minion_testfile), quote(testfile_contents)
                )
            )
            data = salt.utils.yaml.safe_load("\n".join(ret))
            self.assertTrue(data[minion])
            ret = self.run_salt(
                "--out yaml {0} file.remove {1}".format(
                    quote(minion), quote(minion_testfile)
                )
            )
            data = salt.utils.yaml.safe_load("\n".join(ret))
            self.assertTrue(data[minion])