diff --git a/changelog/59802.fixed.md b/changelog/59802.fixed.md
new file mode 100644
index 000000000000..e83222951c7d
--- /dev/null
+++ b/changelog/59802.fixed.md
@@ -0,0 +1 @@
+Fixed merging of complex pillar overrides with salt-ssh states
diff --git a/changelog/60002.fixed.md b/changelog/60002.fixed.md
new file mode 100644
index 000000000000..8d3869b7a3b7
--- /dev/null
+++ b/changelog/60002.fixed.md
@@ -0,0 +1 @@
+Fixed gpg pillar rendering with salt-ssh
diff --git a/changelog/62230.fixed.md b/changelog/62230.fixed.md
new file mode 100644
index 000000000000..8c83287a76fb
--- /dev/null
+++ b/changelog/62230.fixed.md
@@ -0,0 +1 @@
+Made salt-ssh states not re-render pillars unnecessarily
diff --git a/changelog/65483.fixed.md b/changelog/65483.fixed.md
new file mode 100644
index 000000000000..8092c6072d34
--- /dev/null
+++ b/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
diff --git a/salt/client/ssh/__init__.py b/salt/client/ssh/__init__.py
index 8601d8d17457..60f3b6a98bc4 100644
--- a/salt/client/ssh/__init__.py
+++ b/salt/client/ssh/__init__.py
@@ -1196,9 +1196,11 @@ def run_wfunc(self):
                 for grain in self.target["grains"]:
                     opts_pkg["grains"][grain] = self.target["grains"][grain]
 
+            # Pillar compilation needs the master opts primarily,
+            # same as during regular operation.
             popts = {}
-            popts.update(opts_pkg["__master_opts__"])
             popts.update(opts_pkg)
+            popts.update(opts_pkg["__master_opts__"])
             pillar = salt.pillar.Pillar(
                 popts,
                 opts_pkg["grains"],
diff --git a/salt/client/ssh/state.py b/salt/client/ssh/state.py
index 4ee62a293a00..ccc72198b651 100644
--- a/salt/client/ssh/state.py
+++ b/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,28 @@ def load_modules(self, data=None, proxy=None):
         )
         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 = {}
