Update salt-ssh state wrapper pillar handling

Instead of passing the pre-rendered pillar as an override, do it like
the regular `state` execution module does with `salt-call`: Check if the
pillar needs to be rendered, otherwise reuse the already rendered one.

Also, ensure that __pillar__ in wrapper modules contains the same one
used during rendering, same thing for the one passed to `state.pkg`.

Also, ensure that when pillars are rerendered during a state run, they
get the master opts in addition to the minion ones, since some modules
used in the pillar can rely on them to be present.

Also, ensure pillar overrides are accepted for the same functions as with
the regular `state` execution module.

changelog/59802.fixed.md

@@ -0,0 +1 @@
+Fixed merging of complex pillar overrides with salt-ssh states

changelog/62230.fixed.md

@@ -0,0 +1 @@
+Made salt-ssh states not re-render pillars unnecessarily

changelog/65483.fixed.md

@@ -0,0 +1 @@
+Ensured the pillar in SSH wrapper modules is the same as the one used in template rendering when overrides are passed

salt/client/ssh/state.py

@@ -31,10 +31,17 @@ class SSHState(salt.state.State):
     Create a State object which wraps the SSH functions for state operations
     """
 
-    def __init__(self, opts, pillar=None, wrapper=None, context=None):
+    def __init__(
+        self,
+        opts,
+        pillar_override=None,
+        wrapper=None,
+        context=None,
+        initial_pillar=None,
+    ):
         self.wrapper = wrapper
         self.context = context
-        super().__init__(opts, pillar)
+        super().__init__(opts, pillar_override, initial_pillar=initial_pillar)
 
     def load_modules(self, data=None, proxy=None):
         """
@@ -49,6 +56,21 @@ class SSHState(salt.state.State):
         )
         self.rend = salt.loader.render(self.opts, self.functions)
 
+    def _gather_pillar(self):
+        """
+        The opts used during pillar rendering should contain the master
+        opts in the root namespace. self.opts is the modified minion opts,
+        containing the original master opts in `__master_opts__`.
+        """
+        _opts = self.opts
+        popts = {}
+        popts.update(_opts.get("__master_opts__", {}))
+        popts.update(_opts)
+        self.opts = popts
+        pillar = super()._gather_pillar()
+        self.opts = _opts
+        return pillar
+
     def check_refresh(self, data, ret):
         """
         Stub out check_refresh
@@ -69,10 +91,24 @@ class SSHHighState(salt.state.BaseHighState):
 
     stack = []
 
-    def __init__(self, opts, pillar=None, wrapper=None, fsclient=None, context=None):
+    def __init__(
+        self,
+        opts,
+        pillar_override=None,
+        wrapper=None,
+        fsclient=None,
+        context=None,
+        initial_pillar=None,
+    ):
         self.client = fsclient
         salt.state.BaseHighState.__init__(self, opts)
-        self.state = SSHState(opts, pillar, wrapper, context=context)
+        self.state = SSHState(
+            opts,
+            pillar_override,
+            wrapper,
+            context=context,
+            initial_pillar=initial_pillar,
+        )
         self.matchers = salt.loader.matchers(self.opts)
         self.tops = salt.loader.tops(self.opts)
 

salt/client/ssh/wrapper/state.py

@@ -28,7 +28,7 @@ __func_alias__ = {"apply_": "apply"}
 log = logging.getLogger(__name__)
 
 
-def _ssh_state(chunks, st_kwargs, kwargs, test=False):
+def _ssh_state(chunks, st_kwargs, kwargs, pillar, test=False):
     """
     Function to run a state with the given chunk via salt-ssh
     """
@@ -43,7 +43,7 @@ def _ssh_state(chunks, st_kwargs, kwargs, test=False):
         __context__["fileclient"],
         chunks,
         file_refs,
-        __pillar__.value(),
+        pillar,
         st_kwargs["id_"],
     )
     trans_tar_sum = salt.utils.hashutils.get_hash(trans_tar, __opts__["hash_type"])
@@ -173,21 +173,30 @@ def sls(mods, saltenv="base", test=None, exclude=None, **kwargs):
     """
     st_kwargs = __salt__.kwargs
     __opts__["grains"] = __grains__.value()
-    __pillar__.update(kwargs.get("pillar", {}))
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
     opts["test"] = _get_test_value(test, **kwargs)
