(self)
| 110 | thread.join() |
| 111 | |
| 112 | def test_slices(self): |
| 113 | with self.test_session(): |
| 114 | works = [ |
| 115 | b"hdfs://path/to/my/file", |
| 116 | b"hdfs://prj/tables/tbl1", |
| 117 | b"hdfs://prj/tables/tbl1/pt=12", |
| 118 | b"hdfs://prj/tables/tbl1/pt=12/pt1=abc"] |
| 119 | num_epochs = 1 |
| 120 | work_queue = WorkQueue( |
| 121 | works, num_epochs=num_epochs, shuffle=False, num_slices=5) |
| 122 | |
| 123 | local_queue = work_queue.input_producer() |
| 124 | dequeue = local_queue.dequeue() |
| 125 | dequeue_many = local_queue.dequeue_many(len(works) * num_epochs) |
| 126 | |
| 127 | resources.initialize_resources(resources.shared_resources()).run() |
| 128 | variables.global_variables_initializer().run() |
| 129 | variables.local_variables_initializer().run() |
| 130 | threads = queue_runner_impl.start_queue_runners() |
| 131 | |
| 132 | local_works = dequeue_many.eval().tolist() |
| 133 | self.assertEqual( |
| 134 | works * num_epochs, |
| 135 | [item for work in local_works for item in work]) |
| 136 | |
| 137 | # Reached the limit. |
| 138 | with self.assertRaises(errors_impl.OutOfRangeError): |
| 139 | dequeue.eval() |
| 140 | for thread in threads: |
| 141 | thread.join() |
| 142 | |
| 143 | def test_monitored_session(self): |
| 144 | ps_hosts = ["localhost:{}".format(portpicker.pick_unused_port())] |
nothing calls this directly
no test coverage detected