better logic to handling unicode and chunking. (#870)

Chunking was looking at the length of a string but not accounting for the
byte length. closes #716

elasticsearch/helpers/__init__.py

@@ -58,11 +58,12 @@ def _chunk_actions(actions, chunk_size, max_chunk_bytes, serializer):
     for action, data in actions:
         raw_data, raw_action = data, action
         action = serializer.dumps(action)
-        cur_size = len(action) + 1
+        # +1 to account for the trailing new line character
+        cur_size = len(action.encode('utf-8')) + 1
 
         if data is not None:
             data = serializer.dumps(data)
-            cur_size += len(data) + 1
+            cur_size += len(data.encode('utf-8')) + 1
 
         # full chunk, send it and start a new one
         if bulk_actions and (size + cur_size > max_chunk_bytes or action_count == chunk_size):

test_elasticsearch/test_helpers.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 import mock
 import time
 import threading
@@ -46,7 +47,7 @@ class TestParallelBulk(TestCase):
 class TestChunkActions(TestCase):
     def setUp(self):
         super(TestChunkActions, self).setUp()
-        self.actions = [({'index': {}}, {'some': 'data', 'i': i}) for i in range(100)]
+        self.actions = [({'index': {}}, {'some': u'datá', 'i': i}) for i in range(100)]
 
     def test_chunks_are_chopped_by_byte_size(self):
         self.assertEquals(100, len(list(helpers._chunk_actions(self.actions, 100000, 1, JSONSerializer()))))
@@ -54,6 +55,15 @@ class TestChunkActions(TestCase):
     def test_chunks_are_chopped_by_chunk_size(self):
         self.assertEquals(10, len(list(helpers._chunk_actions(self.actions, 10, 99999999, JSONSerializer()))))
 
+    def test_chunks_are_chopped_by_byte_size_properly(self):
+        max_byte_size = 170
+        chunks = list(helpers._chunk_actions(self.actions, 100000, max_byte_size, JSONSerializer()))
+        self.assertEquals(25, len(chunks))
+        for chunk_data, chunk_actions in chunks:
+            chunk = u''.join(chunk_actions)
+            chunk = chunk if isinstance(chunk, str) else chunk.encode('utf-8')
+            self.assertLessEqual(len(chunk), max_byte_size)
+
 class TestExpandActions(TestCase):
     def test_string_actions_are_marked_as_simple_inserts(self):
         self.assertEquals(('{"index":{}}', "whatever"), helpers.expand_action('whatever'))