tests/pytests/unit/utils/test_cache.py - master - githubmirror/salt

tests/pytests/unit/utils/test_cache.py
378 строк11 KB
Daniel A. Wozniak
Merge branch '3006.x' into 3007.x (cycle 2)
01 июл 2026, 09:36
01 июл 2026, 09:36936016e
"""
    tests.unit.utils.cache_test
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~

    Test the salt cache objects
"""

import logging
import pathlib
import time

import pytest

import salt.loader
import salt.payload
import salt.utils.cache as cache
import salt.utils.data
import salt.utils.files
from tests.support.mock import patch


def test_sanity():
    """
    Make sure you can instantiate etc.
    """
    cd = cache.CacheDict(5)
    assert isinstance(cd, cache.CacheDict)

    # do some tests to make sure it looks like a dict
    assert "foo" not in cd
    cd["foo"] = "bar"
    assert cd["foo"] == "bar"
    del cd["foo"]
    assert "foo" not in cd


def test_ttl():
    cd = cache.CacheDict(0.1)
    cd["foo"] = "bar"
    assert "foo" in cd
    assert cd["foo"] == "bar"
    time.sleep(0.2)
    assert "foo" not in cd

    # make sure that a get would get a regular old key error
    with pytest.raises(KeyError):
        cd["foo"]  # pylint: disable=pointless-statement


def test_cache_regex_sweep_with_equal_usage_counts():
    """
    CacheRegex must be able to sweep and remove the outdated or least frequently
    """
    regex_cache = cache.CacheRegex(size=2, keep_fraction=0.5)

    # Populate the cache and make two patterns share the same frequency
    regex_cache.get("pattern1")
    regex_cache.get("pattern2")
    regex_cache.get("pattern1")
    regex_cache.get("pattern2")

    # Add a third pattern without triggering a sweep yet
    regex_cache.get("pattern3")

    # Adding a fourth pattern triggers a sweep internally
    compiled = regex_cache.get("pattern4")

    assert compiled is not None
    assert "pattern4" in regex_cache.cache


@pytest.fixture
def cache_dir(minion_opts):
    return pathlib.Path(minion_opts["cachedir"])


def test_smoke_context(minion_opts):
    """
    Smoke test the context cache
    """
    context_cache = cache.ContextCache(minion_opts, "cache_test")

    data = {"a": "b"}
    context_cache.cache_context(data.copy())

    ret = context_cache.get_cache_context()

    assert ret == data


def test_cache_context_writes_with_owner_only_mode(minion_opts):
    """
    ``ContextCache.cache_context`` must produce a cache file with mode
    ``0o600`` (owner-read+write only). The file holds pillar context
    which can contain credentials (passwords, vault tokens, API keys);
    on the previous implementation it inherited the process umask and
    became world-readable on most Linux distributions.
    """
    import os
    import stat

    context_cache = cache.ContextCache(minion_opts, "cache_test_perms")
    context_cache.cache_context({"secret": "value"})

    mode = stat.S_IMODE(os.stat(context_cache.cache_path).st_mode)
    assert mode == stat.S_IRUSR | stat.S_IWUSR, (
        f"ContextCache file mode is {oct(mode)}, expected 0o600 "
        f"(owner-read+write only). World-readable pillar cache leaks "
        f"credentials to any local user."
    )


