# -*- coding: utf-8 -*- from __future__ import absolute_import, print_function, unicode_literals import logging import os import shutil import subprocess import tempfile import salt.modules.cmdmod as cmd import salt.modules.virtualenv_mod import salt.modules.zcbuildout as buildout import salt.utils.files import salt.utils.path import salt.utils.platform # pylint: disable=import-error,no-name-in-module,redefined-builtin from salt.ext import six from salt.ext.six.moves.urllib.error import URLError from salt.ext.six.moves.urllib.request import urlopen from tests.support.helpers import patched_environ, requires_network, slowTest from tests.support.mixins import LoaderModuleMockMixin from tests.support.runtests import RUNTIME_VARS from tests.support.unit import TestCase, skipIf # Import 3rd-party libs # pylint: enable=import-error,no-name-in-module,redefined-builtin KNOWN_VIRTUALENV_BINARY_NAMES = ( "virtualenv", "virtualenv2", "virtualenv-2.6", "virtualenv-2.7", ) # temp workaround since virtualenv pip wheel package does not include # backports.ssl_match_hostname on windows python2.7 if salt.utils.platform.is_windows() and six.PY2: KNOWN_VIRTUALENV_BINARY_NAMES = ("c:\\Python27\\Scripts\\virtualenv.EXE",) BOOT_INIT = { 1: ["var/ver/1/bootstrap/bootstrap.py"], 2: ["var/ver/2/bootstrap/bootstrap.py", "b/bootstrap.py"], } log = logging.getLogger(__name__) def download_to(url, dest): with salt.utils.files.fopen(dest, "wb") as fic: fic.write(urlopen(url, timeout=10).read()) class Base(TestCase, LoaderModuleMockMixin): def setup_loader_modules(self): return { buildout: { "__salt__": { "cmd.run_all": cmd.run_all, "cmd.run": cmd.run, "cmd.retcode": cmd.retcode, } } } @classmethod def setUpClass(cls): if not os.path.isdir(RUNTIME_VARS.TMP): os.makedirs(RUNTIME_VARS.TMP) cls.root = os.path.join(RUNTIME_VARS.BASE_FILES, "buildout") cls.rdir = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP) cls.tdir = os.path.join(cls.rdir, "test") for idx, url in six.iteritems(buildout._URL_VERSIONS): log.debug("Downloading bootstrap from %s", url) dest = os.path.join(cls.rdir, "{0}_bootstrap.py".format(idx)) try: download_to(url, dest) except URLError: log.debug("Failed to download %s", url) # creating a new setuptools install cls.ppy_st = os.path.join(cls.rdir, "psetuptools") if salt.utils.platform.is_windows(): cls.bin_st = os.path.join(cls.ppy_st, "Scripts") cls.py_st = os.path.join(cls.bin_st, "python") else: cls.bin_st = os.path.join(cls.ppy_st, "bin") cls.py_st = os.path.join(cls.bin_st, "python") # `--no-site-packages` has been deprecated # https://virtualenv.pypa.io/en/stable/reference/#cmdoption-no-site-packages subprocess.check_call( [salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), cls.ppy_st] ) subprocess.check_call( [os.path.join(cls.bin_st, "pip"), "install", "-U", "setuptools"] ) # distribute has been merged back in to setuptools as of v0.7. So, no # need to upgrade distribute, but this seems to be the only way to get # the binary in the right place # https://packaging.python.org/key_projects/#setuptools # Additionally, this part may fail if the certificate store is outdated # on Windows, as it would be in a fresh installation for example. The # following commands will fix that. This should be part of the golden # images. (https://github.com/saltstack/salt-jenkins/pull/1479) # certutil -generateSSTFromWU roots.sst # powershell "(Get-ChildItem -Path .\roots.sst) | Import-Certificate -CertStoreLocation Cert:\LocalMachine\Root" subprocess.check_call( [os.path.join(cls.bin_st, "easy_install"), "-U", "distribute"] ) def setUp(self): if salt.utils.platform.is_darwin and six.PY3: self.patched_environ = patched_environ(__cleanup__=["__PYVENV_LAUNCHER__"]) self.patched_environ.__enter__() self.addCleanup(self.patched_environ.__exit__) super(Base, self).setUp() self._remove_dir() shutil.copytree(self.root, self.tdir) for idx in BOOT_INIT: path = os.path.join(self.rdir, "{0}_bootstrap.py".format(idx)) for fname in BOOT_INIT[idx]: shutil.copy2(path, os.path.join(self.tdir, fname)) def tearDown(self): super(Base, self).tearDown() self._remove_dir() def _remove_dir(self): if os.path.isdir(self.tdir): shutil.rmtree(self.tdir) @skipIf( salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES) is None, "The 'virtualenv' packaged needs to be installed", ) @requires_network() class BuildoutTestCase(Base): @slowTest def test_onlyif_unless(self): b_dir = os.path.join(self.tdir, "b") ret = buildout.buildout(b_dir, onlyif=RUNTIME_VARS.SHELL_FALSE_PATH) self.assertTrue(ret["comment"] == "onlyif condition is false") self.assertTrue(ret["status"] is True) ret = buildout.buildout(b_dir, unless=RUNTIME_VARS.SHELL_TRUE_PATH) self.assertTrue(ret["comment"] == "unless condition is true") self.assertTrue(ret["status"] is True) @slowTest def test_salt_callback(self): @buildout._salt_callback def callback1(a, b=1): for i in buildout.LOG.levels: getattr(buildout.LOG, i)("{0}bar".format(i[0])) return "foo" def callback2(a, b=1): raise Exception("foo") # pylint: disable=invalid-sequence-index ret1 = callback1(1, b=3) # These lines are throwing pylint errors - disabling for now since we are skipping # these tests # self.assertEqual(ret1['status'], True) # self.assertEqual(ret1['logs_by_level']['warn'], ['wbar']) # self.assertEqual(ret1['comment'], '') # These lines are throwing pylint errors - disabling for now since we are skipping # these tests # self.assertTrue( # u'' # u'OUTPUT:\n' # u'foo\n' # u'' # in ret1['outlog'] # ) # These lines are throwing pylint errors - disabling for now since we are skipping # these tests # self.assertTrue(u'Log summary:\n' in ret1['outlog']) # These lines are throwing pylint errors - disabling for now since we are skipping # these tests # self.assertTrue( # u'INFO: ibar\n' # u'WARN: wbar\n' # u'DEBUG: dbar\n' # u'ERROR: ebar\n' # in ret1['outlog'] # ) # These lines are throwing pylint errors - disabling for now since we are skipping # these tests # self.assertTrue('by level' in ret1['outlog_by_level']) # self.assertEqual(ret1['out'], 'foo') ret2 = buildout._salt_callback(callback2)(2, b=6) self.assertEqual(ret2["status"], False) self.assertTrue(ret2["logs_by_level"]["error"][0].startswith("Traceback")) self.assertTrue( "We did not get any " "expectable answer " "from buildout" in ret2["comment"] ) self.assertEqual(ret2["out"], None) for l in buildout.LOG.levels: self.assertTrue(0 == len(buildout.LOG.by_level[l])) # pylint: enable=invalid-sequence-index @slowTest def test_get_bootstrap_url(self): for path in [ os.path.join(self.tdir, "var/ver/1/dumppicked"), os.path.join(self.tdir, "var/ver/1/bootstrap"), os.path.join(self.tdir, "var/ver/1/versions"), ]: self.assertEqual( buildout._URL_VERSIONS[1], buildout._get_bootstrap_url(path), "b1 url for {0}".format(path), ) for path in [ os.path.join(self.tdir, "/non/existing"), os.path.join(self.tdir, "var/ver/2/versions"), os.path.join(self.tdir, "var/ver/2/bootstrap"), os.path.join(self.tdir, "var/ver/2/default"), ]: self.assertEqual( buildout._URL_VERSIONS[2], buildout._get_bootstrap_url(path), "b2 url for {0}".format(path), ) @slowTest def test_get_buildout_ver(self): for path in [ os.path.join(self.tdir, "var/ver/1/dumppicked"), os.path.join(self.tdir, "var/ver/1/bootstrap"), os.path.join(self.tdir, "var/ver/1/versions"), ]: self.assertEqual( 1, buildout._get_buildout_ver(path), "1 for {0}".format(path) ) for path in [ os.path.join(self.tdir, "/non/existing"), os.path.join(self.tdir, "var/ver/2/versions"), os.path.join(self.tdir, "var/ver/2/bootstrap"), os.path.join(self.tdir, "var/ver/2/default"), ]: self.assertEqual( 2, buildout._get_buildout_ver(path), "2 for {0}".format(path) ) @slowTest def test_get_bootstrap_content(self): self.assertEqual( "", buildout._get_bootstrap_content(os.path.join(self.tdir, "non", "existing")), ) self.assertEqual( "", buildout._get_bootstrap_content(os.path.join(self.tdir, "var", "tb", "1")), ) self.assertEqual( "foo{0}".format(os.linesep), buildout._get_bootstrap_content(os.path.join(self.tdir, "var", "tb", "2")), ) @slowTest def test_logger_clean(self): buildout.LOG.clear() # nothing in there self.assertTrue( True not in [len(buildout.LOG.by_level[a]) > 0 for a in buildout.LOG.by_level] ) buildout.LOG.info("foo") self.assertTrue( True in [len(buildout.LOG.by_level[a]) > 0 for a in buildout.LOG.by_level] ) buildout.LOG.clear() self.assertTrue( True not in [len(buildout.LOG.by_level[a]) > 0 for a in buildout.LOG.by_level] ) @slowTest def test_logger_loggers(self): buildout.LOG.clear() # nothing in there for i in buildout.LOG.levels: getattr(buildout.LOG, i)("foo") getattr(buildout.LOG, i)("bar") getattr(buildout.LOG, i)("moo") self.assertTrue(len(buildout.LOG.by_level[i]) == 3) self.assertEqual(buildout.LOG.by_level[i][0], "foo") self.assertEqual(buildout.LOG.by_level[i][-1], "moo") @slowTest def test__find_cfgs(self): result = sorted( [a.replace(self.root, "") for a in buildout._find_cfgs(self.root)] ) assertlist = sorted( [ os.path.join(os.sep, "buildout.cfg"), os.path.join(os.sep, "c", "buildout.cfg"), os.path.join(os.sep, "etc", "buildout.cfg"), os.path.join(os.sep, "e", "buildout.cfg"), os.path.join(os.sep, "b", "buildout.cfg"), os.path.join(os.sep, "b", "bdistribute", "buildout.cfg"), os.path.join(os.sep, "b", "b2", "buildout.cfg"), os.path.join(os.sep, "foo", "buildout.cfg"), ] ) self.assertEqual(result, assertlist) def skip_test_upgrade_bootstrap(self): b_dir = os.path.join(self.tdir, "b") bpy = os.path.join(b_dir, "bootstrap.py") buildout.upgrade_bootstrap(b_dir) time1 = os.stat(bpy).st_mtime with salt.utils.files.fopen(bpy) as fic: data = fic.read() self.assertTrue("setdefaulttimeout(2)" in data) flag = os.path.join(b_dir, ".buildout", "2.updated_bootstrap") self.assertTrue(os.path.exists(flag)) buildout.upgrade_bootstrap(b_dir, buildout_ver=1) time2 = os.stat(bpy).st_mtime with salt.utils.files.fopen(bpy) as fic: data = fic.read() self.assertTrue("setdefaulttimeout(2)" in data) flag = os.path.join(b_dir, ".buildout", "1.updated_bootstrap") self.assertTrue(os.path.exists(flag)) buildout.upgrade_bootstrap(b_dir, buildout_ver=1) time3 = os.stat(bpy).st_mtime self.assertNotEqual(time2, time1) self.assertEqual(time2, time3) @skipIf( salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES) is None, "The 'virtualenv' packaged needs to be installed", ) @requires_network() class BuildoutOnlineTestCase(Base): @classmethod def setUpClass(cls): super(BuildoutOnlineTestCase, cls).setUpClass() cls.ppy_dis = os.path.join(cls.rdir, "pdistribute") cls.ppy_blank = os.path.join(cls.rdir, "pblank") cls.py_dis = os.path.join(cls.ppy_dis, "bin", "python") cls.py_blank = os.path.join(cls.ppy_blank, "bin", "python") # creating a distribute based install try: # `--no-site-packages` has been deprecated # https://virtualenv.pypa.io/en/stable/reference/#cmdoption-no-site-packages subprocess.check_call( [ salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), "--no-setuptools", "--no-pip", cls.ppy_dis, ] ) except subprocess.CalledProcessError: subprocess.check_call( [salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), cls.ppy.dis] ) url = ( "https://pypi.python.org/packages/source" "/d/distribute/distribute-0.6.43.tar.gz" ) download_to( url, os.path.join(cls.ppy_dis, "distribute-0.6.43.tar.gz"), ) subprocess.check_call( [ "tar", "-C", cls.ppy_dis, "-xzvf", "{0}/distribute-0.6.43.tar.gz".format(cls.ppy_dis), ] ) subprocess.check_call( [ "{0}/bin/python".format(cls.ppy_dis), "{0}/distribute-0.6.43/setup.py".format(cls.ppy_dis), "install", ] ) # creating a blank based install try: subprocess.check_call( [ salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), "--no-setuptools", "--no-pip", cls.ppy_blank, ] ) except subprocess.CalledProcessError: subprocess.check_call( [ salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), cls.ppy_blank, ] ) @skipIf(True, "TODO this test should probably be fixed") def test_buildout_bootstrap(self): b_dir = os.path.join(self.tdir, "b") bd_dir = os.path.join(self.tdir, "b", "bdistribute") b2_dir = os.path.join(self.tdir, "b", "b2") self.assertTrue(buildout._has_old_distribute(self.py_dis)) # this is too hard to check as on debian & other where old # packages are present (virtualenv), we can't have # a clean site-packages # self.assertFalse(buildout._has_old_distribute(self.py_blank)) self.assertFalse(buildout._has_old_distribute(self.py_st)) self.assertFalse(buildout._has_setuptools7(self.py_dis)) self.assertTrue(buildout._has_setuptools7(self.py_st)) self.assertFalse(buildout._has_setuptools7(self.py_blank)) ret = buildout.bootstrap(bd_dir, buildout_ver=1, python=self.py_dis) comment = ret["outlog"] self.assertTrue("--distribute" in comment) self.assertTrue("Generated script" in comment) ret = buildout.bootstrap(b_dir, buildout_ver=1, python=self.py_blank) comment = ret["outlog"] # as we may have old packages, this test the two # behaviors (failure with old setuptools/distribute) self.assertTrue( ("Got " in comment and "Generated script" in comment) or ("setuptools>=0.7" in comment) ) ret = buildout.bootstrap(b_dir, buildout_ver=2, python=self.py_blank) comment = ret["outlog"] self.assertTrue( ("setuptools" in comment and "Generated script" in comment) or ("setuptools>=0.7" in comment) ) ret = buildout.bootstrap(b_dir, buildout_ver=2, python=self.py_st) comment = ret["outlog"] self.assertTrue( ("setuptools" in comment and "Generated script" in comment) or ("setuptools>=0.7" in comment) ) ret = buildout.bootstrap(b2_dir, buildout_ver=2, python=self.py_st) comment = ret["outlog"] self.assertTrue( ("setuptools" in comment and "Creating directory" in comment) or ("setuptools>=0.7" in comment) ) @slowTest def test_run_buildout(self): if salt.modules.virtualenv_mod.virtualenv_ver(self.ppy_st) >= (20, 0, 0): self.skipTest( "Skiping until upstream resolved https://github.com/pypa/virtualenv/issues/1715" ) b_dir = os.path.join(self.tdir, "b") ret = buildout.bootstrap(b_dir, buildout_ver=2, python=self.py_st) self.assertTrue(ret["status"]) ret = buildout.run_buildout(b_dir, parts=["a", "b"]) out = ret["out"] self.assertTrue("Installing a" in out) self.assertTrue("Installing b" in out) @slowTest def test_buildout(self): if salt.modules.virtualenv_mod.virtualenv_ver(self.ppy_st) >= (20, 0, 0): self.skipTest( "Skiping until upstream resolved https://github.com/pypa/virtualenv/issues/1715" ) b_dir = os.path.join(self.tdir, "b") ret = buildout.buildout(b_dir, buildout_ver=2, python=self.py_st) self.assertTrue(ret["status"]) out = ret["out"] comment = ret["comment"] self.assertTrue(ret["status"]) self.assertTrue("Creating directory" in out) self.assertTrue("Installing a." in out) self.assertTrue("{0} bootstrap.py".format(self.py_st) in comment) self.assertTrue("buildout -c buildout.cfg" in comment) ret = buildout.buildout( b_dir, parts=["a", "b", "c"], buildout_ver=2, python=self.py_st ) outlog = ret["outlog"] out = ret["out"] comment = ret["comment"] self.assertTrue("Installing single part: a" in outlog) self.assertTrue("buildout -c buildout.cfg -N install a" in comment) self.assertTrue("Installing b." in out) self.assertTrue("Installing c." in out) ret = buildout.buildout( b_dir, parts=["a", "b", "c"], buildout_ver=2, newest=True, python=self.py_st ) outlog = ret["outlog"] out = ret["out"] comment = ret["comment"] self.assertTrue("buildout -c buildout.cfg -n install a" in comment) # TODO: Is this test even still needed? class BuildoutAPITestCase(TestCase): def test_merge(self): buildout.LOG.clear() buildout.LOG.info("àé") buildout.LOG.info("àé") buildout.LOG.error("àé") buildout.LOG.error("àé") ret1 = buildout._set_status({}, out="éà") uret1 = buildout._set_status({}, out="éà") buildout.LOG.clear() buildout.LOG.info("ççàé") buildout.LOG.info("ççàé") buildout.LOG.error("ççàé") buildout.LOG.error("ççàé") ret2 = buildout._set_status({}, out="çéà") uret2 = buildout._set_status({}, out="çéà") uretm = buildout._merge_statuses([ret1, uret1, ret2, uret2]) for ret in ret1, uret1, ret2, uret2: out = ret["out"] if not isinstance(ret["out"], six.text_type): out = ret["out"].decode("utf-8") for out in ["àé", "ççàé"]: self.assertTrue(out in uretm["logs_by_level"]["info"]) self.assertTrue(out in uretm["outlog_by_level"]) def test_setup(self): buildout.LOG.clear() buildout.LOG.info("àé") buildout.LOG.info("àé") buildout.LOG.error("àé") buildout.LOG.error("àé") ret = buildout._set_status({}, out="éà") uret = buildout._set_status({}, out="éà") self.assertTrue(ret["outlog"] == uret["outlog"]) self.assertTrue("àé" in uret["outlog_by_level"])