| 32 | |
| 33 | |
| 34 | class TestRawBSONDocument(IntegrationTest): |
| 35 | # {'_id': ObjectId('556df68b6e32ab21a95e0785'), |
| 36 | # 'name': 'Sherlock', |
| 37 | # 'addresses': [{'street': 'Baker Street'}]} |
| 38 | bson_string = ( |
| 39 | b"Z\x00\x00\x00\x07_id\x00Um\xf6\x8bn2\xab!\xa9^\x07\x85\x02name\x00\t" |
| 40 | b"\x00\x00\x00Sherlock\x00\x04addresses\x00&\x00\x00\x00\x030\x00\x1e" |
| 41 | b"\x00\x00\x00\x02street\x00\r\x00\x00\x00Baker Street\x00\x00\x00\x00" |
| 42 | ) |
| 43 | document = RawBSONDocument(bson_string) |
| 44 | |
| 45 | def tearDown(self): |
| 46 | if client_context.connected: |
| 47 | self.client.pymongo_test.test_raw.drop() |
| 48 | |
| 49 | def test_decode(self): |
| 50 | self.assertEqual("Sherlock", self.document["name"]) |
| 51 | first_address = self.document["addresses"][0] |
| 52 | self.assertIsInstance(first_address, RawBSONDocument) |
| 53 | self.assertEqual("Baker Street", first_address["street"]) |
| 54 | |
| 55 | def test_raw(self): |
| 56 | self.assertEqual(self.bson_string, self.document.raw) |
| 57 | |
| 58 | def test_empty_doc(self): |
| 59 | doc = RawBSONDocument(encode({})) |
| 60 | with self.assertRaises(KeyError): |
| 61 | doc["does-not-exist"] |
| 62 | |
| 63 | def test_invalid_bson_sequence(self): |
| 64 | bson_byte_sequence = encode({"a": 1}) + encode({}) |
| 65 | with self.assertRaisesRegex(InvalidBSON, "invalid object length"): |
| 66 | RawBSONDocument(bson_byte_sequence) |
| 67 | |
| 68 | def test_invalid_bson_eoo(self): |
| 69 | invalid_bson_eoo = encode({"a": 1})[:-1] + b"\x01" |
| 70 | with self.assertRaisesRegex(InvalidBSON, "bad eoo"): |
| 71 | RawBSONDocument(invalid_bson_eoo) |
| 72 | |
| 73 | @client_context.require_connection |
| 74 | def test_round_trip(self): |
| 75 | db = self.client.get_database( |
| 76 | "pymongo_test", codec_options=CodecOptions(document_class=RawBSONDocument) |
| 77 | ) |
| 78 | db.test_raw.insert_one(self.document) |
| 79 | result = db.test_raw.find_one(self.document["_id"]) |
| 80 | assert result is not None |
| 81 | self.assertIsInstance(result, RawBSONDocument) |
| 82 | self.assertEqual(dict(self.document.items()), dict(result.items())) |
| 83 | |
| 84 | @client_context.require_connection |
| 85 | def test_round_trip_raw_uuid(self): |
| 86 | coll = self.client.get_database("pymongo_test").test_raw |
| 87 | uid = uuid.uuid4() |
| 88 | doc = {"_id": 1, "bin4": Binary(uid.bytes, 4), "bin3": Binary(uid.bytes, 3)} |
| 89 | raw = RawBSONDocument(encode(doc)) |
| 90 | coll.insert_one(raw) |
| 91 | self.assertEqual(coll.find_one(), doc) |
nothing calls this directly
no test coverage detected