(self)
| 1270 | self.assertGradientChecks(gc, op, [input_tensor], 0, [0]) |
| 1271 | |
| 1272 | def test_blobs_dequeue_timeout(self): |
| 1273 | op = core.CreateOperator( |
| 1274 | "CreateBlobsQueue", |
| 1275 | [], |
| 1276 | ["queue"], |
| 1277 | capacity=5, |
| 1278 | num_blobs=1) |
| 1279 | self.ws.run(op) |
| 1280 | t = time.time() |
| 1281 | op = core.CreateOperator( |
| 1282 | "DequeueBlobs", |
| 1283 | ["queue"], |
| 1284 | ["out"], |
| 1285 | timeout_secs=0.2) |
| 1286 | self.assertRaises(RuntimeError, lambda: self.ws.run(op)) |
| 1287 | t = time.time() - t |
| 1288 | self.assertGreater(t, 0.19) |
| 1289 | |
| 1290 | @given(num_threads=st.integers(1, 10), # noqa |
| 1291 | num_elements=st.integers(1, 100), |
nothing calls this directly
no test coverage detected