text: import glob
  import logging
  import os
  import shutil
  import threading
  import time
  
  import pytest
  from tests.support.case import SSHCase
  from tests.support.helpers import flaky
  from tests.support.pytest.helpers import temp_state_file
  from tests.support.runtests import RUNTIME_VARS
  
  SSH_SLS = "ssh_state_tests"
  SSH_SLS_FILE = "/tmp/salt_test_file"
  
  log = logging.getLogger(__name__)
  
  
  class SSHStateTest(SSHCase):
      """
      testing the state system with salt-ssh
      """
  
      def _check_dict_ret(self, ret, val, exp_ret, equal=True):
          self.assertIsInstance(ret, dict)
          for key, value in ret.items():
              self.assertIsInstance(value, dict)
              if equal:
                  self.assertEqual(value[val], exp_ret)
              else:
                  self.assertNotEqual(value[val], exp_ret)
  
      def _check_request(self, empty=False):
          check = self.run_function("state.check_request", wipe=False)
          if empty:
              self.assertFalse(bool(check), "bool({}) is not False".format(check))
          else:
              self._check_dict_ret(
                  ret=check["default"]["test_run"]["local"]["return"],
                  val="__sls__",
                  exp_ret=SSH_SLS,
              )
  
      @pytest.mark.slow_test
      def test_state_apply(self):
          """
          test state.apply with salt-ssh
          """
          ret = self.run_function("state.apply", [SSH_SLS])
          self._check_dict_ret(ret=ret, val="__sls__", exp_ret=SSH_SLS)
  
          check_file = self.run_function("file.file_exists", [SSH_SLS_FILE])
          self.assertTrue(check_file)
  
      @pytest.mark.slow_test
      def test_state_sls_id(self):
          """
          test state.sls_id with salt-ssh
          """
          # check state.sls_id with test=True
          ret = self.run_function("state.sls_id", ["ssh-file-test", SSH_SLS, "test=True"])
          self._check_dict_ret(
              ret=ret,
              val="comment",
              exp_ret=(
                  "The file {} is set to be changed\n"
                  "Note: No changes made, actual changes may\n"
                  "be different due to other states."
              ).format(SSH_SLS_FILE),
          )
  
          # check state.sls_id without test=True
          ret = self.run_function("state.sls_id", ["ssh-file-test", SSH_SLS])
          self._check_dict_ret(ret=ret, val="__sls__", exp_ret=SSH_SLS)
  
          # make sure the other id in the state was not run
          self._check_dict_ret(ret=ret, val="__id__", exp_ret="second_id", equal=False)
  
          check_file = self.run_function("file.file_exists", [SSH_SLS_FILE])
          self.assertTrue(check_file)
  
      @pytest.mark.slow_test
      def test_state_sls_wrong_id(self):
          """
          test state.sls_id when id does not exist
          """
          # check state.sls_id with test=True
          ret = self.run_function("state.sls_id", ["doesnotexist", SSH_SLS])
          assert "No matches for ID" in ret
  
      @pytest.mark.slow_test
      def test_state_sls_id_with_pillar(self):
          """
          test state.sls_id with pillar data
          """
          self.run_function(
              "state.sls_id",
              ["ssh-file-test", SSH_SLS, 'pillar=\'{"test_file_suffix": "_pillar"}\''],
          )
          check_file = self.run_function(
              "file.file_exists", ["/tmp/salt_test_file_pillar"]
          )
          self.assertTrue(check_file)
  
      @pytest.mark.slow_test
      def test_state_show_sls(self):
          """
          test state.show_sls with salt-ssh
          """
          ret = self.run_function("state.show_sls", [SSH_SLS])
          self._check_dict_ret(ret=ret, val="__sls__", exp_ret=SSH_SLS)
  
          check_file = self.run_function("file.file_exists", [SSH_SLS_FILE], wipe=False)
          self.assertFalse(check_file)
  
      @pytest.mark.slow_test
      def test_state_show_top(self):
          """
          test state.show_top with salt-ssh
          """
          top_sls = """
          base:
            '*':
              - core
              """
  
          core_state = """
          {}/testfile:
            file:
              - managed
              - source: salt://testfile
              - makedirs: true
              """.format(
              RUNTIME_VARS.TMP
          )
  
          with temp_state_file("top.sls", top_sls), temp_state_file(
              "core.sls", core_state
          ):
              ret = self.run_function("state.show_top")
              self.assertEqual(ret, {"base": ["core", "master_tops_test"]})
  
      @pytest.mark.slow_test
      def test_state_single(self):
          """
          state.single with salt-ssh
          """
          ret_out = {"name": "itworked", "result": True, "comment": "Success!"}
  
          single = self.run_function(
              "state.single", ["test.succeed_with_changes name=itworked"]
          )
  
          self.assertIsInstance(single, dict)
          for key, value in single.items():
              self.assertIsInstance(value, dict)
              self.assertEqual(value["name"], ret_out["name"])
              self.assertEqual(value["result"], ret_out["result"])
              self.assertEqual(value["comment"], ret_out["comment"])
  
      @pytest.mark.slow_test
      def test_show_highstate(self):
          """
          state.show_highstate with salt-ssh
          """
          top_sls = """
          base:
            '*':
              - core
              """
  
          core_state = """
          {}/testfile:
            file:
              - managed
              - source: salt://testfile
              - makedirs: true
              """.format(
              RUNTIME_VARS.TMP
          )
  
          with temp_state_file("top.sls", top_sls), temp_state_file(
              "core.sls", core_state
          ):
              high = self.run_function("state.show_highstate")
              destpath = os.path.join(RUNTIME_VARS.TMP, "testfile")
              self.assertIsInstance(high, dict)
              self.assertIn(destpath, high)
              self.assertEqual(high[destpath]["__env__"], "base")
  
      @pytest.mark.slow_test
      def test_state_high(self):
          """
          state.high with salt-ssh
          """
          ret_out = {"name": "itworked", "result": True, "comment": "Success!"}
  
          high = self.run_function(
              "state.high", ['"{"itworked": {"test": ["succeed_with_changes"]}}"']
          )
  
          self.assertIsInstance(high, dict)
          for key, value in high.items():
              self.assertIsInstance(value, dict)
              self.assertEqual(value["name"], ret_out["name"])
              self.assertEqual(value["result"], ret_out["result"])
              self.assertEqual(value["comment"], ret_out["comment"])
  
      @pytest.mark.slow_test
      def test_show_lowstate(self):
          """
          state.show_lowstate with salt-ssh
          """
          top_sls = """
          base:
            '*':
              - core
              """
  
          core_state = """
          {}/testfile:
            file:
              - managed
              - source: salt://testfile
              - makedirs: true
              """.format(
              RUNTIME_VARS.TMP
          )
  
          with temp_state_file("top.sls", top_sls), temp_state_file(
              "core.sls", core_state
          ):
              low = self.run_function("state.show_lowstate")
              self.assertIsInstance(low, list)
              self.assertIsInstance(low[0], dict)
  
      @pytest.mark.slow_test
      def test_state_low(self):
          """
          state.low with salt-ssh
          """
          ret_out = {"name": "itworked", "result": True, "comment": "Success!"}
  
          low = self.run_function(
              "state.low",
              ['"{"state": "test", "fun": "succeed_with_changes", "name": "itworked"}"'],
          )
  
          self.assertIsInstance(low, dict)
          for key, value in low.items():
              self.assertIsInstance(value, dict)
              self.assertEqual(value["name"], ret_out["name"])
              self.assertEqual(value["result"], ret_out["result"])
              self.assertEqual(value["comment"], ret_out["comment"])
  
      @pytest.mark.slow_test
      def test_state_request_check_clear(self):
          """
          test state.request system with salt-ssh
          while also checking and clearing request
          """
          request = self.run_function("state.request", [SSH_SLS], wipe=False)
          self._check_dict_ret(ret=request, val="__sls__", exp_ret=SSH_SLS)
  
          self._check_request()
  
          clear = self.run_function("state.clear_request", wipe=False)
          self._check_request(empty=True)
  
      @pytest.mark.slow_test
      def test_state_run_request(self):
          """
          test state.request system with salt-ssh
          while also running the request later
          """
          request = self.run_function("state.request", [SSH_SLS], wipe=False)
          self._check_dict_ret(ret=request, val="__sls__", exp_ret=SSH_SLS)
  
          run = self.run_function("state.run_request", wipe=False)
  
          check_file = self.run_function("file.file_exists", [SSH_SLS_FILE], wipe=False)
          self.assertTrue(check_file)
  
      @flaky
      @pytest.mark.slow_test
      def test_state_running(self):
          """
          test state.running with salt-ssh
          """
  
          def _run_in_background():
              self.run_function("state.sls", ["running"], wipe=False)
  
          bg_thread = threading.Thread(target=_run_in_background)
          bg_thread.start()
  
          expected = 'The function "state.pkg" is running as'
          state_ret = []
          for _ in range(30):
              time.sleep(5)
              get_sls = self.run_function("state.running", wipe=False)
              state_ret.append(get_sls)
              if expected in " ".join(get_sls):
                  # We found the expected return
                  break
          else:
              self.fail(
                  "Did not find '{}' in state.running return: {}".format(
                      expected, state_ret
                  )
              )
  
          # make sure we wait until the earlier state is complete
          future = time.time() + 120
          while True:
              if expected not in " ".join(self.run_function("state.running", wipe=False)):
                  break
              if time.time() > future:
                  self.fail(
                      "state.pkg is still running overtime. Test did not clean up correctly."
                  )
  
      def tearDown(self):
          """
          make sure to clean up any old ssh directories
          """
          salt_dir = self.run_function("config.get", ["thin_dir"], wipe=False)
          self.assertIsInstance(salt_dir, (str,))
          if os.path.exists(salt_dir):
              shutil.rmtree(salt_dir)
  
          for test_file_path in glob.glob(SSH_SLS_FILE + "*"):
              os.remove(test_file_path)
  