PYTHON-1090, PYTHON-1098 - Use sane codec options when reading write responses.

When reading a write response from the server, we now use the 'replace'
unicode_decode_error_handler and 'dict' as the document_class.

Author: Luke Lovett

pymongo/bulk.py

@@ -203,7 +203,10 @@ class _Bulk(object):
     def __init__(self, collection, ordered, bypass_document_validation):
         """Initialize a _Bulk instance.
         """
-        self.collection = collection
+        self.collection = collection.with_options(
+            codec_options=collection.codec_options._replace(
+                unicode_decode_error_handler='replace',
+                document_class=dict))
         self.ordered = ordered
         self.ops = []
         self.name = "%s.%s" % (collection.database.name, collection.name)

pymongo/collection.py

@@ -163,6 +163,10 @@ def __init__(self, database, name, create=False, codec_options=None,
         if create or kwargs:
             self.__create(kwargs)
 
+        self.__write_response_codec_options = self.codec_options._replace(
+            unicode_decode_error_handler='replace',
+            document_class=dict)
+
     def _socket_for_reads(self):
         return self.__database.client._socket_for_reads(self.read_preference)
 
@@ -506,17 +510,19 @@ def _insert_one(
             if bypass_doc_val and sock_info.max_wire_version >= 4:
                 command['bypassDocumentValidation'] = True
             # Insert command.
-            result = sock_info.command(self.__database.name,
-                                       command,
-                                       codec_options=self.codec_options,
-                                       check_keys=check_keys)
+            result = sock_info.command(
+                self.__database.name,
+                command,
+                codec_options=self.__write_response_codec_options,
+                check_keys=check_keys)
             _check_write_command_response([(0, result)])
         else:
             # Legacy OP_INSERT.
             self._legacy_write(
                 sock_info, 'insert', command, acknowledged, op_id,
                 bypass_doc_val, message.insert, self.__full_name, [doc],
-                check_keys, acknowledged, concern, False, self.codec_options)
+                check_keys, acknowledged, concern, False,
+                self.__write_response_codec_options)
         if not isinstance(doc, RawBSONDocument):
             return doc.get('_id')
 
@@ -575,13 +581,13 @@ def gen():
             # Batched insert command.
             results = message._do_batched_write_command(
                 self.database.name + ".$cmd", message._INSERT, command,
-                gen(), check_keys, self.codec_options, bwc)
+                gen(), check_keys, self.__write_response_codec_options, bwc)
             _check_write_command_response(results)
         else:
             # Legacy batched OP_INSERT.
             message._do_batched_insert(self.__full_name, gen(), check_keys,
                                        acknowledged, concern, not ordered,
-                                       self.codec_options, bwc)
+                                       self.__write_response_codec_options, bwc)
         return ids
 
     def insert_one(self, document, bypass_document_validation=False):
@@ -704,9 +710,10 @@ def _update(self, sock_info, criteria, document, upsert=False,
 
             # The command result has to be published for APM unmodified
             # so we make a shallow copy here before adding updatedExisting.
-            result = sock_info.command(self.__database.name,
-                                       command,
-                                       codec_options=self.codec_options).copy()
+            result = sock_info.command(
+                self.__database.name,
+                command,
+                codec_options=self.__write_response_codec_options).copy()
             _check_write_command_response([(0, result)])
             # Add the updatedExisting field for compatibility.
             if result.get('n') and 'upserted' not in result:
@@ -725,7 +732,7 @@ def _update(self, sock_info, criteria, document, upsert=False,
                 sock_info, 'update', command, acknowledged, op_id,
                 bypass_doc_val, message.update, self.__full_name, upsert,
                 multi, criteria, document, acknowledged, concern, check_keys,
-                self.codec_options)
+                self.__write_response_codec_options)
 
     def replace_one(self, filter, replacement, upsert=False,
                     bypass_document_validation=False):
