Verifies file_io's object manipulation methods .
()
| 126 | |
| 127 | |
| 128 | def create_object_test(): |
| 129 | """Verifies file_io's object manipulation methods .""" |
| 130 | starttime_ms = int(round(time.time() * 1000)) |
| 131 | dir_name = "%s/tf_gcs_test_%s" % (FLAGS.gcs_bucket_url, starttime_ms) |
| 132 | print("Creating dir %s." % dir_name) |
| 133 | file_io.create_dir(dir_name) |
| 134 | |
| 135 | num_files = 5 |
| 136 | # Create files of 2 different patterns in this directory. |
| 137 | files_pattern_1 = ["%s/test_file_%d.txt" % (dir_name, n) |
| 138 | for n in range(num_files)] |
| 139 | files_pattern_2 = ["%s/testfile%d.txt" % (dir_name, n) |
| 140 | for n in range(num_files)] |
| 141 | |
| 142 | starttime_ms = int(round(time.time() * 1000)) |
| 143 | files_to_create = files_pattern_1 + files_pattern_2 |
| 144 | for file_name in files_to_create: |
| 145 | print("Creating file %s." % file_name) |
| 146 | file_io.write_string_to_file(file_name, "test file creation.") |
| 147 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 148 | print("Created %d files in %s milliseconds" % ( |
| 149 | len(files_to_create), elapsed_ms)) |
| 150 | |
| 151 | # Listing files of pattern1. |
| 152 | list_files_pattern = "%s/test_file*.txt" % dir_name |
| 153 | print("Getting files matching pattern %s." % list_files_pattern) |
| 154 | starttime_ms = int(round(time.time() * 1000)) |
| 155 | files_list = file_io.get_matching_files(list_files_pattern) |
| 156 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 157 | print("Listed files in %s milliseconds" % elapsed_ms) |
| 158 | print(files_list) |
| 159 | assert set(files_list) == set(files_pattern_1) |
| 160 | |
| 161 | # Listing files of pattern2. |
| 162 | list_files_pattern = "%s/testfile*.txt" % dir_name |
| 163 | print("Getting files matching pattern %s." % list_files_pattern) |
| 164 | starttime_ms = int(round(time.time() * 1000)) |
| 165 | files_list = file_io.get_matching_files(list_files_pattern) |
| 166 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 167 | print("Listed files in %s milliseconds" % elapsed_ms) |
| 168 | print(files_list) |
| 169 | assert set(files_list) == set(files_pattern_2) |
| 170 | |
| 171 | # Test renaming file. |
| 172 | file_to_rename = "%s/oldname.txt" % dir_name |
| 173 | file_new_name = "%s/newname.txt" % dir_name |
| 174 | file_io.write_string_to_file(file_to_rename, "test file.") |
| 175 | assert file_io.file_exists(file_to_rename) |
| 176 | assert not file_io.file_exists(file_new_name) |
| 177 | |
| 178 | print("Will try renaming file %s to %s" % (file_to_rename, file_new_name)) |
| 179 | starttime_ms = int(round(time.time() * 1000)) |
| 180 | file_io.rename(file_to_rename, file_new_name) |
| 181 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 182 | print("File %s renamed to %s in %s milliseconds" % ( |
| 183 | file_to_rename, file_new_name, elapsed_ms)) |
| 184 | assert not file_io.file_exists(file_to_rename) |
| 185 | assert file_io.file_exists(file_new_name) |
no test coverage detected