(fs, pathfn)
| 777 | |
| 778 | |
| 779 | def test_get_file_info_with_selector(fs, pathfn): |
| 780 | base_dir = pathfn('selector-dir/') |
| 781 | file_a = pathfn('selector-dir/test_file_a') |
| 782 | file_b = pathfn('selector-dir/test_file_b') |
| 783 | dir_a = pathfn('selector-dir/test_dir_a') |
| 784 | file_c = pathfn('selector-dir/test_dir_a/test_file_c') |
| 785 | dir_b = pathfn('selector-dir/test_dir_b') |
| 786 | |
| 787 | try: |
| 788 | fs.create_dir(base_dir) |
| 789 | with fs.open_output_stream(file_a): |
| 790 | pass |
| 791 | with fs.open_output_stream(file_b): |
| 792 | pass |
| 793 | fs.create_dir(dir_a) |
| 794 | with fs.open_output_stream(file_c): |
| 795 | pass |
| 796 | fs.create_dir(dir_b) |
| 797 | |
| 798 | # recursive selector |
| 799 | selector = FileSelector(base_dir, allow_not_found=False, |
| 800 | recursive=True) |
| 801 | assert selector.base_dir == base_dir |
| 802 | |
| 803 | infos = fs.get_file_info(selector) |
| 804 | if fs.type_name == "py::fsspec+('s3', 's3a')": |
| 805 | # s3fs only lists directories if they are not empty |
| 806 | len(infos) == 4 |
| 807 | else: |
| 808 | assert len(infos) == 5 |
| 809 | |
| 810 | for info in infos: |
| 811 | if (info.path.endswith(file_a) or info.path.endswith(file_b) or |
| 812 | info.path.endswith(file_c)): |
| 813 | assert info.type == FileType.File |
| 814 | elif (info.path.rstrip("/").endswith(dir_a) or |
| 815 | info.path.rstrip("/").endswith(dir_b)): |
| 816 | assert info.type == FileType.Directory |
| 817 | else: |
| 818 | raise ValueError(f'unexpected path {info.path}') |
| 819 | check_mtime_or_absent(info) |
| 820 | |
| 821 | # non-recursive selector -> not selecting the nested file_c |
| 822 | selector = FileSelector(base_dir, recursive=False) |
| 823 | |
| 824 | infos = fs.get_file_info(selector) |
| 825 | if fs.type_name == "py::fsspec+('s3', 's3a')": |
| 826 | # s3fs only lists directories if they are not empty |
| 827 | assert len(infos) == 3 |
| 828 | else: |
| 829 | assert len(infos) == 4 |
| 830 | |
| 831 | finally: |
| 832 | fs.delete_dir(base_dir) |
| 833 | |
| 834 | |
| 835 | def test_create_dir(fs, pathfn): |
nothing calls this directly
no test coverage detected