(device, is_serialized)
| 47 | |
| 48 | |
| 49 | def _test_feed_input(device, is_serialized): |
| 50 | src_pipe, batch_size = build_src_pipe(device) |
| 51 | |
| 52 | dst_pipe = Pipeline(batch_size, 1, 0, exec_async=False, exec_pipelined=False) |
| 53 | dst_pipe.set_outputs(fn.external_source(name="ext", device=device)) |
| 54 | if is_serialized: |
| 55 | serialized = dst_pipe.serialize() |
| 56 | dst_pipe = None |
| 57 | dst_pipe = Pipeline.deserialize( |
| 58 | serialized_pipeline=serialized, |
| 59 | batch_size=batch_size, |
| 60 | num_threads=1, |
| 61 | device_id=0, |
| 62 | exec_async=False, |
| 63 | exec_pipelined=False, |
| 64 | ) |
| 65 | |
| 66 | for _ in range(3): |
| 67 | out1 = src_pipe.run() |
| 68 | dst_pipe.feed_input("ext", out1[0]) |
| 69 | out2 = dst_pipe.run() |
| 70 | check_batch(out2[0], out1[0], batch_size, 0, 0, "XY") |
| 71 | |
| 72 | |
| 73 | def test_feed_input(): |
nothing calls this directly
no test coverage detected