diff --git a/test/asynchronous/test_raw_bson.py b/test/asynchronous/test_raw_bson.py
new file mode 100644
index 0000000000..70832ea668
--- /dev/null
+++ b/test/asynchronous/test_raw_bson.py
@@ -0,0 +1,219 @@
+# 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.
+from __future__ import annotations
+
+import datetime
+import sys
+import uuid
+
+sys.path[0:0] = [""]
+
+from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
+
+from bson import Code, DBRef, decode, encode
+from bson.binary import JAVA_LEGACY, Binary, UuidRepresentation
+from bson.codec_options import CodecOptions
+from bson.errors import InvalidBSON
+from bson.raw_bson import DEFAULT_RAW_BSON_OPTIONS, RawBSONDocument
+from bson.son import SON
+
+_IS_SYNC = False
+
+
+class TestRawBSONDocument(AsyncIntegrationTest):
+    # {'_id': ObjectId('556df68b6e32ab21a95e0785'),
+    #  'name': 'Sherlock',
+    #  'addresses': [{'street': '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)
+
+    async def asyncTearDown(self):
+        if async_client_context.connected:
+            await 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(encode({}))
+        with self.assertRaises(KeyError):
+            doc["does-not-exist"]
+
+    def test_invalid_bson_sequence(self):
+        bson_byte_sequence = encode({"a": 1}) + encode({})
+        with self.assertRaisesRegex(InvalidBSON, "invalid object length"):
+            RawBSONDocument(bson_byte_sequence)
+
+    def test_invalid_bson_eoo(self):
+        invalid_bson_eoo = encode({"a": 1})[:-1] + b"\x01"
+        with self.assertRaisesRegex(InvalidBSON, "bad eoo"):
+            RawBSONDocument(invalid_bson_eoo)
+
+    @async_client_context.require_connection
+    async def test_round_trip(self):
+        db = self.client.get_database(
+            "pymongo_test", codec_options=CodecOptions(document_class=RawBSONDocument)
+        )
+        await db.test_raw.insert_one(self.document)
+        result = await db.test_raw.find_one(self.document["_id"])
+        assert result is not None
+        self.assertIsInstance(result, RawBSONDocument)
+        self.assertEqual(dict(self.document.items()), dict(result.items()))
+
+    @async_client_context.require_connection
+    async def test_round_trip_raw_uuid(self):
+        coll = self.client.get_database("pymongo_test").test_raw
+        uid = uuid.uuid4()
+        doc = {"_id": 1, "bin4": Binary(uid.bytes, 4), "bin3": Binary(uid.bytes, 3)}
+        raw = RawBSONDocument(encode(doc))
+        await coll.insert_one(raw)
+        self.assertEqual(await coll.find_one(), doc)
+        uuid_coll = coll.with_options(
+            codec_options=coll.codec_options.with_options(
+                uuid_representation=UuidRepresentation.STANDARD
+            )
+        )
+        self.assertEqual(
+            await uuid_coll.find_one(), {"_id": 1, "bin4": uid, "bin3": Binary(uid.bytes, 3)}
+        )
+
+        # Test that the raw bytes haven't changed.
+        raw_coll = coll.with_options(codec_options=DEFAULT_RAW_BSON_OPTIONS)
+        self.assertEqual(await raw_coll.find_one(), raw)
+
+    def test_with_codec_options(self):
+        # {'date': datetime.datetime(2015, 6, 3, 18, 40, 50, 826000),
+        #  '_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"])
+
+    @async_client_context.require_connection
+    async 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)
+        )
+        await 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(encode(doc, codec_options=raw_java_legacy)), await coll.find_one()
+        )
+
+    @async_client_context.require_connection
+    async def test_raw_bson_document_embedded(self):
+        doc = {"embedded": self.document}
+        db = self.client.pymongo_test
+        await db.test_raw.insert_one(doc)
+        result = await db.test_raw.find_one()
+        assert result is not None
+        self.assertEqual(decode(self.document.raw), result["embedded"])
+
+        # Make sure that CodecOptions are preserved.
+        # {'embedded': [
+        #   {'date': datetime.datetime(2015, 6, 3, 18, 40, 50, 826000),
+        #    '_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
+            ),
+        )
+
+        await db.test_raw.drop()
+        await db.test_raw.insert_one(rbd)
+        result = await db.get_collection(
+            "test_raw", codec_options=CodecOptions(uuid_representation=JAVA_LEGACY)
+        ).find_one()
+        assert result is not None
+        self.assertEqual(rbd["embedded"][0]["_id"], result["embedded"][0]["_id"])
+
+    @async_client_context.require_connection
+    async 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.
+        await coll.insert_one(self.document)
+        await coll.delete_one(self.document)
+        await coll.insert_many([self.document])
+        await coll.delete_many(self.document)
+        await coll.update_one(self.document, {"$set": {"a": "b"}}, upsert=True)
+        await coll.update_many(self.document, {"$set": {"b": "c"}})
+
+    def test_preserve_key_ordering(self):
+        keyvaluepairs = [
+            ("a", 1),
+            ("b", 2),
+            ("c", 3),
+        ]
+        rawdoc = RawBSONDocument(encode(SON(keyvaluepairs)))
+
+        for rkey, elt in zip(rawdoc, keyvaluepairs):
+            self.assertEqual(rkey, elt[0])
+
+    def test_contains_code_with_scope(self):
+        doc = RawBSONDocument(encode({"value": Code("x=1", scope={})}))
+
+        self.assertEqual(decode(encode(doc)), {"value": Code("x=1", {})})
+        self.assertEqual(doc["value"].scope, RawBSONDocument(encode({})))
+
+    def test_contains_dbref(self):
+        doc = RawBSONDocument(encode({"value": DBRef("test", "id")}))
+        raw = {"$ref": "test", "$id": "id"}
+        raw_encoded = encode(decode(encode(raw)))
+
+        self.assertEqual(decode(encode(doc)), {"value": DBRef("test", "id")})
+        self.assertEqual(doc["value"].raw, raw_encoded)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/test/test_raw_bson.py b/test/test_raw_bson.py
index 11bc80dd9f..4d9a3ceb05 100644
--- a/test/test_raw_bson.py
+++ b/test/test_raw_bson.py
@@ -19,8 +19,7 @@
 
 sys.path[0:0] = [""]
 
-from test import client_context, unittest
-from test.test_client import IntegrationTest
+from test import IntegrationTest, client_context, unittest
 
 from bson import Code, DBRef, decode, encode
 from bson.binary import JAVA_LEGACY, Binary, UuidRepresentation
@@ -29,6 +28,8 @@
 from bson.raw_bson import DEFAULT_RAW_BSON_OPTIONS, RawBSONDocument
 from bson.son import SON
 
+_IS_SYNC = True
+
 
 class TestRawBSONDocument(IntegrationTest):
     # {'_id': ObjectId('556df68b6e32ab21a95e0785'),
diff --git a/tools/synchro.py b/tools/synchro.py
index 59d6e653e5..0eca24b2cf 100644
--- a/tools/synchro.py
+++ b/tools/synchro.py
@@ -171,16 +171,17 @@
     "test_change_stream.py",
     "test_client.py",
     "test_client_bulk_write.py",
+    "test_client_context.py",
     "test_collection.py",
     "test_cursor.py",
     "test_database.py",
     "test_encryption.py",
     "test_grid_file.py",
     "test_logger.py",
+    "test_monitoring.py",
+    "test_raw_bson.py",
     "test_session.py",
     "test_transactions.py",
-    "test_client_context.py",
-    "test_monitoring.py",
 ]
 
 sync_test_files = [