+    initial_pillar = _get_initial_pillar(opts)
+    pillar_override = kwargs.get("pillar")
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__.value(),
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
             err = ["Pillar failed to render with the following messages:"]
             err += st_.opts["pillar"]["_errors"]
             return err
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.push_active()
         mods = _parse_mods(mods)
         high_data, errors = st_.render_highstate(
@@ -231,7 +240,7 @@ def sls(mods, saltenv="base", test=None, exclude=None, **kwargs):
             __context__["fileclient"],
             chunks,
             file_refs,
-            __pillar__.value(),
+            pillar,
             st_kwargs["id_"],
             roster_grains,
         )
@@ -329,12 +338,7 @@ def _check_queue(queue, kwargs):
 
 
 def _get_initial_pillar(opts):
-    return (
-        __pillar__
-        if __opts__["__cli"] == "salt-call"
-        and opts["pillarenv"] == __opts__["pillarenv"]
-        else None
-    )
+    return __pillar__.value() if opts["pillarenv"] == __opts__["pillarenv"] else None
 
 
 def low(data, **kwargs):
@@ -353,10 +357,11 @@ def low(data, **kwargs):
     chunks = [data]
     with salt.client.ssh.state.SSHHighState(
         __opts__,
-        __pillar__.value(),
+        None,
         __salt__.value(),
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=__pillar__.value(),
     ) as st_:
         for chunk in chunks:
             chunk["__id__"] = (
@@ -440,17 +445,26 @@ def high(data, **kwargs):
 
         salt '*' state.high '{"vim": {"pkg": ["installed"]}}'
     """
-    __pillar__.update(kwargs.get("pillar", {}))
     st_kwargs = __salt__.kwargs
     __opts__["grains"] = __grains__.value()
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__.value(),
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.push_active()
         chunks = st_.state.compile_high_data(data)
         file_refs = salt.client.ssh.state.lowstate_file_refs(
@@ -469,7 +483,7 @@ def high(data, **kwargs):
             __context__["fileclient"],
             chunks,
             file_refs,
-            __pillar__.value(),
+            pillar,
             st_kwargs["id_"],
             roster_grains,
         )
@@ -677,23 +691,32 @@ def highstate(test=None, **kwargs):
         salt '*' state.highstate exclude=sls_to_exclude
         salt '*' state.highstate exclude="[{'id': 'id_to_exclude'}, {'sls': 'sls_to_exclude'}]"
     """
-    __pillar__.update(kwargs.get("pillar", {}))
     st_kwargs = __salt__.kwargs
     __opts__["grains"] = __grains__.value()
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
     opts["test"] = _get_test_value(test, **kwargs)
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__.value(),
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
             err = ["Pillar failed to render with the following messages:"]
             err += st_.opts["pillar"]["_errors"]
             return err
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.push_active()
         chunks = st_.compile_low_chunks(context=__context__.value())
         file_refs = salt.client.ssh.state.lowstate_file_refs(
@@ -717,7 +740,7 @@ def highstate(test=None, **kwargs):
             __context__["fileclient"],
             chunks,
             file_refs,
-            __pillar__.value(),
+            pillar,
             st_kwargs["id_"],
             roster_grains,
         )
@@ -764,26 +787,32 @@ def top(topfn, test=None, **kwargs):
         salt '*' state.top reverse_top.sls exclude=sls_to_exclude
         salt '*' state.top reverse_top.sls exclude="[{'id': 'id_to_exclude'}, {'sls': 'sls_to_exclude'}]"
     """
-    __pillar__.update(kwargs.get("pillar", {}))
     st_kwargs = __salt__.kwargs
     __opts__["grains"] = __grains__.value()
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
-    if salt.utils.args.test_mode(test=test, **kwargs):
-        opts["test"] = True
-    else:
-        opts["test"] = __opts__.get("test", None)
+    opts["test"] = _get_test_value(test, **kwargs)
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__.value(),
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
             err = ["Pillar failed to render with the following messages:"]
             err += st_.opts["pillar"]["_errors"]
             return err
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.opts["state_top"] = os.path.join("salt://", topfn)
         st_.push_active()
         chunks = st_.compile_low_chunks(context=__context__.value())
@@ -808,7 +837,7 @@ def top(topfn, test=None, **kwargs):
             __context__["fileclient"],
             chunks,
             file_refs,
-            __pillar__.value(),
+            pillar,
             st_kwargs["id_"],
             roster_grains,
         )
@@ -855,18 +884,28 @@ def show_highstate(**kwargs):
     """
     __opts__["grains"] = __grains__.value()
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__,
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
             err = ["Pillar failed to render with the following messages:"]
             err += st_.opts["pillar"]["_errors"]
             return err
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.push_active()
         chunks = st_.compile_highstate(context=__context__.value())
         # Check for errors
@@ -891,10 +930,11 @@ def show_lowstate(**kwargs):
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        None,
         __salt__,
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=_get_initial_pillar(opts),
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
@@ -939,7 +979,6 @@ def sls_id(id_, mods, test=None, queue=False, **kwargs):
 
         salt '*' state.sls_id my_state my_module,a_common_module
     """
-    __pillar__.update(kwargs.get("pillar", {}))
     st_kwargs = __salt__.kwargs
     conflict = _check_queue(queue, kwargs)
     if conflict is not None:
@@ -953,12 +992,15 @@ def sls_id(id_, mods, test=None, queue=False, **kwargs):
     if opts["saltenv"] is None:
         opts["saltenv"] = "base"
 
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         __opts__,
-        __pillar__.value(),
+        pillar_override,
         __salt__,
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
 
         if not _check_pillar(kwargs, st_.opts["pillar"]):
@@ -967,6 +1009,13 @@ def sls_id(id_, mods, test=None, queue=False, **kwargs):
             err += __pillar__["_errors"]
             return err
 
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         split_mods = _parse_mods(mods)
         st_.push_active()
         high_, errors = st_.render_highstate(
@@ -992,7 +1041,7 @@ def sls_id(id_, mods, test=None, queue=False, **kwargs):
                 )
             )
 
-        ret = _ssh_state(chunk, st_kwargs, kwargs, test=test)
+        ret = _ssh_state(chunk, st_kwargs, kwargs, pillar, test=test)
         _set_retcode(ret, highstate=highstate)
         # Work around Windows multiprocessing bug, set __opts__['test'] back to
         # value from before this function was run.
@@ -1011,25 +1060,31 @@ def show_sls(mods, saltenv="base", test=None, **kwargs):
 
         salt '*' state.show_sls core,edit.vim dev
     """
-    __pillar__.update(kwargs.get("pillar", {}))
     __opts__["grains"] = __grains__.value()
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
-    if salt.utils.args.test_mode(test=test, **kwargs):
-        opts["test"] = True
-    else:
-        opts["test"] = __opts__.get("test", None)
+    opts["test"] = _get_test_value(test, **kwargs)
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__,
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
             err = ["Pillar failed to render with the following messages:"]
             err += st_.opts["pillar"]["_errors"]
             return err
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.push_active()
         mods = _parse_mods(mods)
         high_data, errors = st_.render_highstate(
@@ -1065,26 +1120,31 @@ def show_low_sls(mods, saltenv="base", test=None, **kwargs):
 
         salt '*' state.show_low_sls core,edit.vim dev
     """
-    __pillar__.update(kwargs.get("pillar", {}))
     __opts__["grains"] = __grains__.value()
-
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
-    if salt.utils.args.test_mode(test=test, **kwargs):
-        opts["test"] = True
-    else:
-        opts["test"] = __opts__.get("test", None)
+    opts["test"] = _get_test_value(test, **kwargs)
+    pillar_override = kwargs.get("pillar")
+    initial_pillar = _get_initial_pillar(opts)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        pillar_override,
         __salt__,
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=initial_pillar,
     ) as st_:
         if not _check_pillar(kwargs, st_.opts["pillar"]):
             __context__["retcode"] = salt.defaults.exitcodes.EX_PILLAR_FAILURE
             err = ["Pillar failed to render with the following messages:"]
             err += st_.opts["pillar"]["_errors"]
             return err
+        try:
+            pillar = st_.opts["pillar"].value()
+        except AttributeError:
+            pillar = st_.opts["pillar"]
+        if pillar_override is not None or initial_pillar is None:
+            # Ensure other wrappers use the correct pillar
+            __pillar__.update(pillar)
         st_.push_active()
         mods = _parse_mods(mods)
         high_data, errors = st_.render_highstate(
@@ -1122,10 +1182,11 @@ def show_top(**kwargs):
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
     with salt.client.ssh.state.SSHHighState(
         opts,
-        __pillar__.value(),
+        None,
         __salt__,
         __context__["fileclient"],
         context=__context__.value(),
+        initial_pillar=_get_initial_pillar(opts),
     ) as st_:
         top_data = st_.get_top(context=__context__.value())
         errors = []
@@ -1171,17 +1232,22 @@ def single(fun, name, test=None, **kwargs):
     opts = salt.utils.state.get_sls_opts(__opts__, **kwargs)
 
     # Set test mode
-    if salt.utils.args.test_mode(test=test, **kwargs):
-        opts["test"] = True
-    else:
-        opts["test"] = __opts__.get("test", None)
+    opts["test"] = _get_test_value(test, **kwargs)
 
     # Get the override pillar data
-    __pillar__.update(kwargs.get("pillar", {}))
+    # This needs to be removed from the kwargs, they are called
+    # as a lowstate with one item, not a single chunk
+    pillar_override = kwargs.pop("pillar", None)
 
     # Create the State environment
-    st_ = salt.client.ssh.state.SSHState(opts, __pillar__)
+    st_ = salt.client.ssh.state.SSHState(
+        opts, pillar_override, initial_pillar=_get_initial_pillar(opts)
+    )
 
+    try:
+        pillar = st_.opts["pillar"].value()
+    except AttributeError:
+        pillar = st_.opts["pillar"]
     # Verify the low chunk
     err = st_.verify_data(kwargs)
     if err:
@@ -1208,7 +1274,7 @@ def single(fun, name, test=None, **kwargs):
         __context__["fileclient"],
         chunks,
         file_refs,
-        __pillar__.value(),
+        pillar,
         st_kwargs["id_"],
         roster_grains,
     )

tests/pytests/integration/ssh/test_state.py

@@ -2,6 +2,7 @@ import json
 
 import pytest
 
+import salt.utils.dictupdate
 from salt.defaults.exitcodes import EX_AGGREGATE
 
 pytestmark = [
@@ -561,3 +562,299 @@ class TestStateRunFailRetcode:
     def test_retcode_state_top_run_fail(self, salt_ssh_cli):
         ret = salt_ssh_cli.run("state.top", "top.sls")
         assert ret.returncode == EX_AGGREGATE
+
+
+@pytest.fixture(scope="class")
+def pillar_tree_nested(base_env_pillar_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - nested
+      '127.0.0.1':
+        - nested
+    """
+    nested_pillar = r"""
+    {%- do salt.log.warning("hithere: pillar was rendered") %}
+    monty: python
+    the_meaning:
+      of:
+        life: 42
+        bar: tender
+      for: what
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_pillar_tree_root_dir
+    )
+    nested_tempfile = pytest.helpers.temp_file(
+        "nested.sls", nested_pillar, base_env_pillar_tree_root_dir
+    )
+    with top_tempfile, nested_tempfile:
+        yield
+
+
+@pytest.mark.usefixtures("pillar_tree_nested")
+def test_pillar_is_only_rendered_once_without_overrides(salt_ssh_cli, caplog):
+    ret = salt_ssh_cli.run("state.apply", "test")
+    assert ret.returncode == 0
+    assert isinstance(ret.data, dict)
+    assert ret.data
+    assert ret.data[next(iter(ret.data))]["result"] is True
+    assert caplog.text.count("hithere: pillar was rendered") == 1
+
+
+@pytest.mark.usefixtures("pillar_tree_nested")
+def test_pillar_is_rerendered_with_overrides(salt_ssh_cli, caplog):
+    ret = salt_ssh_cli.run("state.apply", "test", pillar={"foo": "bar"})
+    assert ret.returncode == 0
+    assert isinstance(ret.data, dict)
+    assert ret.data
+    assert ret.data[next(iter(ret.data))]["result"] is True
+    assert caplog.text.count("hithere: pillar was rendered") == 2
+
+
+@pytest.mark.slow_test
+@pytest.mark.usefixtures("pillar_tree_nested")
+class TestStatePillarOverride:
+    """
+    Ensure pillar overrides are merged recursively, that wrapper
+    modules are in sync with the pillar dict in the rendering environment
+    and that the pillars are available on the target.
+    """
+
+    @pytest.fixture(scope="class", autouse=True)
+    def _show_pillar_state(self, base_env_state_tree_root_dir):
+        top_file = """
+        base:
+          'localhost':
+            - showpillar
+          '127.0.0.1':
+            - showpillar
+        """
+        show_pillar_sls = """
+        deep_thought:
+          test.show_notification:
+            - text: '{{ {
+                "raw": {
+                  "the_meaning": pillar.get("the_meaning"),
+                  "btw": pillar.get("btw")},
+                "wrapped": {
+                  "the_meaning": salt["pillar.get"]("the_meaning"),
+                  "btw": salt["pillar.get"]("btw")}}
+                | json }}'
+
+        target_check:
+          test.check_pillar:
+            - present:
+              - the_meaning:of:foo
+              - btw
+              - the_meaning:of:bar
+              - the_meaning:for
+            - listing:
+              - the_meaning:of:life
+        """
+        top_tempfile = pytest.helpers.temp_file(
+            "top.sls", top_file, base_env_state_tree_root_dir
+        )
+        show_tempfile = pytest.helpers.temp_file(
+            "showpillar.sls", show_pillar_sls, base_env_state_tree_root_dir
+        )
+        with top_tempfile, show_tempfile:
+            yield
+
+    @pytest.fixture
+    def base(self):
+        return {"the_meaning": {"of": {"life": 42, "bar": "tender"}, "for": "what"}}
+
+    @pytest.fixture
+    def override(self, base):
+        poverride = {
+            "the_meaning": {"of": {"life": [2.71], "foo": "lish"}},
+            "btw": "turtles",
+        }
+        expected = salt.utils.dictupdate.merge(base, poverride)
+        return expected, poverride
+
+    def test_state_sls(self, salt_ssh_cli, override):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.sls", "showpillar", pillar=override)
+        self._assert_basic(ret)
+        assert len(ret.data) == 2
+        for sid, sret in ret.data.items():
+            if "show" in sid:
+                self._assert_pillar(sret["comment"], expected)
+            else:
+                assert sret["result"] is True
+
+    @pytest.mark.parametrize("sid", ("deep_thought", "target_check"))
+    def test_state_sls_id(self, salt_ssh_cli, sid, override):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.sls_id", sid, "showpillar", pillar=override)
+        self._assert_basic(ret)
+        state_res = ret.data[next(iter(ret.data))]
+        if sid == "deep_thought":
+            self._assert_pillar(state_res["comment"], expected)
+        else:
+            assert state_res["result"] is True
+
+    def test_state_highstate(self, salt_ssh_cli, override):
+        expected, override = override
+        ret = salt_ssh_cli.run(
+            "state.highstate", pillar=override, whitelist=["showpillar"]
+        )
+        self._assert_basic(ret)
+        assert len(ret.data) == 2
+        for sid, sret in ret.data.items():
+            if "show" in sid:
+                self._assert_pillar(sret["comment"], expected)
+            else:
+                assert sret["result"] is True
+
+    def test_state_show_sls(self, salt_ssh_cli, override):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.show_sls", "showpillar", pillar=override)
+        self._assert_basic(ret)
+        pillar = ret.data["deep_thought"]["test"]
+        pillar = next(x["text"] for x in pillar if isinstance(x, dict))
+        self._assert_pillar(pillar, expected)
+
+    def test_state_show_low_sls(self, salt_ssh_cli, override):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.show_low_sls", "showpillar", pillar=override)
+        self._assert_basic(ret, list)
+        pillar = ret.data[0]["text"]
+        self._assert_pillar(pillar, expected)
+
+    def test_state_single(self, salt_ssh_cli, override):
+        expected, override = override
+        ret = salt_ssh_cli.run(
+            "state.single",
+            "test.check_pillar",
+            "foo",
+            present=[
+                "the_meaning:of:foo",
+                "btw",
+                "the_meaning:of:bar",
+                "the_meaning:for",
+            ],
+            listing=["the_meaning:of:life"],
+            pillar=override,
+        )
+        self._assert_basic(ret, dict)
+        state_res = ret.data[next(iter(ret.data))]
+        assert state_res["result"] is True
+
+    def test_state_top(self, salt_ssh_cli, override):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.top", "top.sls", pillar=override)
+        self._assert_basic(ret)
+        assert len(ret.data) == 2
+        for sid, sret in ret.data.items():
+            if "show" in sid:
+                self._assert_pillar(sret["comment"], expected)
+            else:
+                assert sret["result"] is True
+
+    def _assert_pillar(self, pillar, expected):
+        if not isinstance(pillar, dict):
+            pillar = json.loads(pillar)
+        assert pillar["raw"] == expected
+        assert pillar["wrapped"] == expected
+
+    def _assert_basic(self, ret, typ=dict):
+        assert ret.returncode == 0
+        assert isinstance(ret.data, typ)
+        assert ret.data
+
+
+@pytest.mark.slow_test
+@pytest.mark.usefixtures("pillar_tree_nested")
+class TestStatePillarOverrideTemplate:
+    """
+    Specifically ensure that pillars are merged as expected
+    for the target as well and available for renderers.
+    This should be covered by `test.check_pillar` above, but
+    let's check the specific output for the most important funcs.
+    Issue #59802
+    """
+
+    @pytest.fixture
+    def _write_pillar_state(self, base_env_state_tree_root_dir, tmp_path_factory):
+        tmp_path = tmp_path_factory.mktemp("tgtdir")
+        tgt_file = tmp_path / "deepthought.txt"
+        top_file = """
+        base:
+          'localhost':
+            - writepillar
+          '127.0.0.1':
+            - writepillar
+        """
+        nested_pillar_file = f"""
+        deep_thought:
+          file.managed:
+            - name: {tgt_file}
+            - source: salt://deepthought.txt.jinja
+            - template: jinja
+        """
+        # deepthought = "{{ {'the_meaning': pillar.get('the_meaning'), 'btw': pillar.get('btw')} | json }}"
+        deepthought = r"""
+        {{
+          {
+            "raw": {
+              "the_meaning": pillar.get("the_meaning"),
+              "btw": pillar.get("btw")},
+            "modules": {
+              "the_meaning": salt["pillar.get"]("the_meaning"),
+              "btw": salt["pillar.get"]("btw")}
+          } | json }}
+        """
+        top_tempfile = pytest.helpers.temp_file(
+            "top.sls", top_file, base_env_state_tree_root_dir
+        )
+        show_tempfile = pytest.helpers.temp_file(
+            "writepillar.sls", nested_pillar_file, base_env_state_tree_root_dir
+        )
+        deepthought_tempfile = pytest.helpers.temp_file(
+            "deepthought.txt.jinja", deepthought, base_env_state_tree_root_dir
+        )
+
+        with top_tempfile, show_tempfile, deepthought_tempfile:
+            yield tgt_file
+
+    @pytest.fixture
+    def base(self):
+        return {"the_meaning": {"of": {"life": 42, "bar": "tender"}, "for": "what"}}
+
+    @pytest.fixture
+    def override(self, base):
+        poverride = {
+            "the_meaning": {"of": {"life": 2.71, "foo": "lish"}},
+            "btw": "turtles",
+        }
+        expected = salt.utils.dictupdate.merge(base, poverride)
+        return expected, poverride
+
+    def test_state_sls(self, salt_ssh_cli, override, _write_pillar_state):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.sls", "writepillar", pillar=override)
+        self._assert_pillar(ret, expected, _write_pillar_state)
+
+    def test_state_highstate(self, salt_ssh_cli, override, _write_pillar_state):
+        expected, override = override
+        ret = salt_ssh_cli.run(
+            "state.highstate", pillar=override, whitelist=["writepillar"]
+        )
+        self._assert_pillar(ret, expected, _write_pillar_state)
+
+    def test_state_top(self, salt_ssh_cli, override, _write_pillar_state):
+        expected, override = override
+        ret = salt_ssh_cli.run("state.top", "top.sls", pillar=override)
+        self._assert_pillar(ret, expected, _write_pillar_state)
+
+    def _assert_pillar(self, ret, expected, path):
+        assert ret.returncode == 0
+        assert isinstance(ret.data, dict)
+        assert ret.data
+        assert path.exists()
+        pillar = json.loads(path.read_text())
+        assert pillar["raw"] == expected
+        assert pillar["modules"] == expected