1import logging 2import os 3import shutil 4import subprocess 5import tempfile 6import urllib.error 7import urllib.request 8 9import pytest 10import salt.modules.cmdmod as cmd 11import salt.modules.virtualenv_mod 12import salt.modules.zcbuildout as buildout 13import salt.utils.files 14import salt.utils.path 15import salt.utils.platform 16from tests.support.helpers import patched_environ 17from tests.support.mixins import LoaderModuleMockMixin 18from tests.support.runtests import RUNTIME_VARS 19from tests.support.unit import TestCase, skipIf 20 21KNOWN_VIRTUALENV_BINARY_NAMES = ( 22 "virtualenv", 23 "virtualenv2", 24 "virtualenv-2.6", 25 "virtualenv-2.7", 26) 27 28BOOT_INIT = { 29 1: ["var/ver/1/bootstrap/bootstrap.py"], 30 2: ["var/ver/2/bootstrap/bootstrap.py", "b/bootstrap.py"], 31} 32 33log = logging.getLogger(__name__) 34 35 36def download_to(url, dest): 37 req = urllib.request.Request(url) 38 req.add_header( 39 "User-Agent", 40 "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36", 41 ) 42 with salt.utils.files.fopen(dest, "wb") as fic: 43 fic.write(urllib.request.urlopen(req, timeout=10).read()) 44 45 46class Base(TestCase, LoaderModuleMockMixin): 47 def setup_loader_modules(self): 48 return { 49 buildout: { 50 "__salt__": { 51 "cmd.run_all": cmd.run_all, 52 "cmd.run": cmd.run, 53 "cmd.retcode": cmd.retcode, 54 } 55 } 56 } 57 58 @classmethod 59 def setUpClass(cls): 60 if not os.path.isdir(RUNTIME_VARS.TMP): 61 os.makedirs(RUNTIME_VARS.TMP) 62 63 cls.root = os.path.join(RUNTIME_VARS.BASE_FILES, "buildout") 64 cls.rdir = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP) 65 cls.tdir = os.path.join(cls.rdir, "test") 66 for idx, url in buildout._URL_VERSIONS.items(): 67 log.debug("Downloading bootstrap from %s", url) 68 dest = os.path.join(cls.rdir, "{}_bootstrap.py".format(idx)) 69 try: 70 download_to(url, dest) 71 except urllib.error.URLError as exc: 72 log.debug("Failed to download %s: %s", url, exc) 73 # creating a new setuptools install 74 cls.ppy_st = os.path.join(cls.rdir, "psetuptools") 75 if salt.utils.platform.is_windows(): 76 cls.bin_st = os.path.join(cls.ppy_st, "Scripts") 77 cls.py_st = os.path.join(cls.bin_st, "python") 78 else: 79 cls.bin_st = os.path.join(cls.ppy_st, "bin") 80 cls.py_st = os.path.join(cls.bin_st, "python") 81 # `--no-site-packages` has been deprecated 82 # https://virtualenv.pypa.io/en/stable/reference/#cmdoption-no-site-packages 83 subprocess.check_call( 84 [salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), cls.ppy_st] 85 ) 86 # Setuptools >=53.0.0 no longer has easy_install 87 # Between 50.0.0 and 53.0.0 it has problems with salt, use an older version 88 subprocess.check_call( 89 [os.path.join(cls.bin_st, "pip"), "install", "-U", "setuptools<50.0.0"] 90 ) 91 # distribute has been merged back in to setuptools as of v0.7. So, no 92 # need to upgrade distribute, but this seems to be the only way to get 93 # the binary in the right place 94 # https://packaging.python.org/key_projects/#setuptools 95 # Additionally, this part may fail if the certificate store is outdated 96 # on Windows, as it would be in a fresh installation for example. The 97 # following commands will fix that. This should be part of the golden 98 # images. (https://github.com/saltstack/salt-jenkins/pull/1479) 99 # certutil -generateSSTFromWU roots.sst 100 # powershell "(Get-ChildItem -Path .\roots.sst) | Import-Certificate -CertStoreLocation Cert:\LocalMachine\Root" 101 subprocess.check_call( 102 [os.path.join(cls.bin_st, "easy_install"), "-U", "distribute"] 103 ) 104 105 def setUp(self): 106 if salt.utils.platform.is_darwin(): 107 self.patched_environ = patched_environ(__cleanup__=["__PYVENV_LAUNCHER__"]) 108 self.patched_environ.__enter__() 109 self.addCleanup(self.patched_environ.__exit__) 110 111 super().setUp() 112 self._remove_dir() 113 shutil.copytree(self.root, self.tdir) 114 115 for idx in BOOT_INIT: 116 path = os.path.join(self.rdir, "{}_bootstrap.py".format(idx)) 117 for fname in BOOT_INIT[idx]: 118 shutil.copy2(path, os.path.join(self.tdir, fname)) 119 120 def tearDown(self): 121 super().tearDown() 122 self._remove_dir() 123 124 def _remove_dir(self): 125 if os.path.isdir(self.tdir): 126 shutil.rmtree(self.tdir) 127 128 129@skipIf( 130 salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES) is None, 131 "The 'virtualenv' packaged needs to be installed", 132) 133@pytest.mark.requires_network 134class BuildoutTestCase(Base): 135 @pytest.mark.slow_test 136 def test_onlyif_unless(self): 137 b_dir = os.path.join(self.tdir, "b") 138 ret = buildout.buildout(b_dir, onlyif=RUNTIME_VARS.SHELL_FALSE_PATH) 139 self.assertTrue(ret["comment"] == "onlyif condition is false") 140 self.assertTrue(ret["status"] is True) 141 ret = buildout.buildout(b_dir, unless=RUNTIME_VARS.SHELL_TRUE_PATH) 142 self.assertTrue(ret["comment"] == "unless condition is true") 143 self.assertTrue(ret["status"] is True) 144 145 @pytest.mark.slow_test 146 def test_salt_callback(self): 147 @buildout._salt_callback 148 def callback1(a, b=1): 149 for i in buildout.LOG.levels: 150 getattr(buildout.LOG, i)("{}bar".format(i[0])) 151 return "foo" 152 153 def callback2(a, b=1): 154 raise Exception("foo") 155 156 # pylint: disable=invalid-sequence-index 157 ret1 = callback1(1, b=3) 158 # These lines are throwing pylint errors - disabling for now since we are skipping 159 # these tests 160 # self.assertEqual(ret1['status'], True) 161 # self.assertEqual(ret1['logs_by_level']['warn'], ['wbar']) 162 # self.assertEqual(ret1['comment'], '') 163 # These lines are throwing pylint errors - disabling for now since we are skipping 164 # these tests 165 # self.assertTrue( 166 # u'' 167 # u'OUTPUT:\n' 168 # u'foo\n' 169 # u'' 170 # in ret1['outlog'] 171 # ) 172 173 # These lines are throwing pylint errors - disabling for now since we are skipping 174 # these tests 175 # self.assertTrue(u'Log summary:\n' in ret1['outlog']) 176 # These lines are throwing pylint errors - disabling for now since we are skipping 177 # these tests 178 # self.assertTrue( 179 # u'INFO: ibar\n' 180 # u'WARN: wbar\n' 181 # u'DEBUG: dbar\n' 182 # u'ERROR: ebar\n' 183 # in ret1['outlog'] 184 # ) 185 # These lines are throwing pylint errors - disabling for now since we are skipping 186 # these tests 187 # self.assertTrue('by level' in ret1['outlog_by_level']) 188 # self.assertEqual(ret1['out'], 'foo') 189 ret2 = buildout._salt_callback(callback2)(2, b=6) 190 self.assertEqual(ret2["status"], False) 191 self.assertTrue(ret2["logs_by_level"]["error"][0].startswith("Traceback")) 192 self.assertTrue("Unexpected response from buildout" in ret2["comment"]) 193 self.assertEqual(ret2["out"], None) 194 for l in buildout.LOG.levels: 195 self.assertTrue(0 == len(buildout.LOG.by_level[l])) 196 # pylint: enable=invalid-sequence-index 197 198 @pytest.mark.slow_test 199 def test_get_bootstrap_url(self): 200 for path in [ 201 os.path.join(self.tdir, "var/ver/1/dumppicked"), 202 os.path.join(self.tdir, "var/ver/1/bootstrap"), 203 os.path.join(self.tdir, "var/ver/1/versions"), 204 ]: 205 self.assertEqual( 206 buildout._URL_VERSIONS[1], 207 buildout._get_bootstrap_url(path), 208 "b1 url for {}".format(path), 209 ) 210 for path in [ 211 os.path.join(self.tdir, "/non/existing"), 212 os.path.join(self.tdir, "var/ver/2/versions"), 213 os.path.join(self.tdir, "var/ver/2/bootstrap"), 214 os.path.join(self.tdir, "var/ver/2/default"), 215 ]: 216 self.assertEqual( 217 buildout._URL_VERSIONS[2], 218 buildout._get_bootstrap_url(path), 219 "b2 url for {}".format(path), 220 ) 221 222 @pytest.mark.slow_test 223 def test_get_buildout_ver(self): 224 for path in [ 225 os.path.join(self.tdir, "var/ver/1/dumppicked"), 226 os.path.join(self.tdir, "var/ver/1/bootstrap"), 227 os.path.join(self.tdir, "var/ver/1/versions"), 228 ]: 229 self.assertEqual( 230 1, buildout._get_buildout_ver(path), "1 for {}".format(path) 231 ) 232 for path in [ 233 os.path.join(self.tdir, "/non/existing"), 234 os.path.join(self.tdir, "var/ver/2/versions"), 235 os.path.join(self.tdir, "var/ver/2/bootstrap"), 236 os.path.join(self.tdir, "var/ver/2/default"), 237 ]: 238 self.assertEqual( 239 2, buildout._get_buildout_ver(path), "2 for {}".format(path) 240 ) 241 242 @pytest.mark.slow_test 243 def test_get_bootstrap_content(self): 244 self.assertEqual( 245 "", 246 buildout._get_bootstrap_content(os.path.join(self.tdir, "non", "existing")), 247 ) 248 self.assertEqual( 249 "", 250 buildout._get_bootstrap_content(os.path.join(self.tdir, "var", "tb", "1")), 251 ) 252 self.assertEqual( 253 "foo{}".format(os.linesep), 254 buildout._get_bootstrap_content(os.path.join(self.tdir, "var", "tb", "2")), 255 ) 256 257 @pytest.mark.slow_test 258 def test_logger_clean(self): 259 buildout.LOG.clear() 260 # nothing in there 261 self.assertTrue( 262 True 263 not in [len(buildout.LOG.by_level[a]) > 0 for a in buildout.LOG.by_level] 264 ) 265 buildout.LOG.info("foo") 266 self.assertTrue( 267 True in [len(buildout.LOG.by_level[a]) > 0 for a in buildout.LOG.by_level] 268 ) 269 buildout.LOG.clear() 270 self.assertTrue( 271 True 272 not in [len(buildout.LOG.by_level[a]) > 0 for a in buildout.LOG.by_level] 273 ) 274 275 @pytest.mark.slow_test 276 def test_logger_loggers(self): 277 buildout.LOG.clear() 278 # nothing in there 279 for i in buildout.LOG.levels: 280 getattr(buildout.LOG, i)("foo") 281 getattr(buildout.LOG, i)("bar") 282 getattr(buildout.LOG, i)("moo") 283 self.assertTrue(len(buildout.LOG.by_level[i]) == 3) 284 self.assertEqual(buildout.LOG.by_level[i][0], "foo") 285 self.assertEqual(buildout.LOG.by_level[i][-1], "moo") 286 287 @pytest.mark.slow_test 288 def test__find_cfgs(self): 289 result = sorted( 290 [a.replace(self.root, "") for a in buildout._find_cfgs(self.root)] 291 ) 292 assertlist = sorted( 293 [ 294 os.path.join(os.sep, "buildout.cfg"), 295 os.path.join(os.sep, "c", "buildout.cfg"), 296 os.path.join(os.sep, "etc", "buildout.cfg"), 297 os.path.join(os.sep, "e", "buildout.cfg"), 298 os.path.join(os.sep, "b", "buildout.cfg"), 299 os.path.join(os.sep, "b", "bdistribute", "buildout.cfg"), 300 os.path.join(os.sep, "b", "b2", "buildout.cfg"), 301 os.path.join(os.sep, "foo", "buildout.cfg"), 302 ] 303 ) 304 self.assertEqual(result, assertlist) 305 306 def skip_test_upgrade_bootstrap(self): 307 b_dir = os.path.join(self.tdir, "b") 308 bpy = os.path.join(b_dir, "bootstrap.py") 309 buildout.upgrade_bootstrap(b_dir) 310 time1 = os.stat(bpy).st_mtime 311 with salt.utils.files.fopen(bpy) as fic: 312 data = fic.read() 313 self.assertTrue("setdefaulttimeout(2)" in data) 314 flag = os.path.join(b_dir, ".buildout", "2.updated_bootstrap") 315 self.assertTrue(os.path.exists(flag)) 316 buildout.upgrade_bootstrap(b_dir, buildout_ver=1) 317 time2 = os.stat(bpy).st_mtime 318 with salt.utils.files.fopen(bpy) as fic: 319 data = fic.read() 320 self.assertTrue("setdefaulttimeout(2)" in data) 321 flag = os.path.join(b_dir, ".buildout", "1.updated_bootstrap") 322 self.assertTrue(os.path.exists(flag)) 323 buildout.upgrade_bootstrap(b_dir, buildout_ver=1) 324 time3 = os.stat(bpy).st_mtime 325 self.assertNotEqual(time2, time1) 326 self.assertEqual(time2, time3) 327 328 329@skipIf( 330 salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES) is None, 331 "The 'virtualenv' packaged needs to be installed", 332) 333@pytest.mark.requires_network 334class BuildoutOnlineTestCase(Base): 335 @classmethod 336 def setUpClass(cls): 337 super().setUpClass() 338 cls.ppy_dis = os.path.join(cls.rdir, "pdistribute") 339 cls.ppy_blank = os.path.join(cls.rdir, "pblank") 340 cls.py_dis = os.path.join(cls.ppy_dis, "bin", "python") 341 cls.py_blank = os.path.join(cls.ppy_blank, "bin", "python") 342 # creating a distribute based install 343 try: 344 # `--no-site-packages` has been deprecated 345 # https://virtualenv.pypa.io/en/stable/reference/#cmdoption-no-site-packages 346 subprocess.check_call( 347 [ 348 salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), 349 "--no-setuptools", 350 "--no-pip", 351 cls.ppy_dis, 352 ] 353 ) 354 except subprocess.CalledProcessError: 355 subprocess.check_call( 356 [salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), cls.ppy.dis] 357 ) 358 359 url = ( 360 "https://pypi.python.org/packages/source" 361 "/d/distribute/distribute-0.6.43.tar.gz" 362 ) 363 download_to( 364 url, 365 os.path.join(cls.ppy_dis, "distribute-0.6.43.tar.gz"), 366 ) 367 368 subprocess.check_call( 369 [ 370 "tar", 371 "-C", 372 cls.ppy_dis, 373 "-xzvf", 374 "{}/distribute-0.6.43.tar.gz".format(cls.ppy_dis), 375 ] 376 ) 377 378 subprocess.check_call( 379 [ 380 "{}/bin/python".format(cls.ppy_dis), 381 "{}/distribute-0.6.43/setup.py".format(cls.ppy_dis), 382 "install", 383 ] 384 ) 385 386 # creating a blank based install 387 try: 388 subprocess.check_call( 389 [ 390 salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), 391 "--no-setuptools", 392 "--no-pip", 393 cls.ppy_blank, 394 ] 395 ) 396 except subprocess.CalledProcessError: 397 subprocess.check_call( 398 [ 399 salt.utils.path.which_bin(KNOWN_VIRTUALENV_BINARY_NAMES), 400 cls.ppy_blank, 401 ] 402 ) 403 404 @skipIf(True, "TODO this test should probably be fixed") 405 def test_buildout_bootstrap(self): 406 b_dir = os.path.join(self.tdir, "b") 407 bd_dir = os.path.join(self.tdir, "b", "bdistribute") 408 b2_dir = os.path.join(self.tdir, "b", "b2") 409 self.assertTrue(buildout._has_old_distribute(self.py_dis)) 410 # this is too hard to check as on debian & other where old 411 # packages are present (virtualenv), we can't have 412 # a clean site-packages 413 # self.assertFalse(buildout._has_old_distribute(self.py_blank)) 414 self.assertFalse(buildout._has_old_distribute(self.py_st)) 415 self.assertFalse(buildout._has_setuptools7(self.py_dis)) 416 self.assertTrue(buildout._has_setuptools7(self.py_st)) 417 self.assertFalse(buildout._has_setuptools7(self.py_blank)) 418 419 ret = buildout.bootstrap(bd_dir, buildout_ver=1, python=self.py_dis) 420 comment = ret["outlog"] 421 self.assertTrue("--distribute" in comment) 422 self.assertTrue("Generated script" in comment) 423 424 ret = buildout.bootstrap(b_dir, buildout_ver=1, python=self.py_blank) 425 comment = ret["outlog"] 426 # as we may have old packages, this test the two 427 # behaviors (failure with old setuptools/distribute) 428 self.assertTrue( 429 ("Got " in comment and "Generated script" in comment) 430 or ("setuptools>=0.7" in comment) 431 ) 432 433 ret = buildout.bootstrap(b_dir, buildout_ver=2, python=self.py_blank) 434 comment = ret["outlog"] 435 self.assertTrue( 436 ("setuptools" in comment and "Generated script" in comment) 437 or ("setuptools>=0.7" in comment) 438 ) 439 440 ret = buildout.bootstrap(b_dir, buildout_ver=2, python=self.py_st) 441 comment = ret["outlog"] 442 self.assertTrue( 443 ("setuptools" in comment and "Generated script" in comment) 444 or ("setuptools>=0.7" in comment) 445 ) 446 447 ret = buildout.bootstrap(b2_dir, buildout_ver=2, python=self.py_st) 448 comment = ret["outlog"] 449 self.assertTrue( 450 ("setuptools" in comment and "Creating directory" in comment) 451 or ("setuptools>=0.7" in comment) 452 ) 453 454 @pytest.mark.slow_test 455 def test_run_buildout(self): 456 if salt.modules.virtualenv_mod.virtualenv_ver(self.ppy_st) >= (20, 0, 0): 457 self.skipTest( 458 "Skiping until upstream resolved" 459 " https://github.com/pypa/virtualenv/issues/1715" 460 ) 461 462 b_dir = os.path.join(self.tdir, "b") 463 ret = buildout.bootstrap(b_dir, buildout_ver=2, python=self.py_st) 464 self.assertTrue(ret["status"]) 465 ret = buildout.run_buildout(b_dir, parts=["a", "b"]) 466 out = ret["out"] 467 self.assertTrue("Installing a" in out) 468 self.assertTrue("Installing b" in out) 469 470 @pytest.mark.slow_test 471 def test_buildout(self): 472 if salt.modules.virtualenv_mod.virtualenv_ver(self.ppy_st) >= (20, 0, 0): 473 self.skipTest( 474 "Skiping until upstream resolved" 475 " https://github.com/pypa/virtualenv/issues/1715" 476 ) 477 478 b_dir = os.path.join(self.tdir, "b") 479 ret = buildout.buildout(b_dir, buildout_ver=2, python=self.py_st) 480 self.assertTrue(ret["status"]) 481 out = ret["out"] 482 comment = ret["comment"] 483 self.assertTrue(ret["status"]) 484 self.assertTrue("Creating directory" in out) 485 self.assertTrue("Installing a." in out) 486 self.assertTrue("{} bootstrap.py".format(self.py_st) in comment) 487 self.assertTrue("buildout -c buildout.cfg" in comment) 488 ret = buildout.buildout( 489 b_dir, parts=["a", "b", "c"], buildout_ver=2, python=self.py_st 490 ) 491 outlog = ret["outlog"] 492 out = ret["out"] 493 comment = ret["comment"] 494 self.assertTrue("Installing single part: a" in outlog) 495 self.assertTrue("buildout -c buildout.cfg -N install a" in comment) 496 self.assertTrue("Installing b." in out) 497 self.assertTrue("Installing c." in out) 498 ret = buildout.buildout( 499 b_dir, parts=["a", "b", "c"], buildout_ver=2, newest=True, python=self.py_st 500 ) 501 outlog = ret["outlog"] 502 out = ret["out"] 503 comment = ret["comment"] 504 self.assertTrue("buildout -c buildout.cfg -n install a" in comment) 505 506 507# TODO: Is this test even still needed? 508class BuildoutAPITestCase(TestCase): 509 def test_merge(self): 510 buildout.LOG.clear() 511 buildout.LOG.info("àé") 512 buildout.LOG.info("àé") 513 buildout.LOG.error("àé") 514 buildout.LOG.error("àé") 515 ret1 = buildout._set_status({}, out="éà") 516 uret1 = buildout._set_status({}, out="éà") 517 buildout.LOG.clear() 518 buildout.LOG.info("ççàé") 519 buildout.LOG.info("ççàé") 520 buildout.LOG.error("ççàé") 521 buildout.LOG.error("ççàé") 522 ret2 = buildout._set_status({}, out="çéà") 523 uret2 = buildout._set_status({}, out="çéà") 524 uretm = buildout._merge_statuses([ret1, uret1, ret2, uret2]) 525 for ret in ret1, uret1, ret2, uret2: 526 out = ret["out"] 527 if not isinstance(ret["out"], str): 528 out = ret["out"].decode("utf-8") 529 530 for out in ["àé", "ççàé"]: 531 self.assertTrue(out in uretm["logs_by_level"]["info"]) 532 self.assertTrue(out in uretm["outlog_by_level"]) 533 534 def test_setup(self): 535 buildout.LOG.clear() 536 buildout.LOG.info("àé") 537 buildout.LOG.info("àé") 538 buildout.LOG.error("àé") 539 buildout.LOG.error("àé") 540 ret = buildout._set_status({}, out="éà") 541 uret = buildout._set_status({}, out="éà") 542 self.assertTrue(ret["outlog"] == uret["outlog"]) 543 self.assertTrue("àé" in uret["outlog_by_level"]) 544