(rank, axes, device, in_layout)
| 616 | |
| 617 | @nottest |
| 618 | def _test_reduce_empty_data(rank, axes, device, in_layout): |
| 619 | batch_size = 16 |
| 620 | num_batches = 2 |
| 621 | data = empty_batches(rank, axes, batch_size, num_batches) |
| 622 | |
| 623 | pipe = Pipeline(batch_size=batch_size, num_threads=4, device_id=0 if device == "gpu" else None) |
| 624 | input = fn.external_source(data, cycle=True, device=device, layout=in_layout) |
| 625 | reduced = fn.reductions.sum(input, axes=axes) |
| 626 | pipe.set_outputs(reduced) |
| 627 | |
| 628 | for b, batch in enumerate(data): |
| 629 | (out,) = pipe.run() |
| 630 | check_layout(out, in_layout, axes, False) |
| 631 | if device == "gpu": |
| 632 | out = out.as_cpu() |
| 633 | for i in range(batch_size): |
| 634 | ref = np.sum(batch[i].astype(np.float64), axis=axes) |
| 635 | assert np.allclose(out[i], ref, 1e-5, 1e-5) |
| 636 | |
| 637 | |
| 638 | def test_reduce_empty_data(): |
nothing calls this directly
no test coverage detected