@@ -911,17 +918,19 @@ def _delete(
 
         if sock_info.max_wire_version > 1 and acknowledged:
             # Delete command.
-            result = sock_info.command(self.__database.name,
-                                       command,
-                                       codec_options=self.codec_options)
+            result = sock_info.command(
+                self.__database.name,
+                command,
+                codec_options=self.__write_response_codec_options)
             _check_write_command_response([(0, result)])
             return result
         else:
             # Legacy OP_DELETE.
             return self._legacy_write(
                 sock_info, 'delete', command, acknowledged, op_id,
                 False, message.delete, self.__full_name, criteria,
-                acknowledged, concern, self.codec_options, int(not multi))
+                acknowledged, concern, self.__write_response_codec_options,
+                int(not multi))
 
     def delete_one(self, filter):
         """Delete a single document matching the filter.
@@ -1192,11 +1201,11 @@ def parallel_scan(self, num_cursors):
     def _count(self, cmd):
         """Internal count helper."""
         with self._socket_for_reads() as (sock_info, slave_ok):
-            res = self._command(sock_info, cmd, slave_ok,
-                                allowable_errors=["ns missing"],
-                                codec_options=self.codec_options._replace(
-                                    document_class=dict),
-                                read_concern=self.read_concern)
+            res = self._command(
+                sock_info, cmd, slave_ok,
+                allowable_errors=["ns missing"],
+                codec_options=self.__write_response_codec_options,
+                read_concern=self.read_concern)
         if res.get("errmsg", "") == "ns missing":
             return 0
         return int(res["n"])

pymongo/helpers.py

@@ -38,6 +38,9 @@
 
 _UUNDER = u("_")
 
+_UNICODE_REPLACE_CODEC_OPTIONS = CodecOptions(
+    unicode_decode_error_handler='replace')
+
 
 def _gen_index_name(keys):
     """Generate an index name from the set of fields it is over."""
@@ -87,7 +90,9 @@ def _index_document(index_list):
     return index
 
 
-def _unpack_response(response, cursor_id=None, codec_options=CodecOptions()):
+def _unpack_response(response,
+                     cursor_id=None,
+                     codec_options=_UNICODE_REPLACE_CODEC_OPTIONS):
     """Unpack a response from the database.
 
     Check the response for errors and unpack, returning a dictionary

test/test_collection.py

@@ -1116,6 +1116,75 @@ def test_duplicate_key_error(self):
         self.assertIsNotNone(context.exception.details)
         self.assertEqual(1, db.test.count())
 
+    def test_write_error_text_handling(self):
+        db = self.db
+        db.drop_collection("test")
+
+        db.test.create_index("text", unique=True)
+
+        # Test workaround for SERVER-24007
+        data = (b'a\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83'
+                b'\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83\xe2\x98\x83')
+
+        text = data.decode("utf8")
+        db.test.insert_one({"text": text})
+
+        # Should raise DuplicateKeyError, not InvalidBSON
+        self.assertRaises(DuplicateKeyError,
+                          db.test.insert_one,
+                          {"text": text})
+
+        self.assertRaises(DuplicateKeyError,
+                          db.test.insert,
+                          {"text": text})
+
+        self.assertRaises(DuplicateKeyError,
+                          db.test.insert,
+                          [{"text": text}])
+
+        self.assertRaises(DuplicateKeyError,
+                          db.test.replace_one,
+                          {"_id": ObjectId()},
+                          {"text": text},
+                          upsert=True)
+
+        self.assertRaises(DuplicateKeyError,
+                          db.test.update,
+                          {"_id": ObjectId()},
+                          {"text": text},
+                          upsert=True)
+
+        # Should raise BulkWriteError, not InvalidBSON
+        self.assertRaises(BulkWriteError,
+                          db.test.insert_many,
+                          [{"text": text}])
+
     def test_wtimeout(self):
         # Ensure setting wtimeout doesn't disable write concern altogether.
         # See SERVER-12596.

test/test_raw_bson.py

@@ -111,3 +111,17 @@ def test_raw_bson_document_embedded(self):
             uuid_representation=JAVA_LEGACY)).find_one()
         self.assertEqual(rbd['embedded'][0]['_id'],
                          result['embedded'][0]['_id'])
+
+    @client_context.require_connection
+    def test_write_response_raw_bson(self):
+        coll = self.client.get_database(
+            'pymongo_test',
+            codec_options=CodecOptions(document_class=RawBSONDocument)).test_raw
+
+        # No Exceptions raised while handling write response.
+        coll.insert_one(self.document)
+        coll.delete_one(self.document)
+        coll.insert_many([self.document])
+        coll.delete_many(self.document)
+        coll.update_one(self.document, {'$set': {'a': 'b'}}, upsert=True)
+        coll.update_many(self.document, {'$set': {'b': 'c'}})