(BaseIterator, *args, **kwargs)
| 1226 | |
| 1227 | |
| 1228 | def check_iterator_wrapper_first_iteration(BaseIterator, *args, **kwargs): |
| 1229 | # This wrapper is used to test that the base class iterator doesn't invoke |
| 1230 | # the wrapper self.__next__ function accidentally |
| 1231 | class IteratorWrapper(BaseIterator): |
| 1232 | def __init__(self, *args, **kwargs): |
| 1233 | self._allow_next = False |
| 1234 | super(IteratorWrapper, self).__init__(*args, **kwargs) |
| 1235 | |
| 1236 | # Asserting if __next__ is called, unless self._allow_next has been set to True explicitly |
| 1237 | def __next__(self): |
| 1238 | assert self._allow_next |
| 1239 | _ = super(IteratorWrapper, self).__next__() |
| 1240 | |
| 1241 | pipe = Pipeline(batch_size=16, num_threads=1, device_id=0) |
| 1242 | with pipe: |
| 1243 | data = fn.random.uniform(range=(-1, 1), shape=(2, 2, 2), seed=1234) |
| 1244 | pipe.set_outputs(data) |
| 1245 | |
| 1246 | iterator_wrapper = IteratorWrapper([pipe], *args, **kwargs) |
| 1247 | # Only now, we allow the wrapper __next__ to run |
| 1248 | iterator_wrapper._allow_next = True |
| 1249 | for i, _ in enumerate(iterator_wrapper): |
| 1250 | if i == 2: |
| 1251 | break |
| 1252 | |
| 1253 | |
| 1254 | def check_external_source_autoreset(Iterator, *args, to_np=None, **kwargs): |
no test coverage detected