# -*- coding: utf-8 -*- """ :codeauthor: Pedro Algarvio (pedro@algarvio.me) tests.unit.states.pip_test ~~~~~~~~~~~~~~~~~~~~~~~~~~ """ # Import python libs from __future__ import absolute_import, print_function, unicode_literals import logging import os import subprocess import sys import pytest import salt.states.pip_state as pip_state import salt.utils.path # Import salt libs import salt.version from salt.modules.virtualenv_mod import KNOWN_BINARY_NAMES from tests.support.helpers import VirtualEnv, dedent # Import Salt Testing libs from tests.support.mixins import LoaderModuleMockMixin, SaltReturnAssertsMixin from tests.support.mock import MagicMock, patch from tests.support.runtests import RUNTIME_VARS from tests.support.unit import TestCase, skipIf # Import 3rd-party libs try: import pip HAS_PIP = True except ImportError: HAS_PIP = False log = logging.getLogger(__name__) @skipIf(not HAS_PIP, "The 'pip' library is not importable(installed system-wide)") class PipStateTest(TestCase, SaltReturnAssertsMixin, LoaderModuleMockMixin): def setup_loader_modules(self): return { pip_state: { "__env__": "base", "__opts__": {"test": False}, "__salt__": {"cmd.which_bin": lambda _: "pip"}, } } @pytest.mark.slow_test(seconds=1) # Test takes >0.1 and <=1 seconds def test_install_requirements_parsing(self): log.debug("Real pip version is %s", pip.__version__) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.3"}) pip_version = pip.__version__ mock_pip_version = MagicMock(return_value=pip_version) with patch.dict(pip_state.__salt__, {"pip.version": mock_pip_version}): with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list} ): with patch.dict(pip_state.__opts__, {"test": True}): log.debug( "pip_state._from_line globals: %s", [name for name in pip_state._from_line.__globals__], ) ret = pip_state.installed("pep8=1.3.2") self.assertSaltFalseReturn({"test": ret}) self.assertInSaltComment( "Invalid version specification in package pep8=1.3.2. " "'=' is not supported, use '==' instead.", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.3"}) pip_install = MagicMock(return_value={"retcode": 0}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed("pep8>=1.3.2") self.assertSaltTrueReturn({"test": ret}) self.assertInSaltComment( "Python package pep8>=1.3.2 was already installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.3"}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list} ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed("pep8<1.3.2") self.assertSaltNoneReturn({"test": ret}) self.assertInSaltComment( "Python package pep8<1.3.2 is set to be installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.2"}) pip_install = MagicMock(return_value={"retcode": 0}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed("pep8>1.3.1,<1.3.3") self.assertSaltTrueReturn({"test": ret}) self.assertInSaltComment( "Python package pep8>1.3.1,<1.3.3 was already installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.1"}) pip_install = MagicMock(return_value={"retcode": 0}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed("pep8>1.3.1,<1.3.3") self.assertSaltNoneReturn({"test": ret}) self.assertInSaltComment( "Python package pep8>1.3.1,<1.3.3 is set to be installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.1"}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list} ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed( "git+https://github.com/saltstack/salt-testing.git#egg=SaltTesting>=0.5.1" ) self.assertSaltNoneReturn({"test": ret}) self.assertInSaltComment( "Python package git+https://github.com/saltstack/" "salt-testing.git#egg=SaltTesting>=0.5.1 is set to be " "installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.1"}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list} ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed( "git+https://github.com/saltstack/salt-testing.git#egg=SaltTesting" ) self.assertSaltNoneReturn({"test": ret}) self.assertInSaltComment( "Python package git+https://github.com/saltstack/" "salt-testing.git#egg=SaltTesting is set to be " "installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.1"}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list} ): with patch.dict(pip_state.__opts__, {"test": True}): ret = pip_state.installed( "https://pypi.python.org/packages/source/S/SaltTesting/" "SaltTesting-0.5.0.tar.gz" "#md5=e6760af92b7165f8be53b5763e40bc24" ) self.assertSaltNoneReturn({"test": ret}) self.assertInSaltComment( "Python package https://pypi.python.org/packages/source/" "S/SaltTesting/SaltTesting-0.5.0.tar.gz" "#md5=e6760af92b7165f8be53b5763e40bc24 is set to be " "installed", {"test": ret}, ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"SaltTesting": "0.5.0"}) pip_install = MagicMock( return_value={ "retcode": 0, "stderr": "", "stdout": "Downloading/unpacking https://pypi.python.org/packages" "/source/S/SaltTesting/SaltTesting-0.5.0.tar.gz\n " "Downloading SaltTesting-0.5.0.tar.gz\n Running " "setup.py egg_info for package from " "https://pypi.python.org/packages/source/S/SaltTesting/" "SaltTesting-0.5.0.tar.gz\n \nCleaning up...", } ) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): ret = pip_state.installed( "https://pypi.python.org/packages/source/S/SaltTesting/" "SaltTesting-0.5.0.tar.gz" "#md5=e6760af92b7165f8be53b5763e40bc24" ) self.assertSaltTrueReturn({"test": ret}) self.assertInSaltComment( "All packages were successfully installed", {"test": ret} ) self.assertInSaltReturn( "Installed", {"test": ret}, ( "changes", "https://pypi.python.org/packages/source/S/" "SaltTesting/SaltTesting-0.5.0.tar.gz" "#md5=e6760af92b7165f8be53b5763e40bc24==???", ), ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"SaltTesting": "0.5.0"}) pip_install = MagicMock( return_value={"retcode": 0, "stderr": "", "stdout": "Cloned!"} ) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): with patch.dict(pip_state.__opts__, {"test": False}): ret = pip_state.installed( "git+https://github.com/saltstack/salt-testing.git#egg=SaltTesting" ) self.assertSaltTrueReturn({"test": ret}) self.assertInSaltComment( "packages are already installed", {"test": ret} ) mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"pep8": "1.3.1"}) pip_install = MagicMock(return_value={"retcode": 0}) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): with patch.dict(pip_state.__opts__, {"test": False}): ret = pip_state.installed( "arbitrary ID that should be ignored due to requirements specified", requirements="/tmp/non-existing-requirements.txt", ) self.assertSaltTrueReturn({"test": ret}) # Test VCS installations using git+git:// mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={"SaltTesting": "0.5.0"}) pip_install = MagicMock( return_value={"retcode": 0, "stderr": "", "stdout": "Cloned!"} ) with patch.dict( pip_state.__salt__, {"cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install}, ): with patch.dict(pip_state.__opts__, {"test": False}): ret = pip_state.installed( "git+git://github.com/saltstack/salt-testing.git#egg=SaltTesting" ) self.assertSaltTrueReturn({"test": ret}) self.assertInSaltComment( "packages are already installed", {"test": ret} ) def test_install_requirements_custom_pypi(self): """ test requirement parsing for both when a custom pypi index-url is set and when it is not and the requirement is already installed. """ # create requirements file req_filename = os.path.join( RUNTIME_VARS.TMP_STATE_TREE, "custom-pypi-requirements.txt" ) with salt.utils.files.fopen(req_filename, "wb") as reqf: reqf.write(b"pep8\n") site_pkgs = "/tmp/pip-env/lib/python3.7/site-packages" check_stdout = [ ( "Looking in indexes: https://custom-pypi-url.org," "https://pypi.org/simple/\nRequirement already satisfied: pep8 in {1}" "(from -r /tmp/files/prod/{0} (line 1)) (1.7.1)".format( req_filename, site_pkgs ) ), ( "Requirement already satisfied: pep8 in {1}" "(from -r /tmp/files/prod/{0} (line1)) (1.7.1)".format( req_filename, site_pkgs ) ), ] pip_version = pip.__version__ mock_pip_version = MagicMock(return_value=pip_version) for stdout in check_stdout: pip_install = MagicMock(return_value={"retcode": 0, "stdout": stdout}) with patch.dict(pip_state.__salt__, {"pip.version": mock_pip_version}): with patch.dict(pip_state.__salt__, {"pip.install": pip_install}): ret = pip_state.installed(name="", requirements=req_filename) self.assertSaltTrueReturn({"test": ret}) assert "Requirements were already installed." == ret["comment"] def test_install_requirements_custom_pypi_changes(self): """ test requirement parsing for both when a custom pypi index-url is set and when it is not and the requirement is not installed. """ # create requirements file req_filename = os.path.join( RUNTIME_VARS.TMP_STATE_TREE, "custom-pypi-requirements.txt" ) with salt.utils.files.fopen(req_filename, "wb") as reqf: reqf.write(b"pep8\n") site_pkgs = "/tmp/pip-env/lib/python3.7/site-packages" check_stdout = [ ( "Looking in indexes: https://custom-pypi-url.org," "https://pypi.org/simple/\nCollecting pep8\n Using cached" "https://custom-pypi-url.org//packages/42/3f/669429cef5acb4/pep8-1.7.1-py2.py3-none-any.whl" " (41 kB)\nInstalling collected packages: pep8\nSuccessfully installed pep8-1.7.1" ), ( "Collecting pep8\n Using cached" "https://custom-pypi-url.org//packages/42/3f/669429cef5acb4/pep8-1.7.1-py2.py3-none-any.whl" " (41 kB)\nInstalling collected packages: pep8\nSuccessfully installed pep8-1.7.1" ), ] pip_version = pip.__version__ mock_pip_version = MagicMock(return_value=pip_version) for stdout in check_stdout: pip_install = MagicMock(return_value={"retcode": 0, "stdout": stdout}) with patch.dict(pip_state.__salt__, {"pip.version": mock_pip_version}): with patch.dict(pip_state.__salt__, {"pip.install": pip_install}): ret = pip_state.installed(name="", requirements=req_filename) self.assertSaltTrueReturn({"test": ret}) assert ( "Successfully processed requirements file {0}.".format( req_filename ) == ret["comment"] ) def test_install_in_editable_mode(self): """ Check that `name` parameter containing bad characters is not parsed by pip when package is being installed in editable mode. For more information, see issue #21890. """ mock = MagicMock(return_value={"retcode": 0, "stdout": ""}) pip_list = MagicMock(return_value={}) pip_install = MagicMock( return_value={"retcode": 0, "stderr": "", "stdout": "Cloned!"} ) pip_version = MagicMock(return_value="10.0.1") with patch.dict( pip_state.__salt__, { "cmd.run_all": mock, "pip.list": pip_list, "pip.install": pip_install, "pip.version": pip_version, }, ): ret = pip_state.installed( "state@name", cwd="/path/to/project", editable=["."] ) self.assertSaltTrueReturn({"test": ret}) self.assertInSaltComment("successfully installed", {"test": ret}) class PipStateUtilsTest(TestCase): def test_has_internal_exceptions_mod_function(self): assert pip_state.pip_has_internal_exceptions_mod("10.0") assert pip_state.pip_has_internal_exceptions_mod("18.1") assert not pip_state.pip_has_internal_exceptions_mod("9.99") def test_has_exceptions_mod_function(self): assert pip_state.pip_has_exceptions_mod("1.0") assert not pip_state.pip_has_exceptions_mod("0.1") assert not pip_state.pip_has_exceptions_mod("10.0") def test_pip_purge_method_with_pip(self): mock_modules = sys.modules.copy() mock_modules.pop("pip", None) mock_modules["pip"] = object() with patch("sys.modules", mock_modules): pip_state.purge_pip() assert "pip" not in mock_modules def test_pip_purge_method_without_pip(self): mock_modules = sys.modules.copy() mock_modules.pop("pip", None) with patch("sys.modules", mock_modules): pip_state.purge_pip() @skipIf( salt.utils.path.which_bin(KNOWN_BINARY_NAMES) is None, "virtualenv not installed" ) class PipStateInstallationErrorTest(TestCase): @pytest.mark.slow_test(seconds=120) # Test takes >60 and <=120 seconds def test_importable_installation_error(self): extra_requirements = [] for name, version in salt.version.dependency_information(): if name in ["PyYAML"]: extra_requirements.append("{}=={}".format(name, version)) failures = {} pip_version_requirements = [ # Latest pip 8 "<9.0", # Latest pip 9 "<10.0", # Latest pip 18 "<19.0", # Latest pip 19 "<20.0", # Latest pip 20 "<21.0", # Latest pip None, ] code = dedent( """\ import sys import traceback try: import salt.states.pip_state salt.states.pip_state.InstallationError except ImportError as exc: traceback.print_exc(exc, file=sys.stdout) sys.stdout.flush() sys.exit(1) except AttributeError as exc: traceback.print_exc(exc, file=sys.stdout) sys.stdout.flush() sys.exit(2) except Exception as exc: traceback.print_exc(exc, file=sys.stdout) sys.stdout.flush() sys.exit(3) sys.exit(0) """ ) for requirement in list(pip_version_requirements): try: with VirtualEnv() as venv: venv.install(*extra_requirements) if requirement: venv.install("pip{}".format(requirement)) try: subprocess.check_output([venv.venv_python, "-c", code]) except subprocess.CalledProcessError as exc: if exc.returncode == 1: failures[requirement] = "Failed to import pip:\n{}".format( exc.output ) elif exc.returncode == 2: failures[ requirement ] = "Failed to import InstallationError from pip:\n{}".format( exc.output ) else: failures[requirement] = exc.output except Exception as exc: # pylint: disable=broad-except failures[requirement] = str(exc) if failures: errors = "" for requirement, exception in failures.items(): errors += "pip{}: {}\n\n".format(requirement or "", exception) self.fail( "Failed to get InstallationError exception under at least one pip version:\n{}".format( errors ) )