Verifies file_io directory handling methods.
()
| 52 | |
| 53 | |
| 54 | def create_dir_test(): |
| 55 | """Verifies file_io directory handling methods.""" |
| 56 | |
| 57 | # Test directory creation. |
| 58 | starttime_ms = int(round(time.time() * 1000)) |
| 59 | dir_name = "%s/tf_gcs_test_%s" % (FLAGS.gcs_bucket_url, starttime_ms) |
| 60 | print("Creating dir %s" % dir_name) |
| 61 | file_io.create_dir(dir_name) |
| 62 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 63 | print("Created directory in: %d milliseconds" % elapsed_ms) |
| 64 | |
| 65 | # Check that the directory exists. |
| 66 | dir_exists = file_io.is_directory(dir_name) |
| 67 | assert dir_exists |
| 68 | print("%s directory exists: %s" % (dir_name, dir_exists)) |
| 69 | |
| 70 | # Test recursive directory creation. |
| 71 | starttime_ms = int(round(time.time() * 1000)) |
| 72 | recursive_dir_name = "%s/%s/%s" % (dir_name, |
| 73 | "nested_dir1", |
| 74 | "nested_dir2") |
| 75 | print("Creating recursive dir %s" % recursive_dir_name) |
| 76 | file_io.recursive_create_dir(recursive_dir_name) |
| 77 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 78 | print("Created directory recursively in: %d milliseconds" % elapsed_ms) |
| 79 | |
| 80 | # Check that the directory exists. |
| 81 | recursive_dir_exists = file_io.is_directory(recursive_dir_name) |
| 82 | assert recursive_dir_exists |
| 83 | print("%s directory exists: %s" % (recursive_dir_name, recursive_dir_exists)) |
| 84 | |
| 85 | # Create some contents in the just created directory and list the contents. |
| 86 | num_files = 10 |
| 87 | files_to_create = ["file_%d.txt" % n for n in range(num_files)] |
| 88 | for file_num in files_to_create: |
| 89 | file_name = "%s/%s" % (dir_name, file_num) |
| 90 | print("Creating file %s." % file_name) |
| 91 | file_io.write_string_to_file(file_name, "test file.") |
| 92 | |
| 93 | print("Listing directory %s." % dir_name) |
| 94 | starttime_ms = int(round(time.time() * 1000)) |
| 95 | directory_contents = file_io.list_directory(dir_name) |
| 96 | print(directory_contents) |
| 97 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
| 98 | print("Listed directory %s in %s milliseconds" % (dir_name, elapsed_ms)) |
| 99 | assert set(directory_contents) == set(files_to_create + ["nested_dir1/"]) |
| 100 | |
| 101 | # Test directory renaming. |
| 102 | dir_to_rename = "%s/old_dir" % dir_name |
| 103 | new_dir_name = "%s/new_dir" % dir_name |
| 104 | file_io.create_dir(dir_to_rename) |
| 105 | assert file_io.is_directory(dir_to_rename) |
| 106 | assert not file_io.is_directory(new_dir_name) |
| 107 | |
| 108 | starttime_ms = int(round(time.time() * 1000)) |
| 109 | print("Will try renaming directory %s to %s" % (dir_to_rename, new_dir_name)) |
| 110 | file_io.rename(dir_to_rename, new_dir_name) |
| 111 | elapsed_ms = int(round(time.time() * 1000)) - starttime_ms |
no test coverage detected