text: # -*- coding: utf-8 -*-
  '''
      :codeauthor: Pedro Algarvio (pedro@algarvio.me)
  
  
      tests.integration.states.pip_state
      ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  '''
  
  # Import python libs
  from __future__ import absolute_import, print_function, unicode_literals
  import errno
  import os
  import glob
  import pprint
  import shutil
  import sys
  
  try:
      import pwd
      HAS_PWD = True
  except ImportError:
      HAS_PWD = False
  
  # Import Salt Testing libs
  from tests.support.case import ModuleCase
  from tests.support.helpers import (
      destructiveTest,
      requires_system_grains,
      with_system_user,
      skip_if_not_root,
      with_tempdir,
      patched_environ
  )
  from tests.support.mixins import SaltReturnAssertsMixin
  from tests.support.runtests import RUNTIME_VARS
  from tests.support.unit import skipIf
  
  # Import salt libs
  import salt.utils.files
  import salt.utils.path
  import salt.utils.platform
  import salt.utils.versions
  import salt.utils.win_dacl
  import salt.utils.win_functions
  import salt.utils.win_runas
  from salt.modules.virtualenv_mod import KNOWN_BINARY_NAMES
  from salt.exceptions import CommandExecutionError
  
  # Import 3rd-party libs
  from salt.ext import six
  
  
  class VirtualEnv(object):
      def __init__(self, test, venv_dir):
          self.venv_dir = venv_dir
          self.test = test
          self.test.addCleanup(shutil.rmtree, self.venv_dir, ignore_errors=True)
  
      def __enter__(self):
          ret = self.test._create_virtualenv(self.venv_dir)
          self.test.assertEqual(
              ret['retcode'], 0,
              msg='Expected \'retcode\' key did not match. Full return dictionary:\n{}'.format(
                  pprint.pformat(ret)
              )
          )
  
      def __exit__(self, *args):
          pass
  
  
  @skipIf(salt.utils.path.which_bin(KNOWN_BINARY_NAMES) is None, 'virtualenv not installed')
  class PipStateTest(ModuleCase, SaltReturnAssertsMixin):
  
      def _create_virtualenv(self, path, **kwargs):
          '''
          The reason why the virtualenv creation is proxied by this function is mostly
          because under windows, we can't seem to properly create a virtualenv off of
          another virtualenv(we can on linux) and also because, we really don't want to
          test virtualenv creation off of another virtualenv, we want a virtualenv created
          from the original python.
          Also, one windows, we must also point to the virtualenv binary outside the existing
          virtualenv because it will fail otherwise
          '''
          self.addCleanup(shutil.rmtree, path, ignore_errors=True)
          try:
              if salt.utils.platform.is_windows():
                  python = os.path.join(sys.real_prefix, os.path.basename(sys.executable))
              else:
                  python_binary_names = [
                      'python{}.{}'.format(*sys.version_info),
                      'python{}'.format(*sys.version_info),
                      'python'
                  ]
                  for binary_name in python_binary_names:
                      python = os.path.join(sys.real_prefix, 'bin', binary_name)
                      if os.path.exists(python):
                          break
                  else:
                      self.fail(
                          'Couldn\'t find a python binary name under \'{}\' matching: {}'.format(
                              os.path.join(sys.real_prefix, 'bin'),
                              python_binary_names
                          )
                      )
              # We're running off a virtualenv, and we don't want to create a virtualenv off of
              # a virtualenv, let's point to the actual python that created the virtualenv
              kwargs['python'] = python
          except AttributeError:
              # We're running off of the system python
              pass
          return self.run_function('virtualenv.create', [path], **kwargs)
  
      def test_pip_installed_removed(self):
          '''
          Tests installed and removed states
          '''
          name = 'pudb'
          if name in self.run_function('pip.list'):
              self.skipTest('{0} is already installed, uninstall to run this test'.format(name))
          ret = self.run_state('pip.installed', name=name)
          self.assertSaltTrueReturn(ret)
          ret = self.run_state('pip.removed', name=name)
          self.assertSaltTrueReturn(ret)
  
      def test_pip_installed_removed_venv(self):
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, 'pip_installed_removed'
          )
          with VirtualEnv(self, venv_dir):
              name = 'pudb'
              ret = self.run_state('pip.installed', name=name, bin_env=venv_dir)
              self.assertSaltTrueReturn(ret)
              ret = self.run_state('pip.removed', name=name, bin_env=venv_dir)
              self.assertSaltTrueReturn(ret)
  
      def test_pip_installed_errors(self):
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, 'pip-installed-errors'
          )
          self.addCleanup(shutil.rmtree, venv_dir, ignore_errors=True)
          # Since we don't have the virtualenv created, pip.installed will
          # throw an error.
          # Example error strings:
          #  * "Error installing 'pep8': /tmp/pip-installed-errors: not found"
          #  * "Error installing 'pep8': /bin/sh: 1: /tmp/pip-installed-errors: not found"
          #  * "Error installing 'pep8': /bin/bash: /tmp/pip-installed-errors: No such file or directory"
          with patched_environ(SHELL='/bin/sh'):
              ret = self.run_function('state.sls', mods='pip-installed-errors')
              self.assertSaltFalseReturn(ret)
              self.assertSaltCommentRegexpMatches(
                  ret,
                  'Error installing \'pep8\':'
              )
  
              # We now create the missing virtualenv
              ret = self.run_function('virtualenv.create', [venv_dir])
              self.assertEqual(ret['retcode'], 0)
  
              # The state should not have any issues running now
              ret = self.run_function('state.sls', mods='pip-installed-errors')
              self.assertSaltTrueReturn(ret)
  
      @skipIf(six.PY3, 'Issue is specific to carbon module, which is PY2-only')
      @skipIf(salt.utils.platform.is_windows(), "Carbon does not install in Windows")
      @requires_system_grains
      def test_pip_installed_weird_install(self, grains=None):
          # First, check to see if this is running on CentOS 5 or MacOS.
          # If so, skip this test.
          if grains['os'] in ('CentOS',) and grains['osrelease_info'][0] in (5,):
              self.skipTest('This test does not run reliably on CentOS 5')
          if grains['os'] in ('MacOS',):
              self.skipTest('This test does not run reliably on MacOS')
  
          ographite = '/opt/graphite'
          if os.path.isdir(ographite):
              self.skipTest(
                  'You already have \'{0}\'. This test would overwrite this '
                  'directory'.format(ographite)
              )
          try:
              os.makedirs(ographite)
          except OSError as err:
              if err.errno == errno.EACCES:
                  # Permission denied
                  self.skipTest(
                      'You don\'t have the required permissions to run this test'
                  )
          finally:
              if os.path.isdir(ographite):
                  shutil.rmtree(ographite, ignore_errors=True)
  
          venv_dir = os.path.join(RUNTIME_VARS.TMP, 'pip-installed-weird-install')
          try:
              # We may be able to remove this, I had to add it because the custom
              # modules from the test suite weren't available in the jinja
              # context when running the call to state.sls that comes after.
              self.run_function('saltutil.sync_modules')
              # Since we don't have the virtualenv created, pip.installed will
              # throw an error.
              ret = self.run_function(
                  'state.sls', mods='pip-installed-weird-install'
              )
              self.assertSaltTrueReturn(ret)
  
              # We cannot use assertInSaltComment here because we need to skip
              # some of the state return parts
              for key in six.iterkeys(ret):
                  self.assertTrue(ret[key]['result'])
                  if ret[key]['name'] != 'carbon < 1.1':
                      continue
                  self.assertEqual(
                      ret[key]['comment'],
                      'There was no error installing package \'carbon < 1.1\' '
                      'although it does not show when calling \'pip.freeze\'.'
                  )
                  break
              else:
                  raise Exception('Expected state did not run')
          finally:
              if os.path.isdir(ographite):
                  shutil.rmtree(ographite, ignore_errors=True)
  
      def test_issue_2028_pip_installed_state(self):
          ret = self.run_function('state.sls', mods='issue-2028-pip-installed')
  
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, 'issue-2028-pip-installed'
          )
          self.addCleanup(shutil.rmtree, venv_dir, ignore_errors=True)
  
          pep8_bin = os.path.join(venv_dir, 'bin', 'pep8')
          if salt.utils.platform.is_windows():
              pep8_bin = os.path.join(venv_dir, 'Scripts', 'pep8.exe')
  
          self.assertSaltTrueReturn(ret)
          self.assertTrue(
              os.path.isfile(pep8_bin)
          )
  
      def test_issue_2087_missing_pip(self):
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, 'issue-2087-missing-pip'
          )
  
          # Let's create the testing virtualenv
          ret = self._create_virtualenv(venv_dir)
          self.assertEqual(
              ret['retcode'], 0,
              msg='Expected \'retcode\' key did not match. Full return dictionary:\n{}'.format(
                  pprint.pformat(ret)
              )
          )
  
          # Let's remove the pip binary
          pip_bin = os.path.join(venv_dir, 'bin', 'pip')
          site_dir = self.run_function('virtualenv.get_distribution_path', [venv_dir, 'pip'])
          if salt.utils.platform.is_windows():
              pip_bin = os.path.join(venv_dir, 'Scripts', 'pip.exe')
              site_dir = os.path.join(venv_dir, 'lib', 'site-packages')
          if not os.path.isfile(pip_bin):
              self.skipTest(
                  'Failed to find the pip binary to the test virtualenv'
              )
          os.remove(pip_bin)
  
          # Also remove the pip dir from site-packages
          # This is needed now that we're using python -m pip instead of the
          # pip binary directly. python -m pip will still work even if the
          # pip binary is missing
          shutil.rmtree(os.path.join(site_dir, 'pip'))
  
          # Let's run the state which should fail because pip is missing
          ret = self.run_function('state.sls', mods='issue-2087-missing-pip')
          self.assertSaltFalseReturn(ret)
          self.assertInSaltComment(
              'Error installing \'pep8\': Could not find a `pip` binary',
              ret
          )
  
      def test_issue_5940_multiple_pip_mirrors(self):
          '''
          Test multiple pip mirrors.  This test only works with pip < 7.0.0
          '''
          ret = self.run_function(
              'state.sls', mods='issue-5940-multiple-pip-mirrors'
          )
  
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, '5940-multiple-pip-mirrors'
          )
          self.addCleanup(shutil.rmtree, venv_dir, ignore_errors=True)
  
          try:
              self.assertSaltTrueReturn(ret)
              self.assertTrue(
                  os.path.isfile(os.path.join(venv_dir, 'bin', 'pep8'))
              )
          except (AssertionError, CommandExecutionError):
              pip_version = self.run_function('pip.version', [venv_dir])
              if salt.utils.versions.compare(ver1=pip_version, oper='>=', ver2='7.0.0'):
                  self.skipTest('the --mirrors arg has been deprecated and removed in pip==7.0.0')
  
      @destructiveTest
      @skip_if_not_root
      @with_system_user('issue-6912', on_existing='delete', delete=True,
                        password='PassWord1!')
      @with_tempdir()
      def test_issue_6912_wrong_owner(self, temp_dir, username):
          # Setup virtual environment directory to be used throughout the test
          venv_dir = os.path.join(temp_dir, '6912-wrong-owner')
  
          # The virtual environment needs to be in a location that is accessible
          # by both the user running the test and the runas user
          if salt.utils.platform.is_windows():
              salt.utils.win_dacl.set_permissions(temp_dir, username, 'full_control')
          else:
              uid = self.run_function('file.user_to_uid', [username])
              os.chown(temp_dir, uid, -1)
  
          # Create the virtual environment
          venv_create = self._create_virtualenv(venv_dir, user=username, password='PassWord1!')
          if venv_create['retcode'] > 0:
              self.skipTest('Failed to create testcase virtual environment: {0}'
                            ''.format(venv_create))
  
          # pip install passing the package name in `name`
          ret = self.run_state(
              'pip.installed', name='pep8', user=username, bin_env=venv_dir,
              password='PassWord1!')
          self.assertSaltTrueReturn(ret)
  
          if HAS_PWD:
              uid = pwd.getpwnam(username).pw_uid
          for globmatch in (os.path.join(venv_dir, '**', 'pep8*'),
                            os.path.join(venv_dir, '*', '**', 'pep8*'),
                            os.path.join(venv_dir, '*', '*', '**', 'pep8*')):
              for path in glob.glob(globmatch):
                  if HAS_PWD:
                      self.assertEqual(uid, os.stat(path).st_uid)
                  elif salt.utils.platform.is_windows():
                      self.assertEqual(
                          salt.utils.win_dacl.get_owner(path), username)
  
      @destructiveTest
      @skip_if_not_root
      @skipIf(salt.utils.platform.is_darwin(), 'Test is flaky on macosx')
      @with_system_user('issue-6912', on_existing='delete', delete=True,
                        password='PassWord1!')
      @with_tempdir()
      def test_issue_6912_wrong_owner_requirements_file(self, temp_dir, username):
          # Setup virtual environment directory to be used throughout the test
          venv_dir = os.path.join(temp_dir, '6912-wrong-owner')
  
          # The virtual environment needs to be in a location that is accessible
          # by both the user running the test and the runas user
          if salt.utils.platform.is_windows():
              salt.utils.win_dacl.set_permissions(temp_dir, username, 'full_control')
          else:
              uid = self.run_function('file.user_to_uid', [username])
              os.chown(temp_dir, uid, -1)
  
          # Create the virtual environment again as it should have been removed
          venv_create = self._create_virtualenv(venv_dir, user=username, password='PassWord1!')
          if venv_create['retcode'] > 0:
              self.skipTest('failed to create testcase virtual environment: {0}'
                            ''.format(venv_create))
  
          # pip install using a requirements file
          req_filename = os.path.join(
              RUNTIME_VARS.TMP_STATE_TREE, 'issue-6912-requirements.txt'
          )
          with salt.utils.files.fopen(req_filename, 'wb') as reqf:
              reqf.write(b'pep8\n')
  
          ret = self.run_state(
              'pip.installed', name='', user=username, bin_env=venv_dir,
              requirements='salt://issue-6912-requirements.txt',
              password='PassWord1!')
          self.assertSaltTrueReturn(ret)
  
          if HAS_PWD:
              uid = pwd.getpwnam(username).pw_uid
          for globmatch in (os.path.join(venv_dir, '**', 'pep8*'),
                            os.path.join(venv_dir, '*', '**', 'pep8*'),
                            os.path.join(venv_dir, '*', '*', '**', 'pep8*')):
              for path in glob.glob(globmatch):
                  if HAS_PWD:
                      self.assertEqual(uid, os.stat(path).st_uid)
                  elif salt.utils.platform.is_windows():
                      self.assertEqual(
                          salt.utils.win_dacl.get_owner(path), username)
  
      def test_issue_6833_pip_upgrade_pip(self):
          # Create the testing virtualenv
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, '6833-pip-upgrade-pip'
          )
          ret = self._create_virtualenv(venv_dir)
  
          self.assertEqual(
              ret['retcode'], 0,
              msg='Expected \'retcode\' key did not match. Full return dictionary:\n{}'.format(
                  pprint.pformat(ret)
              )
          )
          import salt.modules.virtualenv_mod
          msg = 'New python executable'
          if salt.modules.virtualenv_mod.virtualenv_ver(venv_dir) >= (20, 0, 2):
              msg = 'created virtual environment'
          self.assertIn(
              msg,
              ret['stdout'],
              msg='Expected STDOUT did not match. Full return dictionary:\n{}'.format(
                  pprint.pformat(ret)
              )
          )
  
          # Let's install a fixed version pip over whatever pip was
          # previously installed
          ret = self.run_function(
              'pip.install', ['pip==8.0'], upgrade=True,
              bin_env=venv_dir
          )
  
          if not isinstance(ret, dict):
              self.fail(
                  'The \'pip.install\' command did not return the excepted dictionary. Output:\n{}'.format(ret)
              )
  
          self.assertEqual(ret['retcode'], 0)
          self.assertIn(
              'Successfully installed pip',
              ret['stdout']
          )
  
          # Let's make sure we have pip 8.0 installed
          self.assertEqual(
              self.run_function('pip.list', ['pip'], bin_env=venv_dir),
              {'pip': '8.0.0'}
          )
  
          # Now the actual pip upgrade pip test
          ret = self.run_state(
              'pip.installed', name='pip==8.0.1', upgrade=True,
              bin_env=venv_dir
          )
  
          if not isinstance(ret, dict):
              self.fail(
                  'The \'pip.install\' command did not return the excepted dictionary. Output:\n{}'.format(ret)
              )
  
          self.assertSaltTrueReturn(ret)
          self.assertSaltStateChangesEqual(ret, {'pip==8.0.1': 'Installed'})
  
      def test_pip_installed_specific_env(self):
          # Create the testing virtualenv
          venv_dir = os.path.join(
              RUNTIME_VARS.TMP, 'pip-installed-specific-env'
          )
  
          # Let's write a requirements file
          requirements_file = os.path.join(
              RUNTIME_VARS.TMP_PRODENV_STATE_TREE, 'prod-env-requirements.txt'
          )
          with salt.utils.files.fopen(requirements_file, 'wb') as reqf:
              reqf.write(b'pep8\n')
  
          try:
              self._create_virtualenv(venv_dir)
  
              # The requirements file should not be found the base environment
              ret = self.run_state(
                  'pip.installed', name='', bin_env=venv_dir,
                  requirements='salt://prod-env-requirements.txt'
              )
              self.assertSaltFalseReturn(ret)
              self.assertInSaltComment(
                  "'salt://prod-env-requirements.txt' not found", ret
              )
  
              # The requirements file must be found in the prod environment
              ret = self.run_state(
                  'pip.installed', name='', bin_env=venv_dir, saltenv='prod',
                  requirements='salt://prod-env-requirements.txt'
              )
              self.assertSaltTrueReturn(ret)
              self.assertInSaltComment(
                  'Successfully processed requirements file '
                  'salt://prod-env-requirements.txt', ret
              )
  
              # We're using the base environment but we're passing the prod
              # environment as an url arg to salt://
              ret = self.run_state(
                  'pip.installed', name='', bin_env=venv_dir,
                  requirements='salt://prod-env-requirements.txt?saltenv=prod'
              )
              self.assertSaltTrueReturn(ret)
              self.assertInSaltComment(
                  'Requirements were already installed.',
                  ret
              )
          finally:
              if os.path.isfile(requirements_file):
                  os.unlink(requirements_file)
  
      @skipIf(salt.utils.platform.is_darwin() and six.PY2, 'This test hangs on OS X on Py2')
      def test_22359_pip_installed_unless_does_not_trigger_warnings(self):
          # This test case should be moved to a format_call unit test specific to
          # the state internal keywords
          venv_dir = os.path.join(RUNTIME_VARS.TMP, 'pip-installed-unless')
          venv_create = self._create_virtualenv(venv_dir)
          if venv_create['retcode'] > 0:
              self.skipTest(
                  'Failed to create testcase virtual environment: {0}'.format(
                      venv_create
                  )
              )
  
          false_cmd = RUNTIME_VARS.SHELL_FALSE_PATH
          if salt.utils.platform.is_windows():
              false_cmd = 'exit 1 >nul'
          try:
              ret = self.run_state(
                  'pip.installed', name='pep8', bin_env=venv_dir, unless=false_cmd, timeout=600
              )
              self.assertSaltTrueReturn(ret)
              self.assertNotIn('warnings', next(six.itervalues(ret)))
          finally:
              if os.path.isdir(venv_dir):
                  shutil.rmtree(venv_dir, ignore_errors=True)
  
      @skipIf(sys.version_info[:2] >= (3, 6), 'Old version of virtualenv too old for python3.6')
      @skipIf(salt.utils.platform.is_windows(), "Carbon does not install in Windows")
      def test_46127_pip_env_vars(self):
          '''
          Test that checks if env_vars passed to pip.installed are also passed
          to pip.freeze while checking for existing installations
          '''
          # This issue is most easily checked while installing carbon
          # Much of the code here comes from the test_weird_install function above
          ographite = '/opt/graphite'
          if os.path.isdir(ographite):
              self.skipTest(
                  'You already have \'{0}\'. This test would overwrite this '
                  'directory'.format(ographite)
              )
          try:
              os.makedirs(ographite)
          except OSError as err:
              if err.errno == errno.EACCES:
                  # Permission denied
                  self.skipTest(
                      'You don\'t have the required permissions to run this test'
                  )
          finally:
              if os.path.isdir(ographite):
                  shutil.rmtree(ographite, ignore_errors=True)
  
          venv_dir = os.path.join(RUNTIME_VARS.TMP, 'issue-46127-pip-env-vars')
          try:
              # We may be able to remove this, I had to add it because the custom
              # modules from the test suite weren't available in the jinja
              # context when running the call to state.sls that comes after.
              self.run_function('saltutil.sync_modules')
              # Since we don't have the virtualenv created, pip.installed will
              # throw an error.
              ret = self.run_function(
                  'state.sls', mods='issue-46127-pip-env-vars'
              )
              self.assertSaltTrueReturn(ret)
              for key in six.iterkeys(ret):
                  self.assertTrue(ret[key]['result'])
                  if ret[key]['name'] != 'carbon < 1.3':
                      continue
                  self.assertEqual(
                      ret[key]['comment'],
                      'All packages were successfully installed'
                  )
                  break
              else:
                  raise Exception('Expected state did not run')
              # Run the state again. Now the already installed message should
              # appear
              ret = self.run_function(
                  'state.sls', mods='issue-46127-pip-env-vars'
              )
              self.assertSaltTrueReturn(ret)
              # We cannot use assertInSaltComment here because we need to skip
              # some of the state return parts
              for key in six.iterkeys(ret):
                  self.assertTrue(ret[key]['result'])
                  # As we are re-running the formula, some states will not be run
                  # and "name" may or may not be present, so we use .get() pattern
                  if ret[key].get('name', '') != 'carbon < 1.3':
                      continue
                  self.assertEqual(
                      ret[key]['comment'],
                      ('All packages were successfully installed'))
                  break
              else:
                  raise Exception('Expected state did not run')
          finally:
              if os.path.isdir(ographite):
                  shutil.rmtree(ographite, ignore_errors=True)
              if os.path.isdir(venv_dir):
                  shutil.rmtree(venv_dir)
  
  
  class PipStateInRequisiteTest(ModuleCase, SaltReturnAssertsMixin):
  
      @with_tempdir()
      def test_issue_54755(self, tmpdir):
          '''
          Verify github issue 54755 is resolved. This only fails when there is no
          pip module in the python environment. Since the test suite normally has
          a pip module this test will pass and is here for posterity. See also
  
          unit.states.test_pip_state.PipStateUtilsTest.test_pip_purge_method_with_pip
  
           and
  
          unit.states.test_pip_state.PipStateUtilsTest.test_pip_purge_method_without_pip
  
          Which also validate this issue and will pass/fail regardless of whether
          or not pip is installed.
          '''
          file_path = os.path.join(tmpdir, 'issue-54755')
          ret = self.run_function('state.sls', mods='issue-54755', pillar={'file_path': file_path})
          key = 'file_|-issue-54755_|-{}_|-managed'.format(file_path)
          assert key in ret
          assert ret[key]['result'] is True
          with salt.utils.files.fopen(file_path, 'r') as fp:
              assert fp.read().strip() == 'issue-54755'
  