1import os.path 2 3import pytest 4import salt.modules.archive as archive 5from salt.exceptions import CommandNotFoundError 6from tests.support.mock import MagicMock, patch 7 8 9class ZipFileMock(MagicMock): 10 def __init__(self, files=None, **kwargs): # pylint: disable=W0231 11 if files is None: 12 files = ["salt"] 13 MagicMock.__init__(self, **kwargs) 14 self._files = files 15 self.external_attr = 0o0777 << 16 16 17 def namelist(self): 18 return self._files 19 20 21@pytest.fixture 22def configure_loader_modules(): 23 return {archive: {"__grains__": {"id": 0}}} 24 25 26def test_tar(): 27 with patch("glob.glob", lambda pathname: [pathname]): 28 with patch("salt.utils.path.which", lambda exe: exe): 29 mock = MagicMock(return_value="salt") 30 with patch.dict(archive.__salt__, {"cmd.run": mock}): 31 ret = archive.tar( 32 "-zcvf", 33 "foo.tar", 34 ["/tmp/something-to-compress-1", "/tmp/something-to-compress-2"], 35 template=None, 36 ) 37 assert ["salt"] == ret 38 mock.assert_called_once_with( 39 [ 40 "tar", 41 "-zcvf", 42 "foo.tar", 43 "/tmp/something-to-compress-1", 44 "/tmp/something-to-compress-2", 45 ], 46 runas=None, 47 python_shell=False, 48 template=None, 49 cwd=None, 50 ) 51 52 mock = MagicMock(return_value="salt") 53 with patch.dict(archive.__salt__, {"cmd.run": mock}): 54 ret = archive.tar( 55 "-zcvf", 56 "foo.tar", 57 "/tmp/something-to-compress-1,/tmp/something-to-compress-2", 58 template=None, 59 ) 60 assert ["salt"] == ret 61 mock.assert_called_once_with( 62 [ 63 "tar", 64 "-zcvf", 65 "foo.tar", 66 "/tmp/something-to-compress-1", 67 "/tmp/something-to-compress-2", 68 ], 69 runas=None, 70 python_shell=False, 71 template=None, 72 cwd=None, 73 ) 74 75 76def test_tar_raises_exception_if_not_found(): 77 with patch("salt.utils.path.which", lambda exe: None): 78 mock = MagicMock(return_value="salt") 79 with patch.dict(archive.__salt__, {"cmd.run": mock}): 80 with pytest.raises(CommandNotFoundError): 81 archive.tar("zxvf", "foo.tar", "/tmp/something-to-compress") 82 assert not mock.called 83 84 85def test_gzip(): 86 mock = MagicMock(return_value="salt") 87 with patch.dict(archive.__salt__, {"cmd.run": mock}): 88 with patch("salt.utils.path.which", lambda exe: exe): 89 ret = archive.gzip("/tmp/something-to-compress") 90 assert ["salt"] == ret 91 mock.assert_called_once_with( 92 ["gzip", "/tmp/something-to-compress"], 93 runas=None, 94 python_shell=False, 95 template=None, 96 ) 97 98 99def test_gzip_raises_exception_if_not_found(): 100 mock = MagicMock(return_value="salt") 101 with patch.dict(archive.__salt__, {"cmd.run": mock}): 102 with patch("salt.utils.path.which", lambda exe: None): 103 with pytest.raises(CommandNotFoundError): 104 archive.gzip("/tmp/something-to-compress") 105 assert not mock.called 106 107 108def test_gunzip(): 109 mock = MagicMock(return_value="salt") 110 with patch.dict(archive.__salt__, {"cmd.run": mock}): 111 with patch("salt.utils.path.which", lambda exe: exe): 112 ret = archive.gunzip("/tmp/something-to-decompress.tar.gz") 113 assert ["salt"] == ret 114 mock.assert_called_once_with( 115 ["gunzip", "/tmp/something-to-decompress.tar.gz"], 116 runas=None, 117 python_shell=False, 118 template=None, 119 ) 120 121 122def test_gunzip_raises_exception_if_not_found(): 123 mock = MagicMock(return_value="salt") 124 with patch.dict(archive.__salt__, {"cmd.run": mock}): 125 with patch("salt.utils.path.which", lambda exe: None): 126 with pytest.raises(CommandNotFoundError): 127 archive.gunzip("/tmp/something-to-decompress.tar.gz") 128 assert not mock.called 129 130 131def test_cmd_zip(): 132 with patch("glob.glob", lambda pathname: [pathname]): 133 with patch("salt.utils.path.which", lambda exe: exe): 134 mock = MagicMock(return_value="salt") 135 with patch.dict(archive.__salt__, {"cmd.run": mock}): 136 ret = archive.cmd_zip( 137 "/tmp/salt.{{grains.id}}.zip", 138 "/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 139 template="jinja", 140 ) 141 assert ["salt"] == ret 142 mock.assert_called_once_with( 143 [ 144 "zip", 145 "-r", 146 "/tmp/salt.{{grains.id}}.zip", 147 "/tmp/tmpePe8yO", 148 "/tmp/tmpLeSw1A", 149 ], 150 runas=None, 151 python_shell=False, 152 template="jinja", 153 cwd=None, 154 ) 155 156 mock = MagicMock(return_value="salt") 157 with patch.dict(archive.__salt__, {"cmd.run": mock}): 158 ret = archive.cmd_zip( 159 "/tmp/salt.{{grains.id}}.zip", 160 ["/tmp/tmpePe8yO", "/tmp/tmpLeSw1A"], 161 template="jinja", 162 ) 163 assert ["salt"] == ret 164 mock.assert_called_once_with( 165 [ 166 "zip", 167 "-r", 168 "/tmp/salt.{{grains.id}}.zip", 169 "/tmp/tmpePe8yO", 170 "/tmp/tmpLeSw1A", 171 ], 172 runas=None, 173 python_shell=False, 174 template="jinja", 175 cwd=None, 176 ) 177 178 179def test_zip(): 180 with patch("glob.glob", lambda pathname: [pathname]): 181 with patch.multiple( 182 os.path, 183 **{ 184 "isdir": MagicMock(return_value=False), 185 "exists": MagicMock(return_value=True), 186 } 187 ): 188 with patch("zipfile.ZipFile", MagicMock()): 189 ret = archive.zip_( 190 "/tmp/salt.{{grains.id}}.zip", 191 "/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 192 template="jinja", 193 ) 194 expected = [ 195 os.path.join("tmp", "tmpePe8yO"), 196 os.path.join("tmp", "tmpLeSw1A"), 197 ] 198 assert expected == ret 199 200 201def test_zip_raises_exception_if_not_found(): 202 mock = MagicMock(return_value="salt") 203 with patch.dict(archive.__salt__, {"cmd.run": mock}): 204 with patch("salt.utils.path.which", lambda exe: None): 205 with pytest.raises(CommandNotFoundError): 206 archive.cmd_zip( 207 "/tmp/salt.{{grains.id}}.zip", 208 "/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 209 template="jinja", 210 ) 211 assert not mock.called 212 213 214def test_cmd_unzip(): 215 def _get_mock(): 216 """ 217 Create a new MagicMock for each scenario in this test, so that 218 assert_called_once_with doesn't complain that the same mock object 219 is called more than once. 220 """ 221 return MagicMock( 222 return_value={"stdout": "salt", "stderr": "", "pid": 12345, "retcode": 0} 223 ) 224 225 with patch("salt.utils.path.which", lambda exe: exe): 226 227 mock = _get_mock() 228 with patch.dict(archive.__salt__, {"cmd.run_all": mock}): 229 ret = archive.cmd_unzip( 230 "/tmp/salt.{{grains.id}}.zip", 231 "/tmp/dest", 232 excludes="/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 233 template="jinja", 234 ) 235 assert ["salt"] == ret 236 mock.assert_called_once_with( 237 [ 238 "unzip", 239 "/tmp/salt.{{grains.id}}.zip", 240 "-d", 241 "/tmp/dest", 242 "-x", 243 "/tmp/tmpePe8yO", 244 "/tmp/tmpLeSw1A", 245 ], 246 output_loglevel="debug", 247 python_shell=False, 248 redirect_stderr=True, 249 runas=None, 250 template="jinja", 251 ) 252 253 mock = _get_mock() 254 with patch.dict(archive.__salt__, {"cmd.run_all": mock}): 255 ret = archive.cmd_unzip( 256 "/tmp/salt.{{grains.id}}.zip", 257 "/tmp/dest", 258 excludes=["/tmp/tmpePe8yO", "/tmp/tmpLeSw1A"], 259 template="jinja", 260 ) 261 assert ["salt"] == ret 262 mock.assert_called_once_with( 263 [ 264 "unzip", 265 "/tmp/salt.{{grains.id}}.zip", 266 "-d", 267 "/tmp/dest", 268 "-x", 269 "/tmp/tmpePe8yO", 270 "/tmp/tmpLeSw1A", 271 ], 272 output_loglevel="debug", 273 python_shell=False, 274 redirect_stderr=True, 275 runas=None, 276 template="jinja", 277 ) 278 279 mock = _get_mock() 280 with patch.dict(archive.__salt__, {"cmd.run_all": mock}): 281 ret = archive.cmd_unzip( 282 "/tmp/salt.{{grains.id}}.zip", 283 "/tmp/dest", 284 excludes="/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 285 template="jinja", 286 options="-fo", 287 ) 288 assert ["salt"] == ret 289 mock.assert_called_once_with( 290 [ 291 "unzip", 292 "-fo", 293 "/tmp/salt.{{grains.id}}.zip", 294 "-d", 295 "/tmp/dest", 296 "-x", 297 "/tmp/tmpePe8yO", 298 "/tmp/tmpLeSw1A", 299 ], 300 output_loglevel="debug", 301 python_shell=False, 302 redirect_stderr=True, 303 runas=None, 304 template="jinja", 305 ) 306 307 mock = _get_mock() 308 with patch.dict(archive.__salt__, {"cmd.run_all": mock}): 309 ret = archive.cmd_unzip( 310 "/tmp/salt.{{grains.id}}.zip", 311 "/tmp/dest", 312 excludes=["/tmp/tmpePe8yO", "/tmp/tmpLeSw1A"], 313 template="jinja", 314 options="-fo", 315 ) 316 assert ["salt"] == ret 317 mock.assert_called_once_with( 318 [ 319 "unzip", 320 "-fo", 321 "/tmp/salt.{{grains.id}}.zip", 322 "-d", 323 "/tmp/dest", 324 "-x", 325 "/tmp/tmpePe8yO", 326 "/tmp/tmpLeSw1A", 327 ], 328 output_loglevel="debug", 329 python_shell=False, 330 redirect_stderr=True, 331 runas=None, 332 template="jinja", 333 ) 334 335 mock = _get_mock() 336 with patch.dict(archive.__salt__, {"cmd.run_all": mock}): 337 ret = archive.cmd_unzip( 338 "/tmp/salt.{{grains.id}}.zip", 339 "/tmp/dest", 340 excludes=["/tmp/tmpePe8yO", "/tmp/tmpLeSw1A"], 341 template="jinja", 342 options="-fo", 343 password="asdf", 344 ) 345 assert ["salt"] == ret 346 mock.assert_called_once_with( 347 [ 348 "unzip", 349 "-P", 350 "asdf", 351 "-fo", 352 "/tmp/salt.{{grains.id}}.zip", 353 "-d", 354 "/tmp/dest", 355 "-x", 356 "/tmp/tmpePe8yO", 357 "/tmp/tmpLeSw1A", 358 ], 359 output_loglevel="quiet", 360 python_shell=False, 361 redirect_stderr=True, 362 runas=None, 363 template="jinja", 364 ) 365 366 367def test_unzip(): 368 mock = ZipFileMock() 369 with patch("zipfile.ZipFile", mock): 370 ret = archive.unzip( 371 "/tmp/salt.{{grains.id}}.zip", 372 "/tmp/dest", 373 excludes="/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 374 template="jinja", 375 extract_perms=False, 376 ) 377 assert ["salt"] == ret 378 379 380def test_unzip_raises_exception_if_not_found(): 381 mock = MagicMock(return_value="salt") 382 with patch.dict(archive.__salt__, {"cmd.run": mock}): 383 with patch("salt.utils.path.which", lambda exe: None): 384 with pytest.raises(CommandNotFoundError): 385 archive.cmd_unzip( 386 "/tmp/salt.{{grains.id}}.zip", 387 "/tmp/dest", 388 excludes="/tmp/tmpePe8yO,/tmp/tmpLeSw1A", 389 template="jinja", 390 ) 391 assert not mock.called 392 393 394def test_rar(): 395 with patch("glob.glob", lambda pathname: [pathname]): 396 with patch("salt.utils.path.which", lambda exe: exe): 397 mock = MagicMock(return_value="salt") 398 with patch.dict(archive.__salt__, {"cmd.run": mock}): 399 ret = archive.rar( 400 "/tmp/rarfile.rar", "/tmp/sourcefile1,/tmp/sourcefile2" 401 ) 402 assert ["salt"] == ret 403 mock.assert_called_once_with( 404 [ 405 "rar", 406 "a", 407 "-idp", 408 "/tmp/rarfile.rar", 409 "/tmp/sourcefile1", 410 "/tmp/sourcefile2", 411 ], 412 runas=None, 413 python_shell=False, 414 template=None, 415 cwd=None, 416 ) 417 418 mock = MagicMock(return_value="salt") 419 with patch.dict(archive.__salt__, {"cmd.run": mock}): 420 ret = archive.rar( 421 "/tmp/rarfile.rar", ["/tmp/sourcefile1", "/tmp/sourcefile2"] 422 ) 423 assert ["salt"] == ret 424 mock.assert_called_once_with( 425 [ 426 "rar", 427 "a", 428 "-idp", 429 "/tmp/rarfile.rar", 430 "/tmp/sourcefile1", 431 "/tmp/sourcefile2", 432 ], 433 runas=None, 434 python_shell=False, 435 template=None, 436 cwd=None, 437 ) 438 439 440def test_rar_raises_exception_if_not_found(): 441 mock = MagicMock(return_value="salt") 442 with patch.dict(archive.__salt__, {"cmd.run": mock}): 443 with patch("salt.utils.path.which", lambda exe: None): 444 with pytest.raises(CommandNotFoundError): 445 archive.rar("/tmp/rarfile.rar", "/tmp/sourcefile1,/tmp/sourcefile2") 446 assert not mock.called 447 448 449@pytest.mark.skip_if_binaries_missing("unrar", "rar", reason="unrar not installed") 450def test_unrar(): 451 with patch( 452 "salt.utils.path.which_bin", 453 lambda exe: exe[0] if isinstance(exe, (list, tuple)) else exe, 454 ): 455 with patch( 456 "salt.utils.path.which", 457 lambda exe: exe[0] if isinstance(exe, (list, tuple)) else exe, 458 ): 459 mock = MagicMock(return_value="salt") 460 with patch.dict(archive.__salt__, {"cmd.run": mock}): 461 ret = archive.unrar( 462 "/tmp/rarfile.rar", "/home/strongbad/", excludes="file_1,file_2" 463 ) 464 assert ["salt"] == ret 465 mock.assert_called_once_with( 466 [ 467 "unrar", 468 "x", 469 "-idp", 470 "/tmp/rarfile.rar", 471 "-x", 472 "file_1", 473 "-x", 474 "file_2", 475 "/home/strongbad/", 476 ], 477 runas=None, 478 python_shell=False, 479 template=None, 480 ) 481 482 mock = MagicMock(return_value="salt") 483 with patch.dict(archive.__salt__, {"cmd.run": mock}): 484 ret = archive.unrar( 485 "/tmp/rarfile.rar", 486 "/home/strongbad/", 487 excludes=["file_1", "file_2"], 488 ) 489 assert ["salt"] == ret 490 mock.assert_called_once_with( 491 [ 492 "unrar", 493 "x", 494 "-idp", 495 "/tmp/rarfile.rar", 496 "-x", 497 "file_1", 498 "-x", 499 "file_2", 500 "/home/strongbad/", 501 ], 502 runas=None, 503 python_shell=False, 504 template=None, 505 ) 506 507 508def test_unrar_raises_exception_if_not_found(): 509 with patch("salt.utils.path.which_bin", lambda exe: None): 510 mock = MagicMock(return_value="salt") 511 with patch.dict(archive.__salt__, {"cmd.run": mock}): 512 with pytest.raises(CommandNotFoundError): 513 archive.unrar( 514 "/tmp/rarfile.rar", 515 "/home/strongbad/", 516 ) 517 assert not mock.called 518 519 520def test_trim_files(): 521 with patch("salt.utils.path.which_bin", lambda exe: exe): 522 source = "file.tar.gz" 523 tmp_dir = "temp" 524 files = [ 525 "\n".join(["file1"] * 200), 526 "\n".join(["file2"] * 200), 527 "this\nthat\nother", 528 "this\nthat\nother", 529 "this\nthat\nother", 530 ] 531 trim_opt = [True, False, 3, 1, 0] 532 trim_100 = ["file1"] * 100 533 trim_100.append("List trimmed after 100 files.") 534 expected = [ 535 trim_100, 536 ["file2"] * 200, 537 ["this", "that", "other"], 538 ["this", "List trimmed after 1 files."], 539 ["List trimmed after 0 files."], 540 ] 541 542 for test_files, test_trim_opts, test_expected in zip(files, trim_opt, expected): 543 with patch.dict( 544 archive.__salt__, {"cmd.run": MagicMock(return_value=test_files)} 545 ): 546 ret = archive.unrar(source, tmp_dir, trim_output=test_trim_opts) 547 assert ret == test_expected 548