def verify_shard_state(self, shard_state, **kwargs):
    self.assertTrue(shard_state)

    self.assertEqual(kwargs.get("active", True), shard_state.active)
    self.assertEqual(kwargs.get("processed", 0),
                     shard_state.counters_map.get(COUNTER_MAPPER_CALLS))
    self.assertEqual(kwargs.get("result_status", None),
                     shard_state.result_status)
    self.assertEqual(kwargs.get("slice_retries", 0),
                     shard_state.slice_retries)
    self.assertEqual(kwargs.get("retries", 0),
                     shard_state.retries)
    self.assertEqual(kwargs.get("input_finished", False),
                     shard_state.input_finished)
Even when writes are blocked on reading, loseConnection() waits for
dataset/ETHPy150Open GoogleCloudPlatform/appengine-mapreduce/python/test/mapreduce/handlers_test.py/MapreduceHandlerTestBase.verify_shard_state--crossover with--dataset/ETHPy150Open nlloyd/SubliminalCollaborator/libs/twisted/protocols/test/test_tls.py/TLSProducerTests.test_streamingProducerLoseConnectionWithProducerWBOR