| LEFT | RIGHT |
| 1 # Copyright (C) 2003 Python Software Foundation | 1 # Copyright (C) 2003 Python Software Foundation |
| 2 | 2 |
| 3 import unittest | 3 import unittest |
| 4 import shutil | 4 import shutil |
| 5 import tempfile | 5 import tempfile |
| 6 import sys | 6 import sys |
| 7 import stat | 7 import stat |
| 8 import os | 8 import os |
| 9 import os.path | 9 import os.path |
| 10 import errno | 10 import errno |
| (...skipping 98 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 109 return d | 109 return d |
| 110 | 110 |
| 111 def test_rmtree_works_on_bytes(self): | 111 def test_rmtree_works_on_bytes(self): |
| 112 tmp = self.mkdtemp() | 112 tmp = self.mkdtemp() |
| 113 victim = os.path.join(tmp, 'killme') | 113 victim = os.path.join(tmp, 'killme') |
| 114 os.mkdir(victim) | 114 os.mkdir(victim) |
| 115 write_file(os.path.join(victim, 'somefile'), 'foo') | 115 write_file(os.path.join(victim, 'somefile'), 'foo') |
| 116 victim = os.fsencode(victim) | 116 victim = os.fsencode(victim) |
| 117 self.assertIsInstance(victim, bytes) | 117 self.assertIsInstance(victim, bytes) |
| 118 shutil.rmtree(victim) | 118 shutil.rmtree(victim) |
| 119 |
| 120 @support.skip_unless_symlink |
| 121 def test_rmtree_fails_on_symlink(self): |
| 122 tmp = self.mkdtemp() |
| 123 dir_ = os.path.join(tmp, 'dir') |
| 124 os.mkdir(dir_) |
| 125 link = os.path.join(tmp, 'link') |
| 126 os.symlink(dir_, link) |
| 127 self.assertRaises(OSError, shutil.rmtree, link) |
| 128 self.assertTrue(os.path.exists(dir_)) |
| 129 |
| 130 @support.skip_unless_symlink |
| 131 def test_rmtree_works_on_symlinks(self): |
| 132 tmp = self.mkdtemp() |
| 133 dir1 = os.path.join(tmp, 'dir1') |
| 134 dir2 = os.path.join(dir1, 'dir2') |
| 135 dir3 = os.path.join(tmp, 'dir3') |
| 136 for d in dir1, dir2, dir3: |
| 137 os.mkdir(d) |
| 138 file1 = os.path.join(tmp, 'file1') |
| 139 write_file(file1, 'foo') |
| 140 link1 = os.path.join(dir1, 'link1') |
| 141 os.symlink(dir2, link1) |
| 142 link2 = os.path.join(dir1, 'link2') |
| 143 os.symlink(dir3, link2) |
| 144 link3 = os.path.join(dir1, 'link3') |
| 145 os.symlink(file1, link3) |
| 146 # make sure symlinks are removed but not followed |
| 147 shutil.rmtree(dir1) |
| 148 self.assertFalse(os.path.exists(dir1)) |
| 149 self.assertTrue(os.path.exists(dir3)) |
| 150 self.assertTrue(os.path.exists(file1)) |
| 119 | 151 |
| 120 def test_rmtree_errors(self): | 152 def test_rmtree_errors(self): |
| 121 # filename is guaranteed not to exist | 153 # filename is guaranteed not to exist |
| 122 filename = tempfile.mktemp() | 154 filename = tempfile.mktemp() |
| 123 self.assertRaises(OSError, shutil.rmtree, filename) | 155 self.assertRaises(OSError, shutil.rmtree, filename) |
| 124 | 156 |
| 125 # See bug #1071513 for why we don't run this on cygwin | 157 # See bug #1071513 for why we don't run this on cygwin |
| 126 # and bug #1076467 for why we don't run this as root. | 158 # and bug #1076467 for why we don't run this as root. |
| 127 if (hasattr(os, 'chmod') and sys.platform[:6] != 'cygwin' | 159 if (hasattr(os, 'chmod') and sys.platform[:6] != 'cygwin' |
| 128 and not (hasattr(os, 'geteuid') and os.geteuid() == 0)): | 160 and not (hasattr(os, 'geteuid') and os.geteuid() == 0)): |
| (...skipping 48 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 177 self.errorState += 1 | 209 self.errorState += 1 |
| 178 else: | 210 else: |
| 179 self.assertEqual(func, os.rmdir) | 211 self.assertEqual(func, os.rmdir) |
| 180 self.assertEqual(arg, TESTFN) | 212 self.assertEqual(arg, TESTFN) |
| 181 self.assertTrue(issubclass(exc[0], OSError)) | 213 self.assertTrue(issubclass(exc[0], OSError)) |
| 182 self.errorState = 3 | 214 self.errorState = 3 |
| 183 | 215 |
| 184 def test_rmtree_does_not_choke_on_failing_lstat(self): | 216 def test_rmtree_does_not_choke_on_failing_lstat(self): |
| 185 try: | 217 try: |
| 186 orig_lstat = os.lstat | 218 orig_lstat = os.lstat |
| 187 def raiser(fn): | 219 def raiser(fn, *args, **kwargs): |
| 188 if fn != TESTFN: | 220 if fn != TESTFN: |
| 189 raise OSError() | 221 raise OSError() |
| 190 else: | 222 else: |
| 191 return orig_lstat(fn) | 223 return orig_lstat(fn) |
| 192 os.lstat = raiser | 224 os.lstat = raiser |
| 193 | 225 |
| 194 os.mkdir(TESTFN) | 226 os.mkdir(TESTFN) |
| 195 write_file((TESTFN, 'foo'), 'foo') | 227 write_file((TESTFN, 'foo'), 'foo') |
| 196 shutil.rmtree(TESTFN) | 228 shutil.rmtree(TESTFN) |
| 197 finally: | 229 finally: |
| (...skipping 287 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 485 os.symlink(src, link) | 517 os.symlink(src, link) |
| 486 # don't follow | 518 # don't follow |
| 487 shutil.copyfile(link, dst_link, follow_symlinks=False) | 519 shutil.copyfile(link, dst_link, follow_symlinks=False) |
| 488 self.assertTrue(os.path.islink(dst_link)) | 520 self.assertTrue(os.path.islink(dst_link)) |
| 489 self.assertEqual(os.readlink(link), os.readlink(dst_link)) | 521 self.assertEqual(os.readlink(link), os.readlink(dst_link)) |
| 490 # follow | 522 # follow |
| 491 shutil.copyfile(link, dst) | 523 shutil.copyfile(link, dst) |
| 492 self.assertFalse(os.path.islink(dst)) | 524 self.assertFalse(os.path.islink(dst)) |
| 493 | 525 |
| 494 def test_rmtree_uses_safe_fd_version_if_available(self): | 526 def test_rmtree_uses_safe_fd_version_if_available(self): |
| 495 if os.unlink in os.supports_dir_fd and os.open in os.supports_dir_fd: | 527 _use_fd_functions = ({os.open, os.stat, os.unlink, os.rmdir} <= |
| 528 os.supports_dir_fd and |
| 529 os.listdir in os.supports_fd and |
| 530 os.stat in os.supports_follow_symlinks) |
| 531 if _use_fd_functions: |
| 496 self.assertTrue(shutil._use_fd_functions) | 532 self.assertTrue(shutil._use_fd_functions) |
| 497 self.assertTrue(shutil.rmtree.avoids_symlink_attacks) | 533 self.assertTrue(shutil.rmtree.avoids_symlink_attacks) |
| 498 tmp_dir = self.mkdtemp() | 534 tmp_dir = self.mkdtemp() |
| 499 d = os.path.join(tmp_dir, 'a') | 535 d = os.path.join(tmp_dir, 'a') |
| 500 os.mkdir(d) | 536 os.mkdir(d) |
| 501 try: | 537 try: |
| 502 real_rmtree = shutil._rmtree_safe_fd | 538 real_rmtree = shutil._rmtree_safe_fd |
| 503 class Called(Exception): pass | 539 class Called(Exception): pass |
| 504 def _raiser(*args, **kwargs): | 540 def _raiser(*args, **kwargs): |
| 505 raise Called | 541 raise Called |
| (...skipping 26 matching lines...) Expand all Loading... |
| 532 self.assertTrue(os.path.isdir(os.path.join(dst_dir, 'test_dir'))) | 568 self.assertTrue(os.path.isdir(os.path.join(dst_dir, 'test_dir'))) |
| 533 self.assertTrue(os.path.isfile(os.path.join(dst_dir, 'test_dir', | 569 self.assertTrue(os.path.isfile(os.path.join(dst_dir, 'test_dir', |
| 534 'test.txt'))) | 570 'test.txt'))) |
| 535 actual = read_file((dst_dir, 'test.txt')) | 571 actual = read_file((dst_dir, 'test.txt')) |
| 536 self.assertEqual(actual, '123') | 572 self.assertEqual(actual, '123') |
| 537 actual = read_file((dst_dir, 'test_dir', 'test.txt')) | 573 actual = read_file((dst_dir, 'test_dir', 'test.txt')) |
| 538 self.assertEqual(actual, '456') | 574 self.assertEqual(actual, '456') |
| 539 | 575 |
| 540 @support.skip_unless_symlink | 576 @support.skip_unless_symlink |
| 541 def test_copytree_symlinks(self): | 577 def test_copytree_symlinks(self): |
| 542 tmp_dir = self.mkdtemp() | |
| 543 src_dir = os.path.join(tmp_dir, 'src') | |
| 544 dst_dir = os.path.join(tmp_dir, 'dst') | |
| 545 sub_dir = os.path.join(src_dir, 'sub') | |
| 546 os.mkdir(src_dir) | |
| 547 os.mkdir(sub_dir) | |
| 548 write_file((src_dir, 'file.txt'), 'foo') | |
| 549 src_link = os.path.join(sub_dir, 'link') | |
| 550 dst_link = os.path.join(dst_dir, 'sub/link') | |
| 551 os.symlink(os.path.join(src_dir, 'file.txt'), | |
| 552 src_link) | |
| 553 if hasattr(os, 'lchmod'): | |
| 554 os.lchmod(src_link, stat.S_IRWXU | stat.S_IRWXO) | |
| 555 if hasattr(os, 'lchflags') and hasattr(stat, 'UF_NODUMP'): | |
| 556 os.lchflags(src_link, stat.UF_NODUMP) | |
| 557 src_stat = os.lstat(src_link) | |
| 558 shutil.copytree(src_dir, dst_dir, follow_symlinks=False) | |
| 559 self.assertTrue(os.path.islink(os.path.join(dst_dir, 'sub', 'link'))) | |
| 560 self.assertEqual(os.readlink(os.path.join(dst_dir, 'sub', 'link')), | |
| 561 os.path.join(src_dir, 'file.txt')) | |
| 562 dst_stat = os.lstat(dst_link) | |
| 563 if hasattr(os, 'lchmod'): | |
| 564 self.assertEqual(dst_stat.st_mode, src_stat.st_mode) | |
| 565 if hasattr(os, 'lchflags'): | |
| 566 self.assertEqual(dst_stat.st_flags, src_stat.st_flags) | |
| 567 | |
| 568 def test_copytree_symlinks_old(self): | |
| 569 tmp_dir = self.mkdtemp() | 578 tmp_dir = self.mkdtemp() |
| 570 src_dir = os.path.join(tmp_dir, 'src') | 579 src_dir = os.path.join(tmp_dir, 'src') |
| 571 dst_dir = os.path.join(tmp_dir, 'dst') | 580 dst_dir = os.path.join(tmp_dir, 'dst') |
| 572 sub_dir = os.path.join(src_dir, 'sub') | 581 sub_dir = os.path.join(src_dir, 'sub') |
| 573 os.mkdir(src_dir) | 582 os.mkdir(src_dir) |
| 574 os.mkdir(sub_dir) | 583 os.mkdir(sub_dir) |
| 575 write_file((src_dir, 'file.txt'), 'foo') | 584 write_file((src_dir, 'file.txt'), 'foo') |
| 576 src_link = os.path.join(sub_dir, 'link') | 585 src_link = os.path.join(sub_dir, 'link') |
| 577 dst_link = os.path.join(dst_dir, 'sub/link') | 586 dst_link = os.path.join(dst_dir, 'sub/link') |
| 578 os.symlink(os.path.join(src_dir, 'file.txt'), | 587 os.symlink(os.path.join(src_dir, 'file.txt'), |
| 579 src_link) | 588 src_link) |
| 580 if hasattr(os, 'lchmod'): | 589 if hasattr(os, 'lchmod'): |
| 581 os.lchmod(src_link, stat.S_IRWXU | stat.S_IRWXO) | 590 os.lchmod(src_link, stat.S_IRWXU | stat.S_IRWXO) |
| 582 if hasattr(os, 'lchflags') and hasattr(stat, 'UF_NODUMP'): | 591 if hasattr(os, 'lchflags') and hasattr(stat, 'UF_NODUMP'): |
| 583 os.lchflags(src_link, stat.UF_NODUMP) | 592 os.lchflags(src_link, stat.UF_NODUMP) |
| 584 src_stat = os.lstat(src_link) | 593 src_stat = os.lstat(src_link) |
| 585 shutil.copytree(src_dir, dst_dir, symlinks=True) | 594 shutil.copytree(src_dir, dst_dir, symlinks=True) |
| 586 self.assertTrue(os.path.islink(os.path.join(dst_dir, 'sub', 'link'))) | 595 self.assertTrue(os.path.islink(os.path.join(dst_dir, 'sub', 'link'))) |
| 587 self.assertEqual(os.readlink(os.path.join(dst_dir, 'sub', 'link')), | 596 self.assertEqual(os.readlink(os.path.join(dst_dir, 'sub', 'link')), |
| 588 os.path.join(src_dir, 'file.txt')) | 597 os.path.join(src_dir, 'file.txt')) |
| 589 dst_stat = os.lstat(dst_link) | 598 dst_stat = os.lstat(dst_link) |
| 590 if hasattr(os, 'lchmod'): | 599 if hasattr(os, 'lchmod'): |
| 591 self.assertEqual(dst_stat.st_mode, src_stat.st_mode) | 600 self.assertEqual(dst_stat.st_mode, src_stat.st_mode) |
| 592 if hasattr(os, 'lchflags'): | 601 if hasattr(os, 'lchflags'): |
| 593 self.assertEqual(dst_stat.st_flags, src_stat.st_flags) | 602 self.assertEqual(dst_stat.st_flags, src_stat.st_flags) |
| 594 | 603 |
| 595 def test_copytree_incompatible_args(self): | |
| 596 tmp_dir = self.mkdtemp() | |
| 597 src_dir = os.path.join(tmp_dir, 'src') | |
| 598 dst_dir = os.path.join(tmp_dir, 'dst') | |
| 599 sub_dir = os.path.join(src_dir, 'sub') | |
| 600 os.mkdir(src_dir) | |
| 601 os.mkdir(sub_dir) | |
| 602 self.assertRaises(TypeError, shutil.copytree, src_dir, dst_dir, symlinks
=True, follow_symlinks=False) | |
| 603 | |
| 604 def test_copytree_with_exclude(self): | 604 def test_copytree_with_exclude(self): |
| 605 # creating data | 605 # creating data |
| 606 join = os.path.join | 606 join = os.path.join |
| 607 exists = os.path.exists | 607 exists = os.path.exists |
| 608 src_dir = tempfile.mkdtemp() | 608 src_dir = tempfile.mkdtemp() |
| 609 try: | 609 try: |
| 610 dst_dir = join(tempfile.mkdtemp(), 'destination') | 610 dst_dir = join(tempfile.mkdtemp(), 'destination') |
| 611 write_file((src_dir, 'test.txt'), '123') | 611 write_file((src_dir, 'test.txt'), '123') |
| 612 write_file((src_dir, 'test.tmp'), '123') | 612 write_file((src_dir, 'test.tmp'), '123') |
| 613 os.mkdir(join(src_dir, 'test_dir')) | 613 os.mkdir(join(src_dir, 'test_dir')) |
| (...skipping 163 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 777 os.symlink('IDONTEXIST', os.path.join(src_dir, 'test.txt')) | 777 os.symlink('IDONTEXIST', os.path.join(src_dir, 'test.txt')) |
| 778 os.mkdir(os.path.join(src_dir, 'test_dir')) | 778 os.mkdir(os.path.join(src_dir, 'test_dir')) |
| 779 write_file((src_dir, 'test_dir', 'test.txt'), '456') | 779 write_file((src_dir, 'test_dir', 'test.txt'), '456') |
| 780 self.assertRaises(Error, shutil.copytree, src_dir, dst_dir) | 780 self.assertRaises(Error, shutil.copytree, src_dir, dst_dir) |
| 781 | 781 |
| 782 # a dangling symlink is ignored with the proper flag | 782 # a dangling symlink is ignored with the proper flag |
| 783 dst_dir = os.path.join(self.mkdtemp(), 'destination2') | 783 dst_dir = os.path.join(self.mkdtemp(), 'destination2') |
| 784 shutil.copytree(src_dir, dst_dir, ignore_dangling_symlinks=True) | 784 shutil.copytree(src_dir, dst_dir, ignore_dangling_symlinks=True) |
| 785 self.assertNotIn('test.txt', os.listdir(dst_dir)) | 785 self.assertNotIn('test.txt', os.listdir(dst_dir)) |
| 786 | 786 |
| 787 # a dangling symlink is copied if follow_symlinks=False | 787 # a dangling symlink is copied if symlinks=True |
| 788 dst_dir = os.path.join(self.mkdtemp(), 'destination3') | 788 dst_dir = os.path.join(self.mkdtemp(), 'destination3') |
| 789 shutil.copytree(src_dir, dst_dir, follow_symlinks=False) | 789 shutil.copytree(src_dir, dst_dir, symlinks=True) |
| 790 self.assertIn('test.txt', os.listdir(dst_dir)) | 790 self.assertIn('test.txt', os.listdir(dst_dir)) |
| 791 | 791 |
| 792 def _copy_file(self, method): | 792 def _copy_file(self, method): |
| 793 fname = 'test.txt' | 793 fname = 'test.txt' |
| 794 tmpdir = self.mkdtemp() | 794 tmpdir = self.mkdtemp() |
| 795 write_file((tmpdir, fname), 'xxx') | 795 write_file((tmpdir, fname), 'xxx') |
| 796 file1 = os.path.join(tmpdir, fname) | 796 file1 = os.path.join(tmpdir, fname) |
| 797 tmpdir2 = self.mkdtemp() | 797 tmpdir2 = self.mkdtemp() |
| 798 method(file1, tmpdir2) | 798 method(file1, tmpdir2) |
| 799 file2 = os.path.join(tmpdir2, fname) | 799 file2 = os.path.join(tmpdir2, fname) |
| (...skipping 801 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 1601 | 1601 |
| 1602 self.assertEqual(expected, actual) | 1602 self.assertEqual(expected, actual) |
| 1603 | 1603 |
| 1604 | 1604 |
| 1605 def test_main(): | 1605 def test_main(): |
| 1606 support.run_unittest(TestShutil, TestMove, TestCopyFile, | 1606 support.run_unittest(TestShutil, TestMove, TestCopyFile, |
| 1607 TermsizeTests, TestWhich) | 1607 TermsizeTests, TestWhich) |
| 1608 | 1608 |
| 1609 if __name__ == '__main__': | 1609 if __name__ == '__main__': |
| 1610 test_main() | 1610 test_main() |
| LEFT | RIGHT |