def test_cache_context_creates_parent_directory_with_owner_only_mode(
    minion_opts, tmp_path
):
    """
    The on-disk ``context/`` directory holding cached pillar data must
    be created with mode ``0o700`` (owner-only). The previous
    implementation created the directory with the default ``os.mkdir``
    mode of ``0o755`` modulo the process umask -- which on most Linux
    installs left it world-readable, so any local user could ``ls`` the
    directory and learn which modules and external-pillar backends were
    cached (filenames look like ``salt.loaded.ext.<module>.<name>.p``).
    """
    import os
    import stat

    # Use a fresh cachedir so context/ does not pre-exist from another
    # test, otherwise the mode of the existing directory is whatever
    # earlier code left behind and the test can't verify our behaviour.
    fresh_opts = {**minion_opts, "cachedir": str(tmp_path / "fresh_cachedir")}
    os.makedirs(fresh_opts["cachedir"])
    cc = cache.ContextCache(fresh_opts, "first_write")
    cc.cache_context({"v": 1})

    parent = os.path.dirname(cc.cache_path)
    mode = stat.S_IMODE(os.stat(parent).st_mode)
    assert mode == stat.S_IRWXU, (
        f"context/ dir mode is {oct(mode)}, expected 0o700 "
        f"(owner-only). World-readable parent leaks cache filenames "
        f"and metadata to local users."
    )


def test_cache_context_overwrites_atomically_via_replace(minion_opts):
    """
    ``ContextCache.cache_context`` must overwrite the cache file
    atomically: a concurrent reader during a re-cache must see either
    the previous content or the new one in full, never a partially
    written file. The ``tempfile.mkstemp`` + ``os.replace`` path makes
    this guarantee at the POSIX layer; the previous in-place
    truncate-and-write path did not. Compare inode numbers across two
    successive writes -- atomic replace gives a fresh inode each time
    (the rename swaps the directory entry to a new file), in-place
    truncate keeps the inode constant.
    """
    import os

    cc = cache.ContextCache(minion_opts, "cache_test_atomic")
    cc.cache_context({"v": 1})
    inode_first = os.stat(cc.cache_path).st_ino
    cc.cache_context({"v": 2})
    inode_second = os.stat(cc.cache_path).st_ino

    assert inode_first != inode_second, (
        "ContextCache write reused the same inode, indicating in-place "
        "truncate-and-write rather than atomic mkstemp+replace; readers "
        "concurrent with the write can observe a partial file."
    )
    assert cc.get_cache_context() == {"v": 2}


@pytest.fixture
def cache_mod_name():
    return "cache_mod"


@pytest.fixture
def cache_mods_path(tmp_path, cache_mod_name):
    _cache_mods_path = tmp_path / "cache_mods"
    mod_contents = """
    import salt.utils.cache

    def __virtual__():
        return True

    @salt.utils.cache.context_cache
    def test_context_module():
        if "called" in __context__:
            __context__["called"] += 1
        else:
            __context__["called"] = 0
        return __context__.value()

    @salt.utils.cache.context_cache
    def test_compare_context():
        return __context__.value()
    """
    with pytest.helpers.temp_file(
        cache_mod_name + ".py", mod_contents, _cache_mods_path
    ):
        yield _cache_mods_path


def test_context_wrapper(minion_opts, cache_mods_path):
    """
    Test to ensure that a module which decorates itself
    with a context cache can store and retrieve its contextual
    data
    """

    loader = salt.loader.LazyLoader(
        [str(cache_mods_path)],
        tag="rawmodule",
        virtual_enable=False,
        opts=minion_opts,
    )

    cache_test_func = loader["cache_mod.test_context_module"]

    assert cache_test_func()["called"] == 0
    assert cache_test_func()["called"] == 1


def test_set_cache(minion_opts, cache_mods_path, cache_mod_name, cache_dir):
    """
    Tests to ensure the cache is written correctly
    """

    context = {"c": "d"}
    loader = salt.loader.LazyLoader(
        [str(cache_mods_path)],
        tag="rawmodule",
        virtual_enable=False,
        opts=minion_opts,
        pack={"__context__": context, "__opts__": minion_opts},
    )

    cache_test_func = loader["cache_mod.test_context_module"]

    # Call the function to trigger the context cache
    assert cache_test_func()["called"] == 0
    assert cache_test_func()["called"] == 1
    assert cache_test_func()["called"] == 2

    cache_file_name = f"salt.loaded.ext.rawmodule.{cache_mod_name}.p"

    cached_file = cache_dir / "context" / cache_file_name
    assert cached_file.exists()

    # Test manual de-serialize
    target_cache_data = salt.utils.data.decode(
        salt.payload.loads(cached_file.read_bytes())
    )
    assert target_cache_data == dict(context, called=1)

    # Test cache de-serialize
    cc = cache.ContextCache(minion_opts, f"salt.loaded.ext.rawmodule.{cache_mod_name}")
    retrieved_cache = cc.get_cache_context()
    assert retrieved_cache == dict(context, called=1)


