test_bulk_utility.py 文件源码

def test_bulk_stats_only(self):
        return_value = [succeed([ITEM_SUCCESS, ITEM_SUCCESS, ITEM_FAILED])]
        self.bulk_utility.streaming_bulk = MagicMock(return_value=return_value)
        success, faileds = yield self.bulk_utility.bulk(None, stats_only=True)
        self.assertEqual(2, success)
        self.assertEqual(1, faileds)