(device)
| 78 | |
| 79 | |
| 80 | def _test_permute_batch_fixed(device): |
| 81 | batch_size = 10 |
| 82 | pipe = Pipeline(batch_size, 4, 0) |
| 83 | data = fn.external_source( |
| 84 | source=lambda: gen_data(batch_size, np.int16), device=device, layout="abc" |
| 85 | ) |
| 86 | idxs = [4, 8, 0, 6, 3, 5, 2, 9, 7, 1] |
| 87 | pipe.set_outputs(data, fn.permute_batch(data, indices=idxs)) |
| 88 | |
| 89 | for _ in range(10): |
| 90 | orig, permuted = pipe.run() |
| 91 | orig = orig.as_cpu() |
| 92 | ref = [orig.at(idx) for idx in idxs] |
| 93 | check_batch(permuted, ref, len(ref), 0, 0, "abc") |
| 94 | |
| 95 | |
| 96 | def test_permute_batch_fixed(): |
nothing calls this directly
no test coverage detected