+        # Pillar compilation needs the master opts primarily,
+        # same as during regular operation.
+        popts.update(_opts)
+        popts.update(_opts.get("__master_opts__", {}))
+        # But, salt.state.State takes the parameters for get_pillar from
+        # the opts, so we need to ensure they are correct for the minion.
+        popts["id"] = _opts["id"]
+        popts["saltenv"] = _opts["saltenv"]
+        popts["pillarenv"] = _opts.get("pillarenv")
+        self.opts = popts
+        pillar = super()._gather_pillar()
+        self.opts = _opts
+        return pillar
+
     def check_refresh(self, data, ret):
         """
         Stub out check_refresh
@@ -69,10 +98,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)
 
diff --git a/salt/client/ssh/wrapper/state.py b/salt/client/ssh/wrapper/state.py
index 353d8a0e03eb..aa61e07f81e8 100644
--- a/salt/client/ssh/wrapper/state.py
+++ b/salt/client/ssh/wrapper/state.py
@@ -28,7 +28,7 @@
 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,
     )
diff --git a/tests/pytests/integration/ssh/state/conftest.py b/tests/pytests/integration/ssh/state/conftest.py
new file mode 100644
index 000000000000..14d645ae8e8a
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/conftest.py
@@ -0,0 +1,132 @@
+import pytest
+
+
+@pytest.fixture(scope="module")
+def state_tree(base_env_state_tree_root_dir):
+    top_file = """
+    {%- from "map.jinja" import abc with context %}
+    base:
+      'localhost':
+        - basic
+      '127.0.0.1':
+        - basic
+    """
+    map_file = """
+    {%- set abc = "def" %}
+    """
+    state_file = """
+    {%- from "map.jinja" import abc with context %}
+    Ok with {{ abc }}:
+      test.succeed_without_changes
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    map_tempfile = pytest.helpers.temp_file(
+        "map.jinja", map_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "test.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, map_tempfile, state_tempfile:
+        yield
+
+
+@pytest.fixture(scope="module")
+def state_tree_dir(base_env_state_tree_root_dir):
+    """
+    State tree with files to test salt-ssh
+    when the map.jinja file is in another directory
+    """
+    top_file = """
+    {%- from "test/map.jinja" import abc with context %}
+    base:
+      'localhost':
+        - test
+      '127.0.0.1':
+        - test
+    """
+    map_file = """
+    {%- set abc = "def" %}
+    """
+    state_file = """
+    {%- from "test/map.jinja" import abc with context %}
+
+    Ok with {{ abc }}:
+      test.succeed_without_changes
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    map_tempfile = pytest.helpers.temp_file(
+        "test/map.jinja", map_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "test.sls", state_file, base_env_state_tree_root_dir
+    )
+
+    with top_tempfile, map_tempfile, state_tempfile:
+        yield
+
+
+@pytest.fixture
+def nested_state_tree(base_env_state_tree_root_dir, tmp_path):
+    top_file = """
+    base:
+      'localhost':
+        - basic
+      '127.0.0.1':
+        - basic
+    """
+    state_file = """
+    /{}/file.txt:
+      file.managed:
+        - source: salt://foo/file.jinja
+        - template: jinja
+    """.format(
+        tmp_path
+    )
+    file_jinja = """
+    {% from 'foo/map.jinja' import comment %}{{ comment }}
+    """
+    map_file = """
+    {% set comment = "blah blah" %}
+    """
+    statedir = base_env_state_tree_root_dir / "foo"
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    map_tempfile = pytest.helpers.temp_file("map.jinja", map_file, statedir)
+    file_tempfile = pytest.helpers.temp_file("file.jinja", file_jinja, statedir)
+    state_tempfile = pytest.helpers.temp_file("init.sls", state_file, statedir)
+
+    with top_tempfile, map_tempfile, state_tempfile, file_tempfile:
+        yield
+
+
+@pytest.fixture(scope="module")
+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
diff --git a/tests/pytests/integration/ssh/state/test_pillar_override.py b/tests/pytests/integration/ssh/state/test_pillar_override.py
new file mode 100644
index 000000000000..c7f49b2b816d
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_pillar_override.py
@@ -0,0 +1,189 @@
+"""
+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.
+"""
+
+import json
+
+import pytest
+
+import salt.utils.dictupdate
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.usefixtures("pillar_tree_nested"),
+    pytest.mark.slow_test,
+]
+
+
+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
+
+
+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.fixture(scope="module", autouse=True)
+def _show_pillar_state(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():
+    return {"the_meaning": {"of": {"life": 42, "bar": "tender"}, "for": "what"}}
+
+
+@pytest.fixture
+def override(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(salt_ssh_cli, override):
+    expected, override = override
+    ret = salt_ssh_cli.run("state.sls", "showpillar", pillar=override)
+    _assert_basic(ret)
+    assert len(ret.data) == 2
+    for sid, sret in ret.data.items():
+        if "show" in sid:
+            _assert_pillar(sret["comment"], expected)
+        else:
+            assert sret["result"] is True
+
+
+@pytest.mark.parametrize("sid", ("deep_thought", "target_check"))
+def test_state_sls_id(salt_ssh_cli, sid, override):
+    expected, override = override
+    ret = salt_ssh_cli.run("state.sls_id", sid, "showpillar", pillar=override)
+    _assert_basic(ret)
+    state_res = ret.data[next(iter(ret.data))]
+    if sid == "deep_thought":
+        _assert_pillar(state_res["comment"], expected)
+    else:
+        assert state_res["result"] is True
+
+
+def test_state_highstate(salt_ssh_cli, override):
+    expected, override = override
+    ret = salt_ssh_cli.run("state.highstate", pillar=override, whitelist=["showpillar"])
+    _assert_basic(ret)
+    assert len(ret.data) == 2
+    for sid, sret in ret.data.items():
+        if "show" in sid:
+            _assert_pillar(sret["comment"], expected)
+        else:
+            assert sret["result"] is True
+
+
+def test_state_show_sls(salt_ssh_cli, override):
+    expected, override = override
+    ret = salt_ssh_cli.run("state.show_sls", "showpillar", pillar=override)
+    _assert_basic(ret)
+    pillar = ret.data["deep_thought"]["test"]
+    pillar = next(x["text"] for x in pillar if isinstance(x, dict))
+    _assert_pillar(pillar, expected)
+
+
+def test_state_show_low_sls(salt_ssh_cli, override):
+    expected, override = override
+    ret = salt_ssh_cli.run("state.show_low_sls", "showpillar", pillar=override)
+    _assert_basic(ret, list)
+    pillar = ret.data[0]["text"]
+    _assert_pillar(pillar, expected)
+
+
+def test_state_single(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,
+    )
+    _assert_basic(ret, dict)
+    state_res = ret.data[next(iter(ret.data))]
+    assert state_res["result"] is True
+
+
+def test_state_top(salt_ssh_cli, override):
+    expected, override = override
+    ret = salt_ssh_cli.run("state.top", "top.sls", pillar=override)
+    _assert_basic(ret)
+    assert len(ret.data) == 2
+    for sid, sret in ret.data.items():
+        if "show" in sid:
+            _assert_pillar(sret["comment"], expected)
+        else:
+            assert sret["result"] is True
+
+
+def _assert_pillar(pillar, expected):
+    if not isinstance(pillar, dict):
+        pillar = json.loads(pillar)
+    assert pillar["raw"] == expected
+    assert pillar["wrapped"] == expected
+
+
+def _assert_basic(ret, typ=dict):
+    assert ret.returncode == 0
+    assert isinstance(ret.data, typ)
+    assert ret.data
diff --git a/tests/pytests/integration/ssh/state/test_pillar_override_template.py b/tests/pytests/integration/ssh/state/test_pillar_override_template.py
new file mode 100644
index 000000000000..610efe7040f8
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_pillar_override_template.py
@@ -0,0 +1,97 @@
+"""
+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
+"""
+
+import json
+
+import pytest
+
+import salt.utils.dictupdate
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.usefixtures("pillar_tree_nested"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture
+def _write_pillar_state(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 = 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():
+    return {"the_meaning": {"of": {"life": 42, "bar": "tender"}, "for": "what"}}
+
+
+@pytest.fixture
+def override(base):
+    poverride = {
+        "the_meaning": {"of": {"life": 2.71, "foo": "lish"}},
+        "btw": "turtles",
+    }
+    expected = salt.utils.dictupdate.merge(base, poverride)
+    return expected, poverride
+
+
+@pytest.mark.parametrize(
+    "args,kwargs",
+    (
+        (("state.sls", "writepillar"), {}),
+        (("state.highstate",), {"whitelist": "writepillar"}),
+        (("state.top", "top.sls"), {}),
+    ),
+)
+def test_it(salt_ssh_cli, args, kwargs, override, _write_pillar_state):
+    expected, override = override
+    ret = salt_ssh_cli.run(*args, **kwargs, pillar=override)
+    assert ret.returncode == 0
+    assert isinstance(ret.data, dict)
+    assert ret.data
+    assert _write_pillar_state.exists()
+    pillar = json.loads(_write_pillar_state.read_text())
+    assert pillar["raw"] == expected
+    assert pillar["modules"] == expected
diff --git a/tests/pytests/integration/ssh/state/test_retcode_highstate_verification_requisite_fail.py b/tests/pytests/integration/ssh/state/test_retcode_highstate_verification_requisite_fail.py
new file mode 100644
index 000000000000..6b629a248d0d
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_retcode_highstate_verification_requisite_fail.py
@@ -0,0 +1,62 @@
+"""
+Verify salt-ssh fails with a retcode > 0 when a highstate verification fails.
+``state.show_highstate`` does not validate this.
+"""
+
+import pytest
+
+from salt.defaults.exitcodes import EX_AGGREGATE
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def state_tree_req_fail(base_env_state_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - fail_req
+      '127.0.0.1':
+        - fail_req
+    """
+    state_file = """
+    This has an invalid requisite:
+      test.nop:
+        - name: foo
+        - require_in:
+          - file.managed: invalid_requisite
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "fail_req.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
+@pytest.mark.parametrize(
+    "args,retcode",
+    (
+        (("state.sls", "fail_req"), EX_AGGREGATE),
+        (("state.highstate",), EX_AGGREGATE),
+        (("state.show_sls", "fail_req"), EX_AGGREGATE),
+        (("state.show_low_sls", "fail_req"), EX_AGGREGATE),
+        # state.show_lowstate exits with 0 for non-ssh as well
+        (("state.show_lowstate",), 0),
+        (("state.top", "top.sls"), EX_AGGREGATE),
+    ),
+)
+def test_it(salt_ssh_cli, args, retcode):
+    ret = salt_ssh_cli.run(*args)
+    assert ret.returncode == retcode
+    assert isinstance(ret.data, list)
+    assert ret.data
+    assert isinstance(ret.data[0], str)
+    assert ret.data[0].startswith(
+        "Invalid requisite in require: file.managed for invalid_requisite"
+    )
diff --git a/tests/pytests/integration/ssh/state/test_retcode_highstate_verification_structure_fail.py b/tests/pytests/integration/ssh/state/test_retcode_highstate_verification_structure_fail.py
new file mode 100644
index 000000000000..c74de1e20fd4
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_retcode_highstate_verification_structure_fail.py
@@ -0,0 +1,64 @@
+"""
+Verify salt-ssh fails with a retcode > 0 when a highstate verification fails.
+This targets another step of the verification.
+``state.sls_id`` does not seem to support extends.
+``state.show_highstate`` does not validate this.
+"""
+
+import pytest
+
+from salt.defaults.exitcodes import EX_AGGREGATE
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def state_tree_structure_fail(base_env_state_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - fail_structure
+      '127.0.0.1':
+        - fail_structure
+    """
+    state_file = """
+    extend:
+      Some file state:
+        file:
+            - name: /tmp/bar
+            - contents: bar
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "fail_structure.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
+@pytest.mark.parametrize(
+    "args,retcode",
+    (
+        (("state.sls", "fail_structure"), EX_AGGREGATE),
+        (("state.highstate",), EX_AGGREGATE),
+        (("state.show_sls", "fail_structure"), EX_AGGREGATE),
+        (("state.show_low_sls", "fail_structure"), EX_AGGREGATE),
+        # state.show_lowstate exits with 0 for non-ssh as well
+        (("state.show_lowstate",), 0),
+        (("state.top", "top.sls"), EX_AGGREGATE),
+    ),
+)
+def test_it(salt_ssh_cli, args, retcode):
+    ret = salt_ssh_cli.run(*args)
+    assert ret.returncode == retcode
+    assert isinstance(ret.data, list)
+    assert ret.data
+    assert isinstance(ret.data[0], str)
+    assert ret.data[0].startswith(
+        "Cannot extend ID 'Some file state' in 'base:fail_structure"
+    )
diff --git a/tests/pytests/integration/ssh/state/test_retcode_pillar_render_exception.py b/tests/pytests/integration/ssh/state/test_retcode_pillar_render_exception.py
new file mode 100644
index 000000000000..92cd56dbb39c
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_retcode_pillar_render_exception.py
@@ -0,0 +1,57 @@
+"""
+Verify salt-ssh fails with a retcode > 0 when a pillar rendering fails.
+"""
+
+import pytest
+
+from salt.defaults.exitcodes import EX_AGGREGATE
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def pillar_tree_render_fail(base_env_pillar_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - fail_render
+      '127.0.0.1':
+        - fail_render
+    """
+    pillar_file = r"""
+    not_defined: {{ abc }}
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_pillar_tree_root_dir
+    )
+    pillar_tempfile = pytest.helpers.temp_file(
+        "fail_render.sls", pillar_file, base_env_pillar_tree_root_dir
+    )
+    with top_tempfile, pillar_tempfile:
+        yield
+
+
+@pytest.mark.parametrize(
+    "args",
+    (
+        ("state.sls", "basic"),
+        ("state.highstate",),
+        ("state.sls_id", "foo", "basic"),
+        ("state.show_sls", "basic"),
+        ("state.show_low_sls", "basic"),
+        ("state.show_highstate",),
+        ("state.show_lowstate",),
+        ("state.top", "top.sls"),
+    ),
+)
+def test_it(salt_ssh_cli, args):
+    ret = salt_ssh_cli.run(*args)
+    assert ret.returncode == EX_AGGREGATE
+    assert isinstance(ret.data, list)
+    assert ret.data
+    assert isinstance(ret.data[0], str)
+    assert ret.data[0] == "Pillar failed to render with the following messages:"
+    assert ret.data[1].startswith("Rendering SLS 'fail_render' failed.")
diff --git a/tests/pytests/integration/ssh/state/test_retcode_render_exception.py b/tests/pytests/integration/ssh/state/test_retcode_render_exception.py
new file mode 100644
index 000000000000..5291e107b91f
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_retcode_render_exception.py
@@ -0,0 +1,67 @@
+"""
+Verify salt-ssh fails with a retcode > 0 when a state rendering fails.
+"""
+
+import pytest
+
+from salt.defaults.exitcodes import EX_AGGREGATE
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def state_tree_render_fail(base_env_state_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - fail_render
+      '127.0.0.1':
+        - fail_render
+    """
+    state_file = r"""
+    abc var is not defined {{ abc }}:
+      test.nop
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "fail_render.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
+@pytest.mark.parametrize(
+    "args,retcode",
+    (
+        (("state.sls", "fail_render"), EX_AGGREGATE),
+        (("state.highstate",), EX_AGGREGATE),
+        (("state.sls_id", "foo", "fail_render"), EX_AGGREGATE),
+        (("state.show_sls", "fail_render"), EX_AGGREGATE),
+        (("state.show_low_sls", "fail_render"), EX_AGGREGATE),
+        (("state.show_highstate",), EX_AGGREGATE),
+        # state.show_lowstate exits with 0 for non-ssh as well
+        (("state.show_lowstate",), 0),
+        (("state.top", "top.sls"), EX_AGGREGATE),
+    ),
+)
+def test_it(salt_ssh_cli, args, retcode):
+    ret = salt_ssh_cli.run(*args)
+    assert ret.returncode == retcode
+    assert isinstance(ret.data, list)
+    assert ret.data
+    assert isinstance(ret.data[0], str)
+    assert ret.data[0].startswith(
+        "Rendering SLS 'base:fail_render' failed: Jinja variable 'abc' is undefined;"
+    )
+
+
+def test_state_single(salt_ssh_cli):
+    ret = salt_ssh_cli.run("state.single", "file")
+    assert ret.returncode == EX_AGGREGATE
+    assert isinstance(ret.data, str)
+    assert "single() missing 1 required positional argument" in ret.data
diff --git a/tests/pytests/integration/ssh/state/test_retcode_run_fail.py b/tests/pytests/integration/ssh/state/test_retcode_run_fail.py
new file mode 100644
index 000000000000..e77768dc5102
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_retcode_run_fail.py
@@ -0,0 +1,52 @@
+"""
+Verify salt-ssh passes on a failing retcode from state execution.
+"""
+
+import pytest
+
+from salt.defaults.exitcodes import EX_AGGREGATE
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def state_tree_run_fail(base_env_state_tree_root_dir):
+    top_file = """
+    base:
+      'localhost':
+        - fail_run
+      '127.0.0.1':
+        - fail_run
+    """
+    state_file = """
+    This file state fails:
+      file.managed:
+        - name: /tmp/non/ex/is/tent
+        - makedirs: false
+        - contents: foo
+    """
+    top_tempfile = pytest.helpers.temp_file(
+        "top.sls", top_file, base_env_state_tree_root_dir
+    )
+    state_tempfile = pytest.helpers.temp_file(
+        "fail_run.sls", state_file, base_env_state_tree_root_dir
+    )
+    with top_tempfile, state_tempfile:
+        yield
+
+
+@pytest.mark.parametrize(
+    "args",
+    (
+        ("state.sls", "fail_run"),
+        ("state.highstate",),
+        ("state.sls_id", "This file state fails", "fail_run"),
+        ("state.top", "top.sls"),
+    ),
+)
+def test_it(salt_ssh_cli, args):
+    ret = salt_ssh_cli.run(*args)
+    assert ret.returncode == EX_AGGREGATE
diff --git a/tests/pytests/integration/ssh/state/test_state.py b/tests/pytests/integration/ssh/state/test_state.py
new file mode 100644
index 000000000000..62e8cbf513b1
--- /dev/null
+++ b/tests/pytests/integration/ssh/state/test_state.py
@@ -0,0 +1,103 @@
+import pytest
+
+pytestmark = [
+    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
+    pytest.mark.slow_test,
+]
+
+
+def test_state_with_import(salt_ssh_cli, state_tree):
+    """
+    verify salt-ssh can use imported map files in states
+    """
+    ret = salt_ssh_cli.run("state.sls", "test")
+    assert ret.returncode == 0
+    assert ret.data
+
+
+@pytest.mark.parametrize(
+    "ssh_cmd",
+    [
+        "state.sls",
+        "state.highstate",
+        "state.apply",
+        "state.show_top",
+        "state.show_highstate",
+        "state.show_low_sls",
+        "state.show_lowstate",
+        "state.sls_id",
+        "state.show_sls",
+        "state.top",
+    ],
+)
+def test_state_with_import_dir(salt_ssh_cli, state_tree_dir, ssh_cmd):
+    """
+    verify salt-ssh can use imported map files in states
+    when the map files are in another directory outside of
+    sls files importing them.
+    """
+    if ssh_cmd in ("state.sls", "state.show_low_sls", "state.show_sls"):
+        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd, "test")
+    elif ssh_cmd == "state.top":
+        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd, "top.sls")
+    elif ssh_cmd == "state.sls_id":
+        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd, "Ok with def", "test")
+    else:
+        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd)
+    assert ret.returncode == 0
+    if ssh_cmd == "state.show_top":
+        assert ret.data == {"base": ["test", "master_tops_test"]} or {"base": ["test"]}
+    elif ssh_cmd in ("state.show_highstate", "state.show_sls"):
+        assert ret.data == {
+            "Ok with def": {
+                "__sls__": "test",
+                "__env__": "base",
+                "test": ["succeed_without_changes", {"order": 10000}],
+            }
+        }
+    elif ssh_cmd in ("state.show_low_sls", "state.show_lowstate", "state.show_sls"):
+        assert ret.data == [
+            {
+                "state": "test",
+                "name": "Ok with def",
+                "__sls__": "test",
+                "__env__": "base",
+                "__id__": "Ok with def",
+                "order": 10000,
+                "fun": "succeed_without_changes",
+            }
+        ]
+    else:
+        assert ret.data["test_|-Ok with def_|-Ok with def_|-succeed_without_changes"][
+            "result"
+        ]
+    assert ret.data
+
+
+def test_state_with_import_from_dir(salt_ssh_cli, nested_state_tree):
+    """
+    verify salt-ssh can use imported map files in states
+    """
+    ret = salt_ssh_cli.run(
+        "--extra-filerefs=salt://foo/map.jinja", "state.apply", "foo"
+    )
+    assert ret.returncode == 0
+    assert ret.data
+
+
+def test_state_low(salt_ssh_cli):
+    """
+    test state.low with salt-ssh
+    """
+    ret = salt_ssh_cli.run(
+        "state.low", '{"state": "cmd", "fun": "run", "name": "echo blah"}'
+    )
+    assert ret.data["cmd_|-echo blah_|-echo blah_|-run"]["changes"]["stdout"] == "blah"
+
+
+def test_state_high(salt_ssh_cli):
+    """
+    test state.high with salt-ssh
+    """
+    ret = salt_ssh_cli.run("state.high", '{"echo blah": {"cmd": ["run"]}}')
+    assert ret.data["cmd_|-echo blah_|-echo blah_|-run"]["changes"]["stdout"] == "blah"
diff --git a/tests/pytests/integration/ssh/test_pillar_compilation.py b/tests/pytests/integration/ssh/test_pillar_compilation.py
new file mode 100644
index 000000000000..042f4ea67cdb
--- /dev/null
+++ b/tests/pytests/integration/ssh/test_pillar_compilation.py
@@ -0,0 +1,241 @@
+import logging
+import pathlib
+import shutil
+import subprocess
+import textwrap
+
+import pytest
+from pytestshellutils.utils.processes import ProcessResult
+
+log = logging.getLogger(__name__)
+
+
+# The following fixtures are copied from pytests/functional/pillar/test_gpg.py
+
+
+@pytest.fixture(scope="module")
+def test_key():
+    """
+    Private key for setting up GPG pillar environment.
+    """
+    return textwrap.dedent(
+        """\
+        -----BEGIN PGP PRIVATE KEY BLOCK-----
+
+        lQOYBFiKrcYBCADAj92+fz20uKxxH0ffMwcryGG9IogkiUi2QrNYilB4hwrY5Qt7
+        Sbywlk/mSDMcABxMxS0vegqc5pgglvAnsi9w7j//9nfjiirsyiTYOOD1akTFQr7b
+        qT6zuGFA4oYmYHvfBOena485qvlyitYLKYT9h27TDiiH6Jgt4xSRbjeyhTf3/fKD
+        JzHA9ii5oeVi1pH/8/4USgXanBdKwO0JKQtci+PF0qe/nkzRswqTIkdgx1oyNUqL
+        tYJ0XPOy+UyOC4J4QDIt9PQbAmiur8By4g2lLYWlGOCjs7Fcj3n5meWKzf1pmXoY
+        lAnSab8kUZSSkoWQoTO7RbjFypULKCZui45/ABEBAAEAB/wM1wsAMtfYfx/wgxd1
+        yJ9HyhrKU80kMotIq/Xth3uKLecJQ2yakfYlCEDXqCTQTymT7OnwaoDeqXmnYqks
+        3HLRYvGdjb+8ym/GTkxapqBJfQaM6MB1QTnPHhJOE0zCrlhULK2NulxYihAMFTnk
+        kKYviaJYLG+DcH0FQkkS0XihTKcqnsoJiS6iNd5SME3pa0qijR0D5f78fkvNzzEE
+        9vgAX1TgQ5PDJGN6nYlW2bWxTcg+FR2cUAQPTiP9wXCH6VyJoQay7KHVr3r/7SsU
+        89otfcx5HVDYPrez6xnP6wN0P/mKxCDbkERLDjZjWOmNXg2zn+/t3u02e+ybfAIp
+        kTTxBADY/FmPgLpJ2bpcPH141twpHwhKIbENlTB9745Qknr6aLA0QVCkz49/3joO
+        Sj+SZ7Jhl6cfbynrfHwX3b1bOFTzBUH2Tsi0HX40PezEFH0apf55FLZuMOBt/lc1
+        ET6evpIHF0dcM+BvZa7E7MyTyEq8S7Cc9RoJyfeGbS7MG5FfuwQA4y9QOb/OQglq
+        ZffkVItwY52RKWb/b2WQmt+IcVax/j7DmBva765SIfPDvOCMrYhJBI/uYHQ0Zia7
+        SnC9+ez55wdYqgHkYojc21CIOnUvsPSj+rOpryoXzmcTuvKeVIyIA0h/mQyWjimR
+        ENrikC4+O8GBMY6V4uvS4EFhLfHE9g0D/20lNOKkpAKPenr8iAPWcl0/pijJCGxF
+        agnT7O2GQ9Lr5hSjW86agkevbGktu2ja5t/fHq0wpLQ4DVLMrR0/poaprTr307kW
+        AlQV3z/C2cMHNysz4ulOgQrudQbhUEz2A8nQxRtIfWunkEugKLr1QiCkE1LJW8Np
+        ZLxE6Qp0/KzdQva0HVNhbHQgR1BHIDxlcmlrQHNhbHRzdGFjay5jb20+iQFUBBMB
+        CAA+FiEE+AxQ1ELHGEyFTZPYw5x3k9EbHGsFAliKrcYCGwMFCQPCZwAFCwkIBwIG
+        FQgJCgsCBBYCAwECHgECF4AACgkQw5x3k9EbHGubUAf+PLdp1oTLVokockZgLyIQ
+        wxOd3ofNOgNk4QoAkSMNSbtnYoQFKumRw/yGyPSIoHMsOC/ga98r8TAJEKfx3DLA
+        rsD34oMAaYUT+XUd0KoSmlHqBrtDD1+eBASKYsCosHpCiKuQFfLKSxvpEr2YyL8L
+        X3Q2TY5zFlGA9Eeq5g+rlb++yRZrruFN28EWtY/pyXFZgIB30ReDwPkM9hrioPZM
+        0Qf3+dWZSK1rWViclB51oNy4un9stTiFZptAqz4NTNssU5A4AcNQPwBwnKIYoE58
+        Y/Zyv8HzILGykT+qFebqRlRBI/13eHdzgJOL1iPRfjTk5Cvr+vcyIxAklXOP81ja
+        B50DmARYiq3GAQgArnzu4SPCCQGNcCNxN4QlMP5TNvRsm5KrPbcO9j8HPfB+DRXs
+        6B3mnuR6OJg7YuC0C2A/m2dSHJKkF0f2AwFRpxLjJ2iAFbrZAW/N0vZDx8zO+YAU
+        HyLu0V04wdCE5DTLkgfWNR+0uMa8qZ4Kn56Gv7O+OFE7zgTHeZ7psWlxdafeW7u6
+        zlC/3DWksNtuNb0vQDNMM4vgXbnORIfXdyh41zvEEnr/rKw8DuJAmo20mcv6Qi51
+        PqqyM62ddQOEVfiMs9l4vmwZAjGFNFNInyPXnogL6UPCDmizb6hh8aX/MwG/XFIG
+        KMJWbAVGpyBuqljKIt3qLu/s8ouPqkEN+f+nGwARAQABAAf+NA36d/kieGxZpTQ1
+        oQHP1Jty+OiXhBwP8SPtF0J7ZxuZh07cs+zDsfBok/y6bsepfuFSaIq84OBQis+B
+        kajxkp3cXZPb7l+lQLv5k++7Dd7Ien+ewSE7TQN6HLwYATrM5n5nBcc1M5C6lQGc
+        mr0A5yz42TVG2bHsTpi9kBtsaVRSPUHSh8A8T6eOyCrT+/CAJVEEf7JyNyaqH1dy
+        LuxI1VF3ySDEtFzuwN8EZQP9Yz/4AVyEQEA7WkNEwSQsBi2bWgWEdG+qjqnL+YKa
+        vwe7/aJYPeL1zICnP/Osd/UcpDxR78MbozstbRljML0fTLj7UJ+XDazwv+Kl0193
+        2ZK2QQQAwgXvS19MYNkHO7kbNVLt1VE2ll901iC9GFHBpFUam6gmoHXpCarB+ShH
+        8x25aoUu4MxHmFxXd+Zq3d6q2yb57doWoPgvqcefpGmigaITnb1jhV2rt65V8deA
+        SQazZNqBEBbZNIhfn6ObxHXXvaYaqq/UOEQ7uKyR9WMJT/rmqMEEAOY5h1R1t7AB
+        JZ5VnhyAhdsNWw1gTcXB3o8gKz4vjdnPm0F4aVIPfB3BukETDc3sc2tKmCfUF7I7
+        oOrh7iRez5F0RIC3KDzXF8qUuWBfPViww45JgftdKsecCIlEEYCoc+3goX0su2bP
+        V1MDuHijMGTJCBABDgizNb0oynW5xcrbA/0QnKfpTwi7G3oRcJWv2YebVDRcU+SP
+        dOYhq6SnmWPizEIljRG/X7FHJB+W7tzryO3sCDTAYwxFrfMwvJ2PwnAYI4349zYd
+        lC28HowUkBYNhwBXc48xCfyhPZtD0aLx/OX1oLZ/vi8gd8TusgGupV/JjkFVO+Nd
+        +shN/UEAldwqkkY2iQE8BBgBCAAmFiEE+AxQ1ELHGEyFTZPYw5x3k9EbHGsFAliK
+        rcYCGwwFCQPCZwAACgkQw5x3k9EbHGu4wwf/dRFat91BRX1TJfwJl5otoAXpItYM
+        6kdWWf1Eb1BicAvXhI078MSH4WXdKkJjJr1fFP8Ynil513H4Mzb0rotMAhb0jLSA
+        lSRkMbhMvPxoS2kaYzioaBpp8yXpGiNo7dF+PJXSm/Uwp3AkcFjoVbBOqDWGgxMi
+        DvDAstzLZ9dIcmr+OmcRQykKOKXlhEl3HnR5CyuPrA8hdVup4oeVwdkJhfJFKLLb
+        3fR26wxJOmIOAt24eAUy721WfQ9txNAmhdy8mY842ODZESw6WatrQjRfuqosDgrk
+        jc0cCHsEqJNZ2AB+1uEl3tcH0tyAFJa33F0znSonP17SS1Ff9sgHYBVLUg==
+        =06Tz
+        -----END PGP PRIVATE KEY BLOCK-----
+        """
+    )
+
+
+@pytest.fixture(scope="module")
+def gpg_pillar_yaml():
+    """
+    Yaml data for testing GPG pillar.
+    """
+    return textwrap.dedent(
+        """
+        #!yaml|gpg
+        secrets:
+          foo: |
+            -----BEGIN PGP MESSAGE-----
+
+            hQEMAw2B674HRhwSAQgAhTrN8NizwUv/VunVrqa4/X8t6EUulrnhKcSeb8sZS4th
+            W1Qz3K2NjL4lkUHCQHKZVx/VoZY7zsddBIFvvoGGfj8+2wjkEDwFmFjGE4DEsS74
+            ZLRFIFJC1iB/O0AiQ+oU745skQkU6OEKxqavmKMrKo3rvJ8ZCXDC470+i2/Hqrp7
+            +KWGmaDOO422JaSKRm5D9bQZr9oX7KqnrPG9I1+UbJyQSJdsdtquPWmeIpamEVHb
+            VMDNQRjSezZ1yKC4kCWm3YQbBF76qTHzG1VlLF5qOzuGI9VkyvlMaLfMibriqY73
+            zBbPzf6Bkp2+Y9qyzuveYMmwS4sEOuZL/PetqisWe9JGAWD/O+slQ2KRu9hNww06
+            KMDPJRdyj5bRuBVE4hHkkP23KrYr7SuhW2vpe7O/MvWEJ9uDNegpMLhTWruGngJh
+            iFndxegN9w==
+            =bAuo
+            -----END PGP MESSAGE-----
+        """
+    )
+
+
+@pytest.fixture(scope="module")
+def gpg_homedir(salt_master, test_key):
+    """
+    Setup gpg environment
+    """
+    _gpg_homedir = pathlib.Path(salt_master.config_dir) / "gpgkeys"
+    _gpg_homedir.mkdir(0o700)
+    agent_started = False
+    try:
+        cmd_prefix = ["gpg", "--homedir", str(_gpg_homedir)]
+
+        cmd = cmd_prefix + ["--list-keys"]
+        proc = subprocess.run(
+            cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=True,
+            universal_newlines=True,
+        )
+        ret = ProcessResult(
+            returncode=proc.returncode,
+            stdout=proc.stdout,
+            stderr=proc.stderr or "",
+            cmdline=proc.args,
+        )
+        log.debug("Instantiating gpg keyring...\n%s", ret)
+
+        cmd = cmd_prefix + ["--import", "--allow-secret-key-import"]
+        proc = subprocess.run(
+            cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=True,
+            universal_newlines=True,
+            input=test_key,
+        )
+        ret = ProcessResult(
+            returncode=proc.returncode,
+            stdout=proc.stdout,
+            stderr=proc.stderr or "",
+            cmdline=proc.args,
+        )
+        log.debug("Importing keypair...:\n%s", ret)
+
+        agent_started = True
+
+        yield _gpg_homedir
+    finally:
+        if agent_started:
+            try:
+                cmd = ["gpg-connect-agent", "--homedir", str(_gpg_homedir)]
+                proc = subprocess.run(
+                    cmd,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.STDOUT,
+                    check=True,
+                    universal_newlines=True,
+                    input="KILLAGENT",
+                )
+                ret = ProcessResult(
+                    returncode=proc.returncode,
+                    stdout=proc.stdout,
+                    stderr=proc.stderr or "",
+                    cmdline=proc.args,
+                )
+                log.debug("Killed gpg-agent...\n%s", ret)
+            except (OSError, subprocess.CalledProcessError):
+                log.debug("No need to kill: old gnupg doesn't start the agent.")
+        shutil.rmtree(str(_gpg_homedir), ignore_errors=True)
+
+
+@pytest.fixture(scope="module")
+def pillar_setup(base_env_pillar_tree_root_dir, gpg_pillar_yaml, salt_minion):
+    """
+    Setup gpg pillar
+    """
+    saltutil_contents = f"""
+    saltutil: {{{{ salt["saltutil.runner"]("mine.get", tgt="{salt_minion.id}", fun="test.ping") | json }}}}
+    """
+    top_file_contents = """
+    base:
+      '*':
+        - gpg
+        - saltutil
+    """
+    with pytest.helpers.temp_file(
+        "top.sls", top_file_contents, base_env_pillar_tree_root_dir
+    ), pytest.helpers.temp_file(
+        "gpg.sls", gpg_pillar_yaml, base_env_pillar_tree_root_dir
+    ), pytest.helpers.temp_file(
+        "saltutil.sls", saltutil_contents, base_env_pillar_tree_root_dir
+    ):
+        yield
+
+
+@pytest.mark.skip_if_binaries_missing("gpg")
+@pytest.mark.usefixtures("pillar_setup", "gpg_homedir")
+def test_gpg_pillar(salt_ssh_cli):
+    """
+    Ensure that GPG-encrypted pillars can be decrypted, i.e. the
+    gpg_keydir should not be overridden. This is issue #60002,
+    which has the same cause as the one below.
+    """
+    ret = salt_ssh_cli.run("pillar.items")
+    assert ret.returncode == 0
+    assert isinstance(ret.data, dict)
+    assert ret.data
+    assert "secrets" in ret.data
+    assert "foo" in ret.data["secrets"]
+    assert "BEGIN PGP MESSAGE" not in ret.data["secrets"]["foo"]
+    assert ret.data["secrets"]["foo"] == "supersecret"
+    assert "_errors" not in ret.data
+
+
+@pytest.mark.usefixtures("pillar_setup")
+def test_saltutil_runner(salt_ssh_cli, salt_minion, salt_run_cli):
+    """
+    Ensure that during pillar compilation, the cache dir is not
+    overridden. For a history, see PR #50489 and issue #36796,
+    notice that the initial description is probably unrelated
+    to this.
+    """
+    ret = salt_ssh_cli.run("pillar.items")
+    assert ret.returncode == 0
+    assert isinstance(ret.data, dict)
+    assert ret.data
+    assert "saltutil" in ret.data
+    assert isinstance(ret.data["saltutil"], dict)
+    assert ret.data["saltutil"]
+    assert salt_minion.id in ret.data["saltutil"]
+    assert ret.data["saltutil"][salt_minion.id] is True
+    assert "_errors" not in ret.data
diff --git a/tests/pytests/integration/ssh/test_state.py b/tests/pytests/integration/ssh/test_state.py
deleted file mode 100644
index 5f9bfb45e9f4..000000000000
--- a/tests/pytests/integration/ssh/test_state.py
+++ /dev/null
@@ -1,563 +0,0 @@
-import json
-
-import pytest
-
-from salt.defaults.exitcodes import EX_AGGREGATE
-
-pytestmark = [
-    pytest.mark.skip_on_windows(reason="salt-ssh not available on Windows"),
-]
-
-
-@pytest.fixture(scope="module")
-def state_tree(base_env_state_tree_root_dir):
-    top_file = """
-    {%- from "map.jinja" import abc with context %}
-    base:
-      'localhost':
-        - basic
-      '127.0.0.1':
-        - basic
-    """
-    map_file = """
-    {%- set abc = "def" %}
-    """
-    state_file = """
-    {%- from "map.jinja" import abc with context %}
-    Ok with {{ abc }}:
-      test.succeed_without_changes
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    map_tempfile = pytest.helpers.temp_file(
-        "map.jinja", map_file, base_env_state_tree_root_dir
-    )
-    state_tempfile = pytest.helpers.temp_file(
-        "test.sls", state_file, base_env_state_tree_root_dir
-    )
-    with top_tempfile, map_tempfile, state_tempfile:
-        yield
-
-
-@pytest.fixture(scope="module")
-def state_tree_dir(base_env_state_tree_root_dir):
-    """
-    State tree with files to test salt-ssh
-    when the map.jinja file is in another directory
-    """
-    top_file = """
-    {%- from "test/map.jinja" import abc with context %}
-    base:
-      'localhost':
-        - test
-      '127.0.0.1':
-        - test
-    """
-    map_file = """
-    {%- set abc = "def" %}
-    """
-    state_file = """
-    {%- from "test/map.jinja" import abc with context %}
-
-    Ok with {{ abc }}:
-      test.succeed_without_changes
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    map_tempfile = pytest.helpers.temp_file(
-        "test/map.jinja", map_file, base_env_state_tree_root_dir
-    )
-    state_tempfile = pytest.helpers.temp_file(
-        "test.sls", state_file, base_env_state_tree_root_dir
-    )
-
-    with top_tempfile, map_tempfile, state_tempfile:
-        yield
-
-
-@pytest.fixture(scope="class")
-def state_tree_render_fail(base_env_state_tree_root_dir):
-    top_file = """
-    base:
-      'localhost':
-        - fail_render
-      '127.0.0.1':
-        - fail_render
-    """
-    state_file = r"""
-    abc var is not defined {{ abc }}:
-      test.nop
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    state_tempfile = pytest.helpers.temp_file(
-        "fail_render.sls", state_file, base_env_state_tree_root_dir
-    )
-    with top_tempfile, state_tempfile:
-        yield
-
-
-@pytest.fixture(scope="class")
-def state_tree_req_fail(base_env_state_tree_root_dir):
-    top_file = """
-    base:
-      'localhost':
-        - fail_req
-      '127.0.0.1':
-        - fail_req
-    """
-    state_file = """
-    This has an invalid requisite:
-      test.nop:
-        - name: foo
-        - require_in:
-          - file.managed: invalid_requisite
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    state_tempfile = pytest.helpers.temp_file(
-        "fail_req.sls", state_file, base_env_state_tree_root_dir
-    )
-    with top_tempfile, state_tempfile:
-        yield
-
-
-@pytest.fixture(scope="class")
-def state_tree_structure_fail(base_env_state_tree_root_dir):
-    top_file = """
-    base:
-      'localhost':
-        - fail_structure
-      '127.0.0.1':
-        - fail_structure
-    """
-    state_file = """
-    extend:
-      Some file state:
-        file:
-            - name: /tmp/bar
-            - contents: bar
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    state_tempfile = pytest.helpers.temp_file(
-        "fail_structure.sls", state_file, base_env_state_tree_root_dir
-    )
-    with top_tempfile, state_tempfile:
-        yield
-
-
-@pytest.fixture(scope="class")
-def state_tree_run_fail(base_env_state_tree_root_dir):
-    top_file = """
-    base:
-      'localhost':
-        - fail_run
-      '127.0.0.1':
-        - fail_run
-    """
-    state_file = """
-    This file state fails:
-      file.managed:
-        - name: /tmp/non/ex/is/tent
-        - makedirs: false
-        - contents: foo
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    state_tempfile = pytest.helpers.temp_file(
-        "fail_run.sls", state_file, base_env_state_tree_root_dir
-    )
-    with top_tempfile, state_tempfile:
-        yield
-
-
-@pytest.fixture(scope="class")
-def pillar_tree_render_fail(base_env_pillar_tree_root_dir):
-    top_file = """
-    base:
-      'localhost':
-        - fail_render
-      '127.0.0.1':
-        - fail_render
-    """
-    pillar_file = r"""
-    not_defined: {{ abc }}
-    """
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_pillar_tree_root_dir
-    )
-    pillar_tempfile = pytest.helpers.temp_file(
-        "fail_render.sls", pillar_file, base_env_pillar_tree_root_dir
-    )
-    with top_tempfile, pillar_tempfile:
-        yield
-
-
-@pytest.mark.slow_test
-def test_state_with_import(salt_ssh_cli, state_tree):
-    """
-    verify salt-ssh can use imported map files in states
-    """
-    ret = salt_ssh_cli.run("state.sls", "test")
-    assert ret.returncode == 0
-    assert ret.data
-
-
-@pytest.mark.parametrize(
-    "ssh_cmd",
-    [
-        "state.sls",
-        "state.highstate",
-        "state.apply",
-        "state.show_top",
-        "state.show_highstate",
-        "state.show_low_sls",
-        "state.show_lowstate",
-        "state.sls_id",
-        "state.show_sls",
-        "state.top",
-    ],
-)
-@pytest.mark.slow_test
-def test_state_with_import_dir(salt_ssh_cli, state_tree_dir, ssh_cmd):
-    """
-    verify salt-ssh can use imported map files in states
-    when the map files are in another directory outside of
-    sls files importing them.
-    """
-    if ssh_cmd in ("state.sls", "state.show_low_sls", "state.show_sls"):
-        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd, "test")
-    elif ssh_cmd == "state.top":
-        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd, "top.sls")
-    elif ssh_cmd == "state.sls_id":
-        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd, "Ok with def", "test")
-    else:
-        ret = salt_ssh_cli.run("-w", "-t", ssh_cmd)
-    assert ret.returncode == 0
-    if ssh_cmd == "state.show_top":
-        assert ret.data == {"base": ["test", "master_tops_test"]} or {"base": ["test"]}
-    elif ssh_cmd in ("state.show_highstate", "state.show_sls"):
-        assert ret.data == {
-            "Ok with def": {
-                "__sls__": "test",
-                "__env__": "base",
-                "test": ["succeed_without_changes", {"order": 10000}],
-            }
-        }
-    elif ssh_cmd in ("state.show_low_sls", "state.show_lowstate", "state.show_sls"):
-        assert ret.data == [
-            {
-                "state": "test",
-                "name": "Ok with def",
-                "__sls__": "test",
-                "__env__": "base",
-                "__id__": "Ok with def",
-                "order": 10000,
-                "fun": "succeed_without_changes",
-            }
-        ]
-    else:
-        assert ret.data["test_|-Ok with def_|-Ok with def_|-succeed_without_changes"][
-            "result"
-        ]
-    assert ret.data
-
-
-@pytest.fixture
-def nested_state_tree(base_env_state_tree_root_dir, tmp_path):
-    top_file = """
-    base:
-      'localhost':
-        - basic
-      '127.0.0.1':
-        - basic
-    """
-    state_file = """
-    /{}/file.txt:
-      file.managed:
-        - source: salt://foo/file.jinja
-        - template: jinja
-    """.format(
-        tmp_path
-    )
-    file_jinja = """
-    {% from 'foo/map.jinja' import comment %}{{ comment }}
-    """
-    map_file = """
-    {% set comment = "blah blah" %}
-    """
-    statedir = base_env_state_tree_root_dir / "foo"
-    top_tempfile = pytest.helpers.temp_file(
-        "top.sls", top_file, base_env_state_tree_root_dir
-    )
-    map_tempfile = pytest.helpers.temp_file("map.jinja", map_file, statedir)
-    file_tempfile = pytest.helpers.temp_file("file.jinja", file_jinja, statedir)
-    state_tempfile = pytest.helpers.temp_file("init.sls", state_file, statedir)
-
-    with top_tempfile, map_tempfile, state_tempfile, file_tempfile:
-        yield
-
-
-@pytest.mark.slow_test
-def test_state_with_import_from_dir(salt_ssh_cli, nested_state_tree):
-    """
-    verify salt-ssh can use imported map files in states
-    """
-    ret = salt_ssh_cli.run(
-        "--extra-filerefs=salt://foo/map.jinja", "state.apply", "foo"
-    )
-    assert ret.returncode == 0
-    assert ret.data
-
-
-@pytest.mark.slow_test
-def test_state_low(salt_ssh_cli):
-    """
-    test state.low with salt-ssh
-    """
-    ret = salt_ssh_cli.run(
-        "state.low", '{"state": "cmd", "fun": "run", "name": "echo blah"}'
-    )
-    assert (
-        json.loads(ret.stdout)["localhost"]["cmd_|-echo blah_|-echo blah_|-run"][
-            "changes"
-        ]["stdout"]
-        == "blah"
-    )
-
-
-@pytest.mark.slow_test
-def test_state_high(salt_ssh_cli):
-    """
-    test state.high with salt-ssh
-    """
-    ret = salt_ssh_cli.run("state.high", '{"echo blah": {"cmd": ["run"]}}')
-    assert (
-        json.loads(ret.stdout)["localhost"]["cmd_|-echo blah_|-echo blah_|-run"][
-            "changes"
-        ]["stdout"]
-        == "blah"
-    )
-
-
-@pytest.mark.slow_test
-@pytest.mark.usefixtures("state_tree_render_fail")
-class TestRenderExceptionRetcode:
-    """
-    Verify salt-ssh fails with a retcode > 0 when a state rendering fails.
-    """
-
-    def test_retcode_state_sls_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls", "fail_render")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_highstate_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.highstate")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_sls_id_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls_id", "foo", "fail_render")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_sls_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_sls", "fail_render")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_low_sls_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_low_sls", "fail_render")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_highstate_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_highstate")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_lowstate_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_lowstate")
-        # state.show_lowstate exits with 0 for non-ssh as well
-        self._assert_ret(ret, 0)
-
-    def test_retcode_state_top_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.top", "top.sls")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_single_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.single", "file")
-        assert ret.returncode == EX_AGGREGATE
-        assert isinstance(ret.data, str)
-        assert "single() missing 1 required positional argument" in ret.data
-
-    def _assert_ret(self, ret, retcode):
-        assert ret.returncode == retcode
-        assert isinstance(ret.data, list)
-        assert ret.data
-        assert isinstance(ret.data[0], str)
-        assert ret.data[0].startswith(
-            "Rendering SLS 'base:fail_render' failed: Jinja variable 'abc' is undefined;"
-        )
-
-
-@pytest.mark.slow_test
-@pytest.mark.usefixtures("pillar_tree_render_fail")
-class TestPillarRenderExceptionRetcode:
-    """
-    Verify salt-ssh fails with a retcode > 0 when a pillar rendering fails.
-    """
-
-    def test_retcode_state_sls_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls", "basic")
-        self._assert_ret(ret)
-
-    def test_retcode_state_highstate_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.highstate")
-        self._assert_ret(ret)
-
-    def test_retcode_state_sls_id_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls_id", "foo", "basic")
-        self._assert_ret(ret)
-
-    def test_retcode_state_show_sls_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_sls", "basic")
-        self._assert_ret(ret)
-
-    def test_retcode_state_show_low_sls_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_low_sls", "basic")
-        self._assert_ret(ret)
-
-    def test_retcode_state_show_highstate_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_highstate")
-        self._assert_ret(ret)
-
-    def test_retcode_state_show_lowstate_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_lowstate")
-        self._assert_ret(ret)
-
-    def test_retcode_state_top_pillar_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.top", "top.sls")
-        self._assert_ret(ret)
-
-    def _assert_ret(self, ret):
-        assert ret.returncode == EX_AGGREGATE
-        assert isinstance(ret.data, list)
-        assert ret.data
-        assert isinstance(ret.data[0], str)
-        assert ret.data[0] == "Pillar failed to render with the following messages:"
-        assert ret.data[1].startswith("Rendering SLS 'fail_render' failed.")
-
-
-@pytest.mark.slow_test
-@pytest.mark.usefixtures("state_tree_req_fail")
-class TestStateReqFailRetcode:
-    """
-    Verify salt-ssh fails with a retcode > 0 when a highstate verification fails.
-    ``state.show_highstate`` does not validate this.
-    """
-
-    def test_retcode_state_sls_invalid_requisite(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls", "fail_req")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_highstate_invalid_requisite(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.highstate")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_sls_invalid_requisite(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_sls", "fail_req")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_low_sls_invalid_requisite(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_low_sls", "fail_req")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_lowstate_invalid_requisite(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_lowstate")
-        # state.show_lowstate exits with 0 for non-ssh as well
-        self._assert_ret(ret, 0)
-
-    def test_retcode_state_top_invalid_requisite(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.top", "top.sls")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def _assert_ret(self, ret, retcode):
-        assert ret.returncode == retcode
-        assert isinstance(ret.data, list)
-        assert ret.data
-        assert isinstance(ret.data[0], str)
-        assert ret.data[0].startswith(
-            "Invalid requisite in require: file.managed for invalid_requisite"
-        )
-
-
-@pytest.mark.slow_test
-@pytest.mark.usefixtures("state_tree_structure_fail")
-class TestStateStructureFailRetcode:
-    """
-    Verify salt-ssh fails with a retcode > 0 when a highstate verification fails.
-    This targets another step of the verification.
-    ``state.sls_id`` does not seem to support extends.
-    ``state.show_highstate`` does not validate this.
-    """
-
-    def test_retcode_state_sls_invalid_structure(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls", "fail_structure")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_highstate_invalid_structure(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.highstate")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_sls_invalid_structure(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_sls", "fail_structure")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_low_sls_invalid_structure(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_low_sls", "fail_structure")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def test_retcode_state_show_lowstate_invalid_structure(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.show_lowstate")
-        # state.show_lowstate exits with 0 for non-ssh as well
-        self._assert_ret(ret, 0)
-
-    def test_retcode_state_top_invalid_structure(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.top", "top.sls")
-        self._assert_ret(ret, EX_AGGREGATE)
-
-    def _assert_ret(self, ret, retcode):
-        assert ret.returncode == retcode
-        assert isinstance(ret.data, list)
-        assert ret.data
-        assert isinstance(ret.data[0], str)
-        assert ret.data[0].startswith(
-            "Cannot extend ID 'Some file state' in 'base:fail_structure"
-        )
-
-
-@pytest.mark.slow_test
-@pytest.mark.usefixtures("state_tree_run_fail")
-class TestStateRunFailRetcode:
-    """
-    Verify salt-ssh passes on a failing retcode from state execution.
-    """
-
-    def test_retcode_state_sls_run_fail(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls", "fail_run")
-        assert ret.returncode == EX_AGGREGATE
-
-    def test_retcode_state_highstate_run_fail(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.highstate")
-        assert ret.returncode == EX_AGGREGATE
-
-    def test_retcode_state_sls_id_render_exception(self, salt_ssh_cli):
-        ret = salt_ssh_cli.run("state.sls_id", "This file state fails", "fail_run")
-        assert ret.returncode == EX_AGGREGATE
-
-    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