(use_root, use_labels, shuffle)
| 63 | |
| 64 | |
| 65 | def _test_reader_files_arg(use_root, use_labels, shuffle): |
| 66 | root = g_root |
| 67 | fnames = g_files |
| 68 | if not use_root: |
| 69 | fnames = [os.path.join(root, f) for f in fnames] |
| 70 | root = None |
| 71 | |
| 72 | lbl = None |
| 73 | if use_labels: |
| 74 | lbl = [10000 + i for i in range(len(fnames))] |
| 75 | |
| 76 | batch_size = 3 |
| 77 | pipe = Pipeline(batch_size, 1, 0) |
| 78 | files, labels = fn.readers.file( |
| 79 | file_root=root, files=fnames, labels=lbl, random_shuffle=shuffle |
| 80 | ) |
| 81 | pipe.set_outputs(files, labels) |
| 82 | |
| 83 | num_iters = (len(fnames) + 2 * batch_size) // batch_size |
| 84 | for i in range(num_iters): |
| 85 | out_f, out_l = pipe.run() |
| 86 | for j in range(batch_size): |
| 87 | contents = bytes(out_f.at(j)).decode("utf-8") |
| 88 | label = out_l.at(j)[0] |
| 89 | index = label - 10000 if use_labels else label |
| 90 | assert contents == ref_contents(fnames[index]) |
| 91 | |
| 92 | |
| 93 | def test_file_reader(): |
nothing calls this directly
no test coverage detected