codekingpro/portable-devtools

1from hashlib import md52from itertools import product3 4import pytest5 6from fsspec.tests.abstract.common import GLOB_EDGE_CASES_TESTS7 8 9class AbstractCopyTests:10    def test_copy_file_to_existing_directory(11        self,12        fs,13        fs_join,14        fs_bulk_operations_scenario_0,15        fs_target,16        supports_empty_directories,17    ):18        # Copy scenario 1a19        source = fs_bulk_operations_scenario_020 21        target = fs_target22        fs.mkdir(target)23        if not supports_empty_directories:24            # Force target directory to exist by adding a dummy file25            fs.touch(fs_join(target, "dummy"))26        assert fs.isdir(target)27 28        target_file2 = fs_join(target, "file2")29        target_subfile1 = fs_join(target, "subfile1")30 31        # Copy from source directory32        fs.cp(fs_join(source, "file2"), target)33        assert fs.isfile(target_file2)34 35        # Copy from sub directory36        fs.cp(fs_join(source, "subdir", "subfile1"), target)37        assert fs.isfile(target_subfile1)38 39        # Remove copied files40        fs.rm([target_file2, target_subfile1])41        assert not fs.exists(target_file2)42        assert not fs.exists(target_subfile1)43 44        # Repeat with trailing slash on target45        fs.cp(fs_join(source, "file2"), target + "/")46        assert fs.isdir(target)47        assert fs.isfile(target_file2)48 49        fs.cp(fs_join(source, "subdir", "subfile1"), target + "/")50        assert fs.isfile(target_subfile1)51 52    def test_copy_file_to_new_directory(53        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target54    ):55        # Copy scenario 1b56        source = fs_bulk_operations_scenario_057 58        target = fs_target59        fs.mkdir(target)60 61        fs.cp(62            fs_join(source, "subdir", "subfile1"), fs_join(target, "newdir/")63        )  # Note trailing slash64        assert fs.isdir(target)65        assert fs.isdir(fs_join(target, "newdir"))66        assert fs.isfile(fs_join(target, "newdir", "subfile1"))67 68    def test_copy_file_to_file_in_existing_directory(69        self,70        fs,71        fs_join,72        fs_bulk_operations_scenario_0,73        fs_target,74        supports_empty_directories,75    ):76        # Copy scenario 1c77        source = fs_bulk_operations_scenario_078 79        target = fs_target80        fs.mkdir(target)81        if not supports_empty_directories:82            # Force target directory to exist by adding a dummy file83            fs.touch(fs_join(target, "dummy"))84        assert fs.isdir(target)85 86        fs.cp(fs_join(source, "subdir", "subfile1"), fs_join(target, "newfile"))87        assert fs.isfile(fs_join(target, "newfile"))88 89    def test_copy_file_to_file_in_new_directory(90        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target91    ):92        # Copy scenario 1d93        source = fs_bulk_operations_scenario_094 95        target = fs_target96        fs.mkdir(target)97 98        fs.cp(99            fs_join(source, "subdir", "subfile1"), fs_join(target, "newdir", "newfile")100        )101        assert fs.isdir(fs_join(target, "newdir"))102        assert fs.isfile(fs_join(target, "newdir", "newfile"))103 104    def test_copy_directory_to_existing_directory(105        self,106        fs,107        fs_join,108        fs_bulk_operations_scenario_0,109        fs_target,110        supports_empty_directories,111    ):112        # Copy scenario 1e113        source = fs_bulk_operations_scenario_0114 115        target = fs_target116        fs.mkdir(target)117        if not supports_empty_directories:118            # Force target directory to exist by adding a dummy file119            dummy = fs_join(target, "dummy")120            fs.touch(dummy)121        assert fs.isdir(target)122 123        for source_slash, target_slash in zip([False, True], [False, True]):124            s = fs_join(source, "subdir")125            if source_slash:126                s += "/"127            t = target + "/" if target_slash else target128 129            # Without recursive does nothing130            fs.cp(s, t)131            assert fs.ls(target, detail=False) == (132                [] if supports_empty_directories else [dummy]133            )134 135            # With recursive136            fs.cp(s, t, recursive=True)137            if source_slash:138                assert fs.isfile(fs_join(target, "subfile1"))139                assert fs.isfile(fs_join(target, "subfile2"))140                assert fs.isdir(fs_join(target, "nesteddir"))141                assert fs.isfile(fs_join(target, "nesteddir", "nestedfile"))142                assert not fs.exists(fs_join(target, "subdir"))143 144                fs.rm(145                    [146                        fs_join(target, "subfile1"),147                        fs_join(target, "subfile2"),148                        fs_join(target, "nesteddir"),149                    ],150                    recursive=True,151                )152            else:153                assert fs.isdir(fs_join(target, "subdir"))154                assert fs.isfile(fs_join(target, "subdir", "subfile1"))155                assert fs.isfile(fs_join(target, "subdir", "subfile2"))156                assert fs.isdir(fs_join(target, "subdir", "nesteddir"))157                assert fs.isfile(fs_join(target, "subdir", "nesteddir", "nestedfile"))158 159                fs.rm(fs_join(target, "subdir"), recursive=True)160            assert fs.ls(target, detail=False) == (161                [] if supports_empty_directories else [dummy]162            )163 164            # Limit recursive by maxdepth165            fs.cp(s, t, recursive=True, maxdepth=1)166            if source_slash:167                assert fs.isfile(fs_join(target, "subfile1"))168                assert fs.isfile(fs_join(target, "subfile2"))169                assert not fs.exists(fs_join(target, "nesteddir"))170                assert not fs.exists(fs_join(target, "subdir"))171 172                fs.rm(173                    [174                        fs_join(target, "subfile1"),175                        fs_join(target, "subfile2"),176                    ],177                    recursive=True,178                )179            else:180                assert fs.isdir(fs_join(target, "subdir"))181                assert fs.isfile(fs_join(target, "subdir", "subfile1"))182                assert fs.isfile(fs_join(target, "subdir", "subfile2"))183                assert not fs.exists(fs_join(target, "subdir", "nesteddir"))184 185                fs.rm(fs_join(target, "subdir"), recursive=True)186            assert fs.ls(target, detail=False) == (187                [] if supports_empty_directories else [dummy]188            )189 190    def test_copy_directory_to_new_directory(191        self,192        fs,193        fs_join,194        fs_bulk_operations_scenario_0,195        fs_target,196        supports_empty_directories,197    ):198        # Copy scenario 1f199        source = fs_bulk_operations_scenario_0200 201        target = fs_target202        fs.mkdir(target)203 204        for source_slash, target_slash in zip([False, True], [False, True]):205            s = fs_join(source, "subdir")206            if source_slash:207                s += "/"208            t = fs_join(target, "newdir")209            if target_slash:210                t += "/"211 212            # Without recursive does nothing213            fs.cp(s, t)214            if supports_empty_directories:215                assert fs.ls(target) == []216            else:217                with pytest.raises(FileNotFoundError):218                    fs.ls(target)219 220            # With recursive221            fs.cp(s, t, recursive=True)222            assert fs.isdir(fs_join(target, "newdir"))223            assert fs.isfile(fs_join(target, "newdir", "subfile1"))224            assert fs.isfile(fs_join(target, "newdir", "subfile2"))225            assert fs.isdir(fs_join(target, "newdir", "nesteddir"))226            assert fs.isfile(fs_join(target, "newdir", "nesteddir", "nestedfile"))227            assert not fs.exists(fs_join(target, "subdir"))228 229            fs.rm(fs_join(target, "newdir"), recursive=True)230            assert not fs.exists(fs_join(target, "newdir"))231 232            # Limit recursive by maxdepth233            fs.cp(s, t, recursive=True, maxdepth=1)234            assert fs.isdir(fs_join(target, "newdir"))235            assert fs.isfile(fs_join(target, "newdir", "subfile1"))236            assert fs.isfile(fs_join(target, "newdir", "subfile2"))237            assert not fs.exists(fs_join(target, "newdir", "nesteddir"))238            assert not fs.exists(fs_join(target, "subdir"))239 240            fs.rm(fs_join(target, "newdir"), recursive=True)241            assert not fs.exists(fs_join(target, "newdir"))242 243    def test_copy_glob_to_existing_directory(244        self,245        fs,246        fs_join,247        fs_bulk_operations_scenario_0,248        fs_target,249        supports_empty_directories,250    ):251        # Copy scenario 1g252        source = fs_bulk_operations_scenario_0253 254        target = fs_target255        fs.mkdir(target)256        if not supports_empty_directories:257            # Force target directory to exist by adding a dummy file258            dummy = fs_join(target, "dummy")259            fs.touch(dummy)260        assert fs.isdir(target)261 262        for target_slash in [False, True]:263            t = target + "/" if target_slash else target264 265            # Without recursive266            fs.cp(fs_join(source, "subdir", "*"), t)267            assert fs.isfile(fs_join(target, "subfile1"))268            assert fs.isfile(fs_join(target, "subfile2"))269            assert not fs.isdir(fs_join(target, "nesteddir"))270            assert not fs.exists(fs_join(target, "nesteddir", "nestedfile"))271            assert not fs.exists(fs_join(target, "subdir"))272 273            fs.rm(274                [275                    fs_join(target, "subfile1"),276                    fs_join(target, "subfile2"),277                ],278                recursive=True,279            )280            assert fs.ls(target, detail=False) == (281                [] if supports_empty_directories else [dummy]282            )283 284            # With recursive285            for glob, recursive in zip(["*", "**"], [True, False]):286                fs.cp(fs_join(source, "subdir", glob), t, recursive=recursive)287                assert fs.isfile(fs_join(target, "subfile1"))288                assert fs.isfile(fs_join(target, "subfile2"))289                assert fs.isdir(fs_join(target, "nesteddir"))290                assert fs.isfile(fs_join(target, "nesteddir", "nestedfile"))291                assert not fs.exists(fs_join(target, "subdir"))292 293                fs.rm(294                    [295                        fs_join(target, "subfile1"),296                        fs_join(target, "subfile2"),297                        fs_join(target, "nesteddir"),298                    ],299                    recursive=True,300                )301                assert fs.ls(target, detail=False) == (302                    [] if supports_empty_directories else [dummy]303                )304 305                # Limit recursive by maxdepth306                fs.cp(307                    fs_join(source, "subdir", glob), t, recursive=recursive, maxdepth=1308                )309                assert fs.isfile(fs_join(target, "subfile1"))310                assert fs.isfile(fs_join(target, "subfile2"))311                assert not fs.exists(fs_join(target, "nesteddir"))312                assert not fs.exists(fs_join(target, "subdir"))313 314                fs.rm(315                    [316                        fs_join(target, "subfile1"),317                        fs_join(target, "subfile2"),318                    ],319                    recursive=True,320                )321                assert fs.ls(target, detail=False) == (322                    [] if supports_empty_directories else [dummy]323                )324 325    def test_copy_glob_to_new_directory(326        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target327    ):328        # Copy scenario 1h329        source = fs_bulk_operations_scenario_0330 331        target = fs_target332        fs.mkdir(target)333 334        for target_slash in [False, True]:335            t = fs_join(target, "newdir")336            if target_slash:337                t += "/"338 339            # Without recursive340            fs.cp(fs_join(source, "subdir", "*"), t)341            assert fs.isdir(fs_join(target, "newdir"))342            assert fs.isfile(fs_join(target, "newdir", "subfile1"))343            assert fs.isfile(fs_join(target, "newdir", "subfile2"))344            assert not fs.exists(fs_join(target, "newdir", "nesteddir"))345            assert not fs.exists(fs_join(target, "newdir", "nesteddir", "nestedfile"))346            assert not fs.exists(fs_join(target, "subdir"))347            assert not fs.exists(fs_join(target, "newdir", "subdir"))348 349            fs.rm(fs_join(target, "newdir"), recursive=True)350            assert not fs.exists(fs_join(target, "newdir"))351 352            # With recursive353            for glob, recursive in zip(["*", "**"], [True, False]):354                fs.cp(fs_join(source, "subdir", glob), t, recursive=recursive)355                assert fs.isdir(fs_join(target, "newdir"))356                assert fs.isfile(fs_join(target, "newdir", "subfile1"))357                assert fs.isfile(fs_join(target, "newdir", "subfile2"))358                assert fs.isdir(fs_join(target, "newdir", "nesteddir"))359                assert fs.isfile(fs_join(target, "newdir", "nesteddir", "nestedfile"))360                assert not fs.exists(fs_join(target, "subdir"))361                assert not fs.exists(fs_join(target, "newdir", "subdir"))362 363                fs.rm(fs_join(target, "newdir"), recursive=True)364                assert not fs.exists(fs_join(target, "newdir"))365 366                # Limit recursive by maxdepth367                fs.cp(368                    fs_join(source, "subdir", glob), t, recursive=recursive, maxdepth=1369                )370                assert fs.isdir(fs_join(target, "newdir"))371                assert fs.isfile(fs_join(target, "newdir", "subfile1"))372                assert fs.isfile(fs_join(target, "newdir", "subfile2"))373                assert not fs.exists(fs_join(target, "newdir", "nesteddir"))374                assert not fs.exists(fs_join(target, "subdir"))375                assert not fs.exists(fs_join(target, "newdir", "subdir"))376 377                fs.rm(fs_join(target, "newdir"), recursive=True)378                assert not fs.exists(fs_join(target, "newdir"))379 380    @pytest.mark.parametrize(381        GLOB_EDGE_CASES_TESTS["argnames"],382        GLOB_EDGE_CASES_TESTS["argvalues"],383    )384    def test_copy_glob_edge_cases(385        self,386        path,387        recursive,388        maxdepth,389        expected,390        fs,391        fs_join,392        fs_glob_edge_cases_files,393        fs_target,394        fs_sanitize_path,395    ):396        # Copy scenario 1g397        source = fs_glob_edge_cases_files398 399        target = fs_target400 401        for new_dir, target_slash in product([True, False], [True, False]):402            fs.mkdir(target)403 404            t = fs_join(target, "newdir") if new_dir else target405            t = t + "/" if target_slash else t406 407            fs.copy(fs_join(source, path), t, recursive=recursive, maxdepth=maxdepth)408 409            output = fs.find(target)410            if new_dir:411                prefixed_expected = [412                    fs_sanitize_path(fs_join(target, "newdir", p)) for p in expected413                ]414            else:415                prefixed_expected = [416                    fs_sanitize_path(fs_join(target, p)) for p in expected417                ]418            assert sorted(output) == sorted(prefixed_expected)419 420            try:421                fs.rm(target, recursive=True)422            except FileNotFoundError:423                pass424 425    def test_copy_list_of_files_to_existing_directory(426        self,427        fs,428        fs_join,429        fs_bulk_operations_scenario_0,430        fs_target,431        supports_empty_directories,432    ):433        # Copy scenario 2a434        source = fs_bulk_operations_scenario_0435 436        target = fs_target437        fs.mkdir(target)438        if not supports_empty_directories:439            # Force target directory to exist by adding a dummy file440            dummy = fs_join(target, "dummy")441            fs.touch(dummy)442        assert fs.isdir(target)443 444        source_files = [445            fs_join(source, "file1"),446            fs_join(source, "file2"),447            fs_join(source, "subdir", "subfile1"),448        ]449 450        for target_slash in [False, True]:451            t = target + "/" if target_slash else target452 453            fs.cp(source_files, t)454            assert fs.isfile(fs_join(target, "file1"))455            assert fs.isfile(fs_join(target, "file2"))456            assert fs.isfile(fs_join(target, "subfile1"))457 458            fs.rm(459                [460                    fs_join(target, "file1"),461                    fs_join(target, "file2"),462                    fs_join(target, "subfile1"),463                ],464                recursive=True,465            )466            assert fs.ls(target, detail=False) == (467                [] if supports_empty_directories else [dummy]468            )469 470    def test_copy_list_of_files_to_new_directory(471        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target472    ):473        # Copy scenario 2b474        source = fs_bulk_operations_scenario_0475 476        target = fs_target477        fs.mkdir(target)478 479        source_files = [480            fs_join(source, "file1"),481            fs_join(source, "file2"),482            fs_join(source, "subdir", "subfile1"),483        ]484 485        fs.cp(source_files, fs_join(target, "newdir") + "/")  # Note trailing slash486        assert fs.isdir(fs_join(target, "newdir"))487        assert fs.isfile(fs_join(target, "newdir", "file1"))488        assert fs.isfile(fs_join(target, "newdir", "file2"))489        assert fs.isfile(fs_join(target, "newdir", "subfile1"))490 491    def test_copy_two_files_new_directory(492        self, fs, fs_join, fs_bulk_operations_scenario_0, fs_target493    ):494        # This is a duplicate of test_copy_list_of_files_to_new_directory and495        # can eventually be removed.496        source = fs_bulk_operations_scenario_0497 498        target = fs_target499        assert not fs.exists(target)500        fs.cp([fs_join(source, "file1"), fs_join(source, "file2")], target)501 502        assert fs.isdir(target)503        assert fs.isfile(fs_join(target, "file1"))504        assert fs.isfile(fs_join(target, "file2"))505 506    def test_copy_directory_without_files_with_same_name_prefix(507        self,508        fs,509        fs_join,510        fs_target,511        fs_dir_and_file_with_same_name_prefix,512        supports_empty_directories,513    ):514        # Create the test dirs515        source = fs_dir_and_file_with_same_name_prefix516        target = fs_target517 518        # Test without glob519        fs.cp(fs_join(source, "subdir"), target, recursive=True)520 521        assert fs.isfile(fs_join(target, "subfile.txt"))522        assert not fs.isfile(fs_join(target, "subdir.txt"))523 524        fs.rm([fs_join(target, "subfile.txt")])525        if supports_empty_directories:526            assert fs.ls(target) == []527        else:528            assert not fs.exists(target)529 530        # Test with glob531        fs.cp(fs_join(source, "subdir*"), target, recursive=True)532 533        assert fs.isdir(fs_join(target, "subdir"))534        assert fs.isfile(fs_join(target, "subdir", "subfile.txt"))535        assert fs.isfile(fs_join(target, "subdir.txt"))536 537    def test_copy_with_source_and_destination_as_list(538        self, fs, fs_target, fs_join, fs_10_files_with_hashed_names539    ):540        # Create the test dir541        source = fs_10_files_with_hashed_names542        target = fs_target543 544        # Create list of files for source and destination545        source_files = []546        destination_files = []547        for i in range(10):548            hashed_i = md5(str(i).encode("utf-8")).hexdigest()549            source_files.append(fs_join(source, f"{hashed_i}.txt"))550            destination_files.append(fs_join(target, f"{hashed_i}.txt"))551 552        # Copy and assert order was kept553        fs.copy(path1=source_files, path2=destination_files)554 555        for i in range(10):556            file_content = fs.cat(destination_files[i]).decode("utf-8")557            assert file_content == str(i)558