(self)
| 38 | class DataWorkersTest(unittest.TestCase): |
| 39 | |
| 40 | def testNonParallelModel(self): |
| 41 | workspace.ResetWorkspace() |
| 42 | |
| 43 | model = model_helper.ModelHelper(name="test") |
| 44 | old_seq_id = data_workers.global_coordinator._fetcher_id_seq |
| 45 | coordinator = data_workers.init_data_input_workers( |
| 46 | model, |
| 47 | ["data", "label"], |
| 48 | dummy_fetcher, |
| 49 | 32, |
| 50 | 2, |
| 51 | input_source_name="unittest" |
| 52 | ) |
| 53 | new_seq_id = data_workers.global_coordinator._fetcher_id_seq |
| 54 | self.assertEqual(new_seq_id, old_seq_id + 2) |
| 55 | |
| 56 | coordinator.start() |
| 57 | |
| 58 | workspace.RunNetOnce(model.param_init_net) |
| 59 | workspace.CreateNet(model.net) |
| 60 | |
| 61 | for _i in range(500): |
| 62 | with timeout_guard.CompleteInTimeOrDie(5): |
| 63 | workspace.RunNet(model.net.Proto().name) |
| 64 | |
| 65 | data = workspace.FetchBlob("data") |
| 66 | labels = workspace.FetchBlob("label") |
| 67 | |
| 68 | self.assertEqual(data.shape[0], labels.shape[0]) |
| 69 | self.assertEqual(data.shape[0], 32) |
| 70 | |
| 71 | for j in range(32): |
| 72 | self.assertEqual(labels[j], data[j, 0]) |
| 73 | self.assertEqual(labels[j], data[j, 1]) |
| 74 | self.assertEqual(labels[j], data[j, 2]) |
| 75 | |
| 76 | coordinator.stop_coordinator("unittest") |
| 77 | self.assertEqual(coordinator._coordinators, []) |
| 78 | |
| 79 | def testRNNInput(self): |
| 80 | workspace.ResetWorkspace() |
nothing calls this directly
no test coverage detected