Make sure actions aren't modified when passed into bulk

elasticsearch/helpers.py

@@ -20,6 +20,8 @@ def expand_action(data):
     action/data lines needed for elasticsearch's
     :meth:`~elasticsearch.Elasticsearch.bulk` api.
     """
+    # make sure we don't alter the action
+    data = data.copy()
     op_type = data.pop('_op_type', 'index')
     action = {op_type: {}}
     for key in ('_index', '_parent', '_percolate', '_routing', '_timestamp',
@@ -57,8 +59,7 @@ def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False, expand
         }
 
     Alternatively, if `_source` is not present, it will pop all metadata fields
-    from the doc and use the rest as the document data. The dict passed in will
-    be modified - metadata fields will be popped out.
+    from the doc and use the rest as the document data.
 
     Alternative actions (`_op_type` field defaults to `index`) can be sent as
     well::

test_elasticsearch/test_server/test_helpers.py

@@ -4,6 +4,12 @@ from . import ElasticTestCase
 from ..test_cases import SkipTest
 
 class TestStreamingBulk(ElasticTestCase):
+    def test_actions_remain_unchanged(self):
+        actions = [{'_id': 1}, {'_id': 2}]
+        for ok, item in helpers.streaming_bulk(self.client, actions, index='test-index', doc_type='answers'):
+            self.assertTrue(ok)
+        self.assertEquals([{'_id': 1}, {'_id': 2}], actions)
+
     def test_all_documents_get_inserted(self):
         docs = [{"answer": x, '_id': x} for x in range(100)]
         for ok, item in helpers.streaming_bulk(self.client, docs, index='test-index', doc_type='answers', refresh=True):