text: # Copyright 2015-present MongoDB, Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  # http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  import datetime
  import uuid
  
  from bson import BSON
  from bson.binary import JAVA_LEGACY
  from bson.codec_options import CodecOptions
  from bson.errors import InvalidBSON
  from bson.raw_bson import RawBSONDocument
  from bson.son import SON
  from test import client_context, unittest
  
  
  class TestRawBSONDocument(unittest.TestCase):
  
      # {u'_id': ObjectId('556df68b6e32ab21a95e0785'),
      #  u'name': u'Sherlock',
      #  u'addresses': [{u'street': u'Baker Street'}]}
      bson_string = (
          b'Z\x00\x00\x00\x07_id\x00Um\xf6\x8bn2\xab!\xa9^\x07\x85\x02name\x00\t'
          b'\x00\x00\x00Sherlock\x00\x04addresses\x00&\x00\x00\x00\x030\x00\x1e'
          b'\x00\x00\x00\x02street\x00\r\x00\x00\x00Baker Street\x00\x00\x00\x00'
      )
      document = RawBSONDocument(bson_string)
  
      @classmethod
      def setUpClass(cls):
          cls.client = client_context.client
  
      def tearDown(self):
          if client_context.connected:
              self.client.pymongo_test.test_raw.drop()
  
      def test_decode(self):
          self.assertEqual('Sherlock', self.document['name'])
          first_address = self.document['addresses'][0]
          self.assertIsInstance(first_address, RawBSONDocument)
          self.assertEqual('Baker Street', first_address['street'])
  
      def test_raw(self):
          self.assertEqual(self.bson_string, self.document.raw)
  
      def test_empty_doc(self):
          doc = RawBSONDocument(BSON.encode({}))
          with self.assertRaises(KeyError):
              doc['does-not-exist']
  
      def test_invalid_bson_sequence(self):
          bson_byte_sequence = BSON.encode({'a': 1})+BSON.encode({})
          with self.assertRaisesRegex(InvalidBSON, 'invalid object length'):
              RawBSONDocument(bson_byte_sequence)
  
      def test_invalid_bson_eoo(self):
          invalid_bson_eoo = BSON.encode({'a': 1})[:-1] + b'\x01'
          with self.assertRaisesRegex(InvalidBSON, 'bad eoo'):
              RawBSONDocument(invalid_bson_eoo)
  
      @client_context.require_connection
      def test_round_trip(self):
          db = self.client.get_database(
              'pymongo_test',
              codec_options=CodecOptions(document_class=RawBSONDocument))
          db.test_raw.insert_one(self.document)
          result = db.test_raw.find_one(self.document['_id'])
          self.assertIsInstance(result, RawBSONDocument)
          self.assertEqual(dict(self.document.items()), dict(result.items()))
  
      def test_with_codec_options(self):
          # {u'date': datetime.datetime(2015, 6, 3, 18, 40, 50, 826000),
          #  u'_id': UUID('026fab8f-975f-4965-9fbf-85ad874c60ff')}
          # encoded with JAVA_LEGACY uuid representation.
          bson_string = (
              b'-\x00\x00\x00\x05_id\x00\x10\x00\x00\x00\x03eI_\x97\x8f\xabo\x02'
              b'\xff`L\x87\xad\x85\xbf\x9f\tdate\x00\x8a\xd6\xb9\xbaM'
              b'\x01\x00\x00\x00'
          )
          document = RawBSONDocument(
              bson_string,
              codec_options=CodecOptions(uuid_representation=JAVA_LEGACY,
                                         document_class=RawBSONDocument))
  
          self.assertEqual(uuid.UUID('026fab8f-975f-4965-9fbf-85ad874c60ff'),
                           document['_id'])
  
      @client_context.require_connection
      def test_round_trip_codec_options(self):
          doc = {
              'date': datetime.datetime(2015, 6, 3, 18, 40, 50, 826000),
              '_id': uuid.UUID('026fab8f-975f-4965-9fbf-85ad874c60ff')
          }
          db = self.client.pymongo_test
          coll = db.get_collection(
              'test_raw',
              codec_options=CodecOptions(uuid_representation=JAVA_LEGACY))
          coll.insert_one(doc)
          raw_java_legacy = CodecOptions(uuid_representation=JAVA_LEGACY,
                                         document_class=RawBSONDocument)
          coll = db.get_collection('test_raw', codec_options=raw_java_legacy)
          self.assertEqual(
              RawBSONDocument(BSON.encode(doc, codec_options=raw_java_legacy)),
              coll.find_one())
  
      @client_context.require_connection
      def test_raw_bson_document_embedded(self):
          doc = {'embedded': self.document}
          db = self.client.pymongo_test
          db.test_raw.insert_one(doc)
          result = db.test_raw.find_one()
          self.assertEqual(BSON(self.document.raw).decode(), result['embedded'])
  
          # Make sure that CodecOptions are preserved.
          # {'embedded': [
          #   {u'date': datetime.datetime(2015, 6, 3, 18, 40, 50, 826000),
          #    u'_id': UUID('026fab8f-975f-4965-9fbf-85ad874c60ff')}
          # ]}
          # encoded with JAVA_LEGACY uuid representation.
          bson_string = (
              b'D\x00\x00\x00\x04embedded\x005\x00\x00\x00\x030\x00-\x00\x00\x00'
              b'\tdate\x00\x8a\xd6\xb9\xbaM\x01\x00\x00\x05_id\x00\x10\x00\x00'
              b'\x00\x03eI_\x97\x8f\xabo\x02\xff`L\x87\xad\x85\xbf\x9f\x00\x00'
              b'\x00'
          )
          rbd = RawBSONDocument(
              bson_string,
              codec_options=CodecOptions(uuid_representation=JAVA_LEGACY,
                                         document_class=RawBSONDocument))
  
          db.test_raw.drop()
          db.test_raw.insert_one(rbd)
          result = db.get_collection('test_raw', codec_options=CodecOptions(
              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'}})
  
      def test_preserve_key_ordering(self):
          keyvaluepairs = [('a', 1), ('b', 2), ('c', 3),]
          rawdoc = RawBSONDocument(BSON.encode(SON(keyvaluepairs)))
  
          for rkey, elt in zip(rawdoc, keyvaluepairs):
              self.assertEqual(rkey, elt[0])