def test_refill_cache(minion_opts, cache_mods_path):
    """
    Tests to ensure that the context cache can rehydrate a wrapped function
    """
    context = {"c": "d"}
    loader = salt.loader.LazyLoader(
        [str(cache_mods_path)],
        tag="rawmodule",
        virtual_enable=False,
        opts=minion_opts,
        pack={"__context__": context, "__opts__": minion_opts},
    )

    cache_test_func = loader["cache_mod.test_compare_context"]
    # First populate the cache
    ret = cache_test_func()
    assert ret == context

    # Then try to rehydrate a func
    context_copy = context.copy()
    context.clear()

    # Compare to the context before it was emptied
    ret = cache_test_func()
    assert ret == context_copy


def test_everything(cache_dir):
    """
    Make sure you can instantiate, add, update, remove, expire
    """
    path = str(cache_dir / "minion")

    # test instantiation
    cd = cache.CacheDisk(0.3, path)
    assert isinstance(cd, cache.CacheDisk)

    # test to make sure it looks like a dict
    assert "foo" not in cd
    cd["foo"] = "bar"
    assert "foo" in cd
    assert cd["foo"] == "bar"
    del cd["foo"]
    assert "foo" not in cd

    # test persistence
    cd["foo"] = "bar"
    cd2 = cache.CacheDisk(0.3, path)
    assert "foo" in cd2
    assert cd2["foo"] == "bar"

    # test ttl
    time.sleep(0.5)
    assert "foo" not in cd
    assert "foo" not in cd2


@pytest.mark.parametrize(
    "data",
    [
        b"PK\x03\x04\n\x00\x00\x00\x00\x00\xb6B\x05S\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x06\x00\x1c\x00test2/",
        b"\xc3\x83\xc2\xa6\xc3\x83\xc2\xb8\xc3\x83\xc2\xa5",
    ],
)
def test_unicode_error(cache_dir, data, caplog):
    """
    Test when the data in the cache raises a UnicodeDecodeError
    we do not raise an error.
    """
    path = cache_dir / "minion"
    path.touch()
    cache_data = {
        "CacheDisk_data": {
            b"poc-minion": {
                None: {
                    b"secrets": {
                        b"itsasecret": data,
                        b"CacheDisk_cachetime": {b"poc-minion": 1649339137.1236317},
                    }
                }
            }
        }
    }
    with patch.object(
        salt.utils.msgpack, "load", return_value=cache_data
    ), caplog.at_level(logging.DEBUG):
        cd = cache.CacheDisk(0.3, str(path))
        # this test used to rely on msgpack throwing errors if attempt to read an empty file
        # code now checks if file empty and returns, so we should never attempt msgpack load
        assert cd._dict == {}
        assert not (
            f"Error reading cache file at '{path}': Unpack failed: incomplete input"
            in caplog.messages
        )


def test_cache_corruption(cache_dir):
    """
    Tests if the CacheDisk can survive a corrupted cache file.
    """
    # Write valid cache file
    cache_file = cache_dir / "minion"
    cd = cache.CacheDisk(0.3, str(cache_file))
    cd["test-key"] = "test-value"
    del cd

    # Add random string to the data to make the msgpack structure un-decodable
    with cache_file.open("a") as f:
        f.write("I am data that should corrupt the msgpack file")

    # Reopen cache, try to fetch key
    cd = cache.CacheDisk(0.3, str(cache_file))
    # If the cache is unreadable, we want it to act like an empty cache (as
    # if the file did not exist in the first place), and should raise a KeyError
    with pytest.raises(KeyError):
        assert cd["test-key"]