123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457 |
- """
- Tests for the salt fileclient
- """
- import errno
- import logging
- import os
- import shutil
- import salt.utils.files
- from salt import fileclient
- from tests.support.mixins import (
- AdaptedConfigurationTestCaseMixin,
- LoaderModuleMockMixin,
- )
- from tests.support.mock import MagicMock, Mock, patch
- from tests.support.runtests import RUNTIME_VARS
- from tests.support.unit import TestCase
- log = logging.getLogger(__name__)
- class FileclientTestCase(TestCase):
- """
- Fileclient test
- """
- opts = {
- "extension_modules": "",
- "cachedir": "/__test__",
- }
- def _fake_makedir(self, num=errno.EEXIST):
- def _side_effect(*args, **kwargs):
- raise OSError(num, "Errno {}".format(num))
- return Mock(side_effect=_side_effect)
- def test_cache_skips_makedirs_on_race_condition(self):
- """
- If cache contains already a directory, do not raise an exception.
- """
- with patch("os.path.isfile", lambda prm: False):
- for exists in range(2):
- with patch("os.makedirs", self._fake_makedir()):
- with fileclient.Client(self.opts)._cache_loc(
- "testfile"
- ) as c_ref_itr:
- assert c_ref_itr == os.sep + os.sep.join(
- ["__test__", "files", "base", "testfile"]
- )
- def test_cache_raises_exception_on_non_eexist_ioerror(self):
- """
- If makedirs raises other than EEXIST errno, an exception should be raised.
- """
- with patch("os.path.isfile", lambda prm: False):
- with patch("os.makedirs", self._fake_makedir(num=errno.EROFS)):
- with self.assertRaises(OSError):
- with fileclient.Client(self.opts)._cache_loc(
- "testfile"
- ) as c_ref_itr:
- assert c_ref_itr == "/__test__/files/base/testfile"
- def test_extrn_path_with_long_filename(self):
- safe_file_name = os.path.split(
- fileclient.Client(self.opts)._extrn_path(
- "https://test.com/" + ("A" * 254), "base"
- )
- )[-1]
- assert safe_file_name == "A" * 254
- oversized_file_name = os.path.split(
- fileclient.Client(self.opts)._extrn_path(
- "https://test.com/" + ("A" * 255), "base"
- )
- )[-1]
- assert len(oversized_file_name) < 256
- assert oversized_file_name != "A" * 255
- oversized_file_with_query_params = os.path.split(
- fileclient.Client(self.opts)._extrn_path(
- "https://test.com/file?" + ("A" * 255), "base"
- )
- )[-1]
- assert len(oversized_file_with_query_params) < 256
- SALTENVS = ("base", "dev")
- SUBDIR = "subdir"
- SUBDIR_FILES = ("foo.txt", "bar.txt", "baz.txt")
- def _get_file_roots(fs_root):
- return {x: [os.path.join(fs_root, x)] for x in SALTENVS}
- class FileClientTest(
- TestCase, AdaptedConfigurationTestCaseMixin, LoaderModuleMockMixin
- ):
- def setup_loader_modules(self):
- FS_ROOT = os.path.join(RUNTIME_VARS.TMP, "fileclient_fs_root")
- CACHE_ROOT = os.path.join(RUNTIME_VARS.TMP, "fileclient_cache_root")
- MOCKED_OPTS = {
- "file_roots": _get_file_roots(FS_ROOT),
- "fileserver_backend": ["roots"],
- "cachedir": CACHE_ROOT,
- "file_client": "local",
- }
- self.addCleanup(shutil.rmtree, FS_ROOT, ignore_errors=True)
- self.addCleanup(shutil.rmtree, CACHE_ROOT, ignore_errors=True)
- return {fileclient: {"__opts__": MOCKED_OPTS}}
- def setUp(self):
- self.file_client = fileclient.Client(self.master_opts)
- def tearDown(self):
- del self.file_client
- def test_file_list_emptydirs(self):
- """
- Ensure that the fileclient class won't allow a direct call to file_list_emptydirs()
- """
- with self.assertRaises(NotImplementedError):
- self.file_client.file_list_emptydirs()
- def test_get_file(self):
- """
- Ensure that the fileclient class won't allow a direct call to get_file()
- """
- with self.assertRaises(NotImplementedError):
- self.file_client.get_file(None)
- def test_get_file_client(self):
- minion_opts = self.get_temp_config("minion")
- minion_opts["file_client"] = "remote"
- with patch(
- "salt.fileclient.RemoteClient", MagicMock(return_value="remote_client")
- ):
- ret = fileclient.get_file_client(minion_opts)
- self.assertEqual("remote_client", ret)
- class FileclientCacheTest(
- TestCase, AdaptedConfigurationTestCaseMixin, LoaderModuleMockMixin
- ):
- """
- Tests for the fileclient caching. The LocalClient is the only thing we can
- test as it is the only way we can mock the fileclient (the tests run from
- the minion process, so the master cannot be mocked from test code).
- """
- def setup_loader_modules(self):
- self.FS_ROOT = os.path.join(RUNTIME_VARS.TMP, "fileclient_fs_root")
- self.CACHE_ROOT = os.path.join(RUNTIME_VARS.TMP, "fileclient_cache_root")
- self.MOCKED_OPTS = {
- "file_roots": _get_file_roots(self.FS_ROOT),
- "fileserver_backend": ["roots"],
- "cachedir": self.CACHE_ROOT,
- "file_client": "local",
- }
- self.addCleanup(shutil.rmtree, self.FS_ROOT, ignore_errors=True)
- self.addCleanup(shutil.rmtree, self.CACHE_ROOT, ignore_errors=True)
- return {fileclient: {"__opts__": self.MOCKED_OPTS}}
- def setUp(self):
- """
- No need to add a dummy foo.txt to muddy up the github repo, just make
- our own fileserver root on-the-fly.
- """
- def _new_dir(path):
- """
- Add a new dir at ``path`` using os.makedirs. If the directory
- already exists, remove it recursively and then try to create it
- again.
- """
- try:
- os.makedirs(path)
- except OSError as exc:
- if exc.errno == errno.EEXIST:
- # Just in case a previous test was interrupted, remove the
- # directory and try adding it again.
- shutil.rmtree(path)
- os.makedirs(path)
- else:
- raise
- # Crete the FS_ROOT
- for saltenv in SALTENVS:
- saltenv_root = os.path.join(self.FS_ROOT, saltenv)
- # Make sure we have a fresh root dir for this saltenv
- _new_dir(saltenv_root)
- path = os.path.join(saltenv_root, "foo.txt")
- with salt.utils.files.fopen(path, "w") as fp_:
- fp_.write("This is a test file in the '{}' saltenv.\n".format(saltenv))
- subdir_abspath = os.path.join(saltenv_root, SUBDIR)
- os.makedirs(subdir_abspath)
- for subdir_file in SUBDIR_FILES:
- path = os.path.join(subdir_abspath, subdir_file)
- with salt.utils.files.fopen(path, "w") as fp_:
- fp_.write(
- "This is file '{}' in subdir '{} from saltenv "
- "'{}'".format(subdir_file, SUBDIR, saltenv)
- )
- # Create the CACHE_ROOT
- _new_dir(self.CACHE_ROOT)
- def test_cache_dir(self):
- """
- Ensure entire directory is cached to correct location
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- for saltenv in SALTENVS:
- self.assertTrue(
- client.cache_dir("salt://{}".format(SUBDIR), saltenv, cachedir=None)
- )
- for subdir_file in SUBDIR_FILES:
- cache_loc = os.path.join(
- fileclient.__opts__["cachedir"],
- "files",
- saltenv,
- SUBDIR,
- subdir_file,
- )
- # Double check that the content of the cached file
- # identifies it as being from the correct saltenv. The
- # setUp function creates the file with the name of the
- # saltenv mentioned in the file, so a simple 'in' check is
- # sufficient here. If opening the file raises an exception,
- # this is a problem, so we are not catching the exception
- # and letting it be raised so that the test fails.
- with salt.utils.files.fopen(cache_loc) as fp_:
- content = fp_.read()
- log.debug("cache_loc = %s", cache_loc)
- log.debug("content = %s", content)
- self.assertTrue(subdir_file in content)
- self.assertTrue(SUBDIR in content)
- self.assertTrue(saltenv in content)
- def test_cache_dir_with_alternate_cachedir_and_absolute_path(self):
- """
- Ensure entire directory is cached to correct location when an alternate
- cachedir is specified and that cachedir is an absolute path
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- alt_cachedir = os.path.join(RUNTIME_VARS.TMP, "abs_cachedir")
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- for saltenv in SALTENVS:
- self.assertTrue(
- client.cache_dir(
- "salt://{}".format(SUBDIR), saltenv, cachedir=alt_cachedir
- )
- )
- for subdir_file in SUBDIR_FILES:
- cache_loc = os.path.join(
- alt_cachedir, "files", saltenv, SUBDIR, subdir_file
- )
- # Double check that the content of the cached file
- # identifies it as being from the correct saltenv. The
- # setUp function creates the file with the name of the
- # saltenv mentioned in the file, so a simple 'in' check is
- # sufficient here. If opening the file raises an exception,
- # this is a problem, so we are not catching the exception
- # and letting it be raised so that the test fails.
- with salt.utils.files.fopen(cache_loc) as fp_:
- content = fp_.read()
- log.debug("cache_loc = %s", cache_loc)
- log.debug("content = %s", content)
- self.assertTrue(subdir_file in content)
- self.assertTrue(SUBDIR in content)
- self.assertTrue(saltenv in content)
- def test_cache_dir_with_alternate_cachedir_and_relative_path(self):
- """
- Ensure entire directory is cached to correct location when an alternate
- cachedir is specified and that cachedir is a relative path
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- alt_cachedir = "foo"
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- for saltenv in SALTENVS:
- self.assertTrue(
- client.cache_dir(
- "salt://{}".format(SUBDIR), saltenv, cachedir=alt_cachedir
- )
- )
- for subdir_file in SUBDIR_FILES:
- cache_loc = os.path.join(
- fileclient.__opts__["cachedir"],
- alt_cachedir,
- "files",
- saltenv,
- SUBDIR,
- subdir_file,
- )
- # Double check that the content of the cached file
- # identifies it as being from the correct saltenv. The
- # setUp function creates the file with the name of the
- # saltenv mentioned in the file, so a simple 'in' check is
- # sufficient here. If opening the file raises an exception,
- # this is a problem, so we are not catching the exception
- # and letting it be raised so that the test fails.
- with salt.utils.files.fopen(cache_loc) as fp_:
- content = fp_.read()
- log.debug("cache_loc = %s", cache_loc)
- log.debug("content = %s", content)
- self.assertTrue(subdir_file in content)
- self.assertTrue(SUBDIR in content)
- self.assertTrue(saltenv in content)
- def test_cache_file(self):
- """
- Ensure file is cached to correct location
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- for saltenv in SALTENVS:
- self.assertTrue(
- client.cache_file("salt://foo.txt", saltenv, cachedir=None)
- )
- cache_loc = os.path.join(
- fileclient.__opts__["cachedir"], "files", saltenv, "foo.txt"
- )
- # Double check that the content of the cached file identifies
- # it as being from the correct saltenv. The setUp function
- # creates the file with the name of the saltenv mentioned in
- # the file, so a simple 'in' check is sufficient here. If
- # opening the file raises an exception, this is a problem, so
- # we are not catching the exception and letting it be raised so
- # that the test fails.
- with salt.utils.files.fopen(cache_loc) as fp_:
- content = fp_.read()
- log.debug("cache_loc = %s", cache_loc)
- log.debug("content = %s", content)
- self.assertTrue(saltenv in content)
- def test_cache_file_with_alternate_cachedir_and_absolute_path(self):
- """
- Ensure file is cached to correct location when an alternate cachedir is
- specified and that cachedir is an absolute path
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- alt_cachedir = os.path.join(RUNTIME_VARS.TMP, "abs_cachedir")
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- for saltenv in SALTENVS:
- self.assertTrue(
- client.cache_file("salt://foo.txt", saltenv, cachedir=alt_cachedir)
- )
- cache_loc = os.path.join(alt_cachedir, "files", saltenv, "foo.txt")
- # Double check that the content of the cached file identifies
- # it as being from the correct saltenv. The setUp function
- # creates the file with the name of the saltenv mentioned in
- # the file, so a simple 'in' check is sufficient here. If
- # opening the file raises an exception, this is a problem, so
- # we are not catching the exception and letting it be raised so
- # that the test fails.
- with salt.utils.files.fopen(cache_loc) as fp_:
- content = fp_.read()
- log.debug("cache_loc = %s", cache_loc)
- log.debug("content = %s", content)
- self.assertTrue(saltenv in content)
- def test_cache_file_with_alternate_cachedir_and_relative_path(self):
- """
- Ensure file is cached to correct location when an alternate cachedir is
- specified and that cachedir is a relative path
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- alt_cachedir = "foo"
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- for saltenv in SALTENVS:
- self.assertTrue(
- client.cache_file("salt://foo.txt", saltenv, cachedir=alt_cachedir)
- )
- cache_loc = os.path.join(
- fileclient.__opts__["cachedir"],
- alt_cachedir,
- "files",
- saltenv,
- "foo.txt",
- )
- # Double check that the content of the cached file identifies
- # it as being from the correct saltenv. The setUp function
- # creates the file with the name of the saltenv mentioned in
- # the file, so a simple 'in' check is sufficient here. If
- # opening the file raises an exception, this is a problem, so
- # we are not catching the exception and letting it be raised so
- # that the test fails.
- with salt.utils.files.fopen(cache_loc) as fp_:
- content = fp_.read()
- log.debug("cache_loc = %s", cache_loc)
- log.debug("content = %s", content)
- self.assertTrue(saltenv in content)
- def test_cache_dest(self):
- """
- Tests functionality for cache_dest
- """
- patched_opts = {x: y for x, y in self.minion_opts.items()}
- patched_opts.update(self.MOCKED_OPTS)
- relpath = "foo.com/bar.txt"
- cachedir = self.minion_opts["cachedir"]
- def _external(saltenv="base"):
- return salt.utils.path.join(
- patched_opts["cachedir"], "extrn_files", saltenv, relpath
- )
- def _salt(saltenv="base"):
- return salt.utils.path.join(
- patched_opts["cachedir"], "files", saltenv, relpath
- )
- def _check(ret, expected):
- assert ret == expected, "{} != {}".format(ret, expected)
- with patch.dict(fileclient.__opts__, patched_opts):
- client = fileclient.get_file_client(fileclient.__opts__, pillar=False)
- _check(client.cache_dest("https://" + relpath), _external())
- _check(client.cache_dest("https://" + relpath, "dev"), _external("dev"))
- _check(client.cache_dest("salt://" + relpath), _salt())
- _check(client.cache_dest("salt://" + relpath, "dev"), _salt("dev"))
- _check(
- client.cache_dest("salt://" + relpath + "?saltenv=dev"), _salt("dev")
- )
- _check("/foo/bar", "/foo/bar")
|