diff --git a/.evergreen/scripts/setup_tests.py b/.evergreen/scripts/setup_tests.py
index 1a41b3a8b0..90f700bc6b 100644
--- a/.evergreen/scripts/setup_tests.py
+++ b/.evergreen/scripts/setup_tests.py
@@ -353,29 +353,42 @@ def handle_test_env() -> None:
         UV_ARGS.append("--extra zstd")
 
     if test_name in ["encryption", "kms"]:
-        # Check for libmongocrypt download.
-        if not (ROOT / "libmongocrypt").exists():
-            setup_libmongocrypt()
+        # The "String" algorithm and the GA prefix/suffix/substring query types
+        # need libmongocrypt 1.19.0+, which is only exercised against MongoDB
+        # 9.0+. Servers before 9.0 test the preview query types instead, which
+        # need the "textPreview" algorithm, so pin to the released pymongocrypt
+        # and use the libmongocrypt bundled in its wheel rather than the
+        # unreleased master build.
+        use_released_pymongocrypt = os.environ.get("MONGODB_VERSION", "").startswith("8.")
+
+        if not use_released_pymongocrypt:
+            # Check for libmongocrypt download.
+            if not (ROOT / "libmongocrypt").exists():
+                setup_libmongocrypt()
 
         if not opts.test_min_deps:
-            UV_ARGS.append(
-                "--with pymongocrypt@git+https://github.com/mongodb/libmongocrypt@master#subdirectory=bindings/python"
-            )
-
-        # Use the nocrypto build to avoid dependency issues with older windows/python versions.
-        BASE = ROOT / "libmongocrypt/nocrypto"
-        if PLATFORM == "linux":
-            if (BASE / "lib/libmongocrypt.so").exists():
-                PYMONGOCRYPT_LIB = BASE / "lib/libmongocrypt.so"
+            if use_released_pymongocrypt:
+                UV_ARGS.append("--with pymongocrypt<1.19")
             else:
-                PYMONGOCRYPT_LIB = BASE / "lib64/libmongocrypt.so"
-        elif PLATFORM == "darwin":
-            PYMONGOCRYPT_LIB = BASE / "lib/libmongocrypt.dylib"
-        else:
-            PYMONGOCRYPT_LIB = BASE / "bin/mongocrypt.dll"
-        if not PYMONGOCRYPT_LIB.exists():
-            raise RuntimeError("Cannot find libmongocrypt shared object file")
-        write_env("PYMONGOCRYPT_LIB", PYMONGOCRYPT_LIB.as_posix())
+                UV_ARGS.append(
+                    "--with pymongocrypt@git+https://github.com/mongodb/libmongocrypt@master#subdirectory=bindings/python"
+                )
+
+        if not use_released_pymongocrypt:
+            # Use the nocrypto build to avoid dependency issues with older windows/python versions.
+            BASE = ROOT / "libmongocrypt/nocrypto"
+            if PLATFORM == "linux":
+                if (BASE / "lib/libmongocrypt.so").exists():
+                    PYMONGOCRYPT_LIB = BASE / "lib/libmongocrypt.so"
+                else:
+                    PYMONGOCRYPT_LIB = BASE / "lib64/libmongocrypt.so"
+            elif PLATFORM == "darwin":
+                PYMONGOCRYPT_LIB = BASE / "lib/libmongocrypt.dylib"
+            else:
+                PYMONGOCRYPT_LIB = BASE / "bin/mongocrypt.dll"
+            if not PYMONGOCRYPT_LIB.exists():
+                raise RuntimeError("Cannot find libmongocrypt shared object file")
+            write_env("PYMONGOCRYPT_LIB", PYMONGOCRYPT_LIB.as_posix())
         # PATH is updated by configure-env.sh for access to mongocryptd.
 
     if test_name == "encryption":
diff --git a/doc/changelog.rst b/doc/changelog.rst
index fb7d300b2e..14e5238be8 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -39,6 +39,23 @@ PyMongo 4.18 brings a number of changes including:
 - Fixed a bug on Windows, and on macOS when using PyOpenSSL, where
   ``SSL_CERT_FILE``/``SSL_CERT_DIR`` were merged with, rather than replacing,
   the OS/certifi certificate store.
+- Added general availability support for Queryable Encryption prefix, suffix,
+  and substring queries against MongoDB 9.0+. Prefix and suffix queries require
+  libmongocrypt 1.19.0 or later; substring queries require libmongocrypt 1.20.0
+  or later:
+
+  - Added :attr:`~pymongo.encryption.Algorithm.STRING` and
+    :class:`~pymongo.encryption_options.StringOpts`, replacing
+    ``Algorithm.TEXTPREVIEW`` and ``TextOpts``, which are now deprecated.
+  - Added :attr:`~pymongo.encryption.QueryType.PREFIX`,
+    :attr:`~pymongo.encryption.QueryType.SUFFIX`, and
+    :attr:`~pymongo.encryption.QueryType.SUBSTRING`. The corresponding
+    ``PREFIXPREVIEW``, ``SUFFIXPREVIEW``, and ``SUBSTRINGPREVIEW`` query types
+    remain for experimental use with MongoDB versions before 9.0.
+  - Added the ``string_opts`` parameter to
+    :meth:`~pymongo.encryption.ClientEncryption.encrypt` and
+    :meth:`~pymongo.asynchronous.encryption.AsyncClientEncryption.encrypt`,
+    deprecating ``text_opts``.
 
 Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
diff --git a/pymongo/asynchronous/encryption.py b/pymongo/asynchronous/encryption.py
index 524ae45c11..49b32edd93 100644
--- a/pymongo/asynchronous/encryption.py
+++ b/pymongo/asynchronous/encryption.py
@@ -22,6 +22,7 @@
 import socket
 import time as time  # noqa: PLC0414 # needed in sync version
 import uuid
+import warnings
 import weakref
 from collections.abc import AsyncGenerator, Iterator, Mapping, MutableMapping, Sequence
 from copy import deepcopy
@@ -65,7 +66,10 @@
 from pymongo.encryption_options import (
     AutoEncryptionOpts,
     RangeOpts,
-    TextOpts,
+    StringOpts,
+    # Re-exported for backwards compatibility: TextOpts is deprecated but must
+    # remain importable from this module until it is removed.
+    TextOpts,  # noqa: F401
     check_min_pymongocrypt,
 )
 from pymongo.errors import (
@@ -529,8 +533,15 @@ class Algorithm(str, enum.Enum):
 
     .. versionadded:: 4.4
     """
+    STRING = "String"
+    """String.
+
+    .. versionadded:: 4.18
+    """
     TEXTPREVIEW = "TextPreview"
-    """**BETA** - TextPreview.
+    """**DEPRECATED** - TextPreview.
+
+    .. note:: Support for TextPreview is deprecated. Use :attr:`Algorithm.STRING` instead.
 
     .. versionadded:: 4.15
     """
@@ -559,25 +570,77 @@ class QueryType(str, enum.Enum):
     .. versionadded:: 4.4
     """
 
+    PREFIX = "prefix"
+    """Used to encrypt a value for a prefix query.
+
+    Used for the ``$encStrStartsWith`` operator. Requires MongoDB 9.0+.
+
+    .. versionadded:: 4.18
+    """
+
+    SUFFIX = "suffix"
+    """Used to encrypt a value for a suffix query.
+
+    Used for the ``$encStrEndsWith`` operator. Requires MongoDB 9.0+.
+
+    .. versionadded:: 4.18
+    """
+
+    SUBSTRING = "substring"
+    """Used to encrypt a value for a substring query.
+
+    Used for the ``$encStrContains`` operator. Requires MongoDB 9.0+.
+
+    .. versionadded:: 4.18
+    """
+
     PREFIXPREVIEW = "prefixPreview"
     """**BETA** - Used to encrypt a value for a prefixPreview query.
 
+    .. note:: The preview query types are for experimental workloads only and
+       are only supported by MongoDB versions before 9.0. Use
+       :attr:`QueryType.PREFIX` instead.
+
     .. versionadded:: 4.15
     """
 
     SUFFIXPREVIEW = "suffixPreview"
     """**BETA** - Used to encrypt a value for a suffixPreview query.
 
+    .. note:: The preview query types are for experimental workloads only and
+       are only supported by MongoDB versions before 9.0. Use
+       :attr:`QueryType.SUFFIX` instead.
+
     .. versionadded:: 4.15
     """
 
     SUBSTRINGPREVIEW = "substringPreview"
     """**BETA** - Used to encrypt a value for a substringPreview query.
 
+    .. note:: The preview query types are for experimental workloads only and
+       are only supported by MongoDB versions before 9.0. Use
+       :attr:`QueryType.SUBSTRING` instead.
+
     .. versionadded:: 4.15
     """
 
 
+def _resolve_string_opts(
+    string_opts: Optional[StringOpts], text_opts: Optional[StringOpts]
+) -> Optional[StringOpts]:
+    """Resolve the deprecated text_opts alias for string_opts."""
+    if text_opts is None:
+        return string_opts
+    if string_opts is not None:
+        raise ConfigurationError("Cannot set both string_opts and text_opts")
+    warnings.warn(
+        "The text_opts parameter is deprecated. Use string_opts instead.",
+        DeprecationWarning,
+        stacklevel=3,
+    )
+    return text_opts
+
+
 def _create_mongocrypt_options(**kwargs: Any) -> MongoCryptOptions:
     # For compat with pymongocrypt <1.13, avoid setting the default key_expiration_ms.
     if kwargs.get("key_expiration_ms") is None:
@@ -917,7 +980,7 @@ async def _encrypt_helper(
         contention_factor: Optional[int] = None,
         range_opts: Optional[RangeOpts] = None,
         is_expression: bool = False,
-        text_opts: Optional[TextOpts] = None,
+        string_opts: Optional[StringOpts] = None,
     ) -> Any:
         self._check_closed()
         if isinstance(key_id, uuid.UUID):
@@ -937,10 +1000,10 @@ async def _encrypt_helper(
                 range_opts.document,
                 codec_options=self._codec_options,
             )
-        text_opts_bytes = None
-        if text_opts:
-            text_opts_bytes = encode(
-                text_opts.document,
+        string_opts_bytes = None
+        if string_opts:
+            string_opts_bytes = encode(
+                string_opts.document,
                 codec_options=self._codec_options,
             )
         with _wrap_encryption_errors():
@@ -953,8 +1016,9 @@ async def _encrypt_helper(
                 contention_factor=contention_factor,
                 range_opts=range_opts_bytes,
                 is_expression=is_expression,
+                # pymongocrypt still names this parameter text_opts.
                 # For compatibility with pymongocrypt < 1.16:
-                **{"text_opts": text_opts_bytes} if text_opts_bytes else {},
+                **{"text_opts": string_opts_bytes} if string_opts_bytes else {},
             )
             return decode(encrypted_doc)["v"]
 
@@ -967,7 +1031,8 @@ async def encrypt(
         query_type: Optional[str] = None,
         contention_factor: Optional[int] = None,
         range_opts: Optional[RangeOpts] = None,
-        text_opts: Optional[TextOpts] = None,
+        string_opts: Optional[StringOpts] = None,
+        text_opts: Optional[StringOpts] = None,
     ) -> Binary:
         """Encrypt a BSON value with a given key and algorithm.
 
@@ -988,11 +1053,15 @@ async def encrypt(
             used.
         :param range_opts: Index options for `range` queries. See
             :class:`RangeOpts` for some valid options.
-        :param text_opts: Index options for `textPreview` queries. See
-            :class:`TextOpts` for some valid options.
+        :param string_opts: Index options for `prefix`, `suffix`, and
+            `substring` queries. See :class:`StringOpts` for some valid options.
+        :param text_opts: **DEPRECATED** - Alias for `string_opts`.
 
         :return: The encrypted value, a :class:`~bson.binary.Binary` with subtype 6.
 
+        .. versionchanged:: 4.18
+           Added the `string_opts` parameter and deprecated `text_opts`.
+
         .. versionchanged:: 4.9
            Added the `text_opts` parameter.
 
@@ -1016,7 +1085,7 @@ async def encrypt(
                 contention_factor=contention_factor,
                 range_opts=range_opts,
                 is_expression=False,
-                text_opts=text_opts,
+                string_opts=_resolve_string_opts(string_opts, text_opts),
             ),
         )
 
diff --git a/pymongo/encryption_options.py b/pymongo/encryption_options.py
index f2fcd47c65..065e7f1590 100644
--- a/pymongo/encryption_options.py
+++ b/pymongo/encryption_options.py
@@ -19,6 +19,7 @@
 
 from __future__ import annotations
 
+import warnings
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Optional, TypedDict
 
@@ -312,10 +313,8 @@ def document(self) -> dict[str, Any]:
         return doc
 
 
-class TextOpts:
-    """**BETA** Options to configure encrypted queries using the text algorithm.
-
-    TextOpts is currently unstable API and subject to backwards breaking changes."""
+class StringOpts:
+    """Options to configure encrypted queries using the string algorithm."""
 
     def __init__(
         self,
@@ -325,15 +324,16 @@ def __init__(
         case_sensitive: Optional[bool] = None,
         diacritic_sensitive: Optional[bool] = None,
     ) -> None:
-        """Options to configure encrypted queries using the text algorithm.
+        """Options to configure encrypted queries using the string algorithm.
 
         :param substring: Further options to support substring queries.
         :param prefix: Further options to support prefix queries.
         :param suffix: Further options to support suffix queries.
-        :param case_sensitive: Whether text indexes for this field are case sensitive.
-        :param diacritic_sensitive: Whether text indexes for this field are diacritic sensitive.
+        :param case_sensitive: Whether string indexes for this field are case sensitive.
+        :param diacritic_sensitive: Whether string indexes for this field are diacritic sensitive.
 
-        .. versionadded:: 4.15
+        .. versionadded:: 4.18
+           ``StringOpts`` replaces ``TextOpts``, which is deprecated.
         """
         self.substring = substring
         self.prefix = prefix
@@ -357,9 +357,9 @@ def document(self) -> dict[str, Any]:
 
 
 class SubstringOpts(TypedDict):
-    """**BETA** Options for substring text queries.
+    """Options for substring string queries.
 
-    SubstringOpts is currently unstable API and subject to backwards breaking changes.
+    .. versionadded:: 4.15
     """
 
     # strMaxLength is the maximum allowed length to insert. Inserting longer strings will error.
@@ -371,9 +371,9 @@ class SubstringOpts(TypedDict):
 
 
 class PrefixOpts(TypedDict):
-    """**BETA** Options for prefix text queries.
+    """Options for prefix string queries.
 
-    PrefixOpts is currently unstable API and subject to backwards breaking changes.
+    .. versionadded:: 4.15
     """
 
     # strMinQueryLength is the minimum allowed query length. Querying with a shorter string will error.
@@ -383,12 +383,32 @@ class PrefixOpts(TypedDict):
 
 
 class SuffixOpts(TypedDict):
-    """**BETA** Options for suffix text queries.
+    """Options for suffix string queries.
 
-    SuffixOpts is currently unstable API and subject to backwards breaking changes.
+    .. versionadded:: 4.15
     """
 
     # strMinQueryLength is the minimum allowed query length. Querying with a shorter string will error.
     strMinQueryLength: int
     # strMaxQueryLength is the maximum allowed query length. Querying with a longer string will error.
     strMaxQueryLength: int
+
+
+class TextOpts(StringOpts):
+    """**DEPRECATED** Options to configure encrypted queries using the text algorithm.
+
+    .. note:: ``TextOpts`` is deprecated. Use :class:`StringOpts` instead.
+
+    .. versionadded:: 4.15
+
+    .. versionchanged:: 4.18
+       Deprecated in favor of :class:`StringOpts`.
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        warnings.warn(
+            "TextOpts is deprecated. Use StringOpts instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(*args, **kwargs)
diff --git a/pymongo/synchronous/encryption.py b/pymongo/synchronous/encryption.py
index 014d162e2b..5516237146 100644
--- a/pymongo/synchronous/encryption.py
+++ b/pymongo/synchronous/encryption.py
@@ -21,6 +21,7 @@
 import socket
 import time as time  # noqa: PLC0414 # needed in sync version
 import uuid
+import warnings
 import weakref
 from collections.abc import Generator, Iterator, Mapping, MutableMapping, Sequence
 from copy import deepcopy
@@ -60,7 +61,10 @@
 from pymongo.encryption_options import (
     AutoEncryptionOpts,
     RangeOpts,
-    TextOpts,
+    StringOpts,
+    # Re-exported for backwards compatibility: TextOpts is deprecated but must
+    # remain importable from this module until it is removed.
+    TextOpts,  # noqa: F401
     check_min_pymongocrypt,
 )
 from pymongo.errors import (
@@ -526,8 +530,15 @@ class Algorithm(str, enum.Enum):
 
     .. versionadded:: 4.4
     """
+    STRING = "String"
+    """String.
+
+    .. versionadded:: 4.18
+    """
     TEXTPREVIEW = "TextPreview"
-    """**BETA** - TextPreview.
+    """**DEPRECATED** - TextPreview.
+
+    .. note:: Support for TextPreview is deprecated. Use :attr:`Algorithm.STRING` instead.
 
     .. versionadded:: 4.15
     """
@@ -556,25 +567,77 @@ class QueryType(str, enum.Enum):
     .. versionadded:: 4.4
     """
 
+    PREFIX = "prefix"
+    """Used to encrypt a value for a prefix query.
+
+    Used for the ``$encStrStartsWith`` operator. Requires MongoDB 9.0+.
+
+    .. versionadded:: 4.18
+    """
+
+    SUFFIX = "suffix"
+    """Used to encrypt a value for a suffix query.
+
+    Used for the ``$encStrEndsWith`` operator. Requires MongoDB 9.0+.
+
+    .. versionadded:: 4.18
+    """
+
+    SUBSTRING = "substring"
+    """Used to encrypt a value for a substring query.
+
+    Used for the ``$encStrContains`` operator. Requires MongoDB 9.0+.
+
+    .. versionadded:: 4.18
+    """
+
     PREFIXPREVIEW = "prefixPreview"
     """**BETA** - Used to encrypt a value for a prefixPreview query.
 
+    .. note:: The preview query types are for experimental workloads only and
+       are only supported by MongoDB versions before 9.0. Use
+       :attr:`QueryType.PREFIX` instead.
+
     .. versionadded:: 4.15
     """
 
     SUFFIXPREVIEW = "suffixPreview"
     """**BETA** - Used to encrypt a value for a suffixPreview query.
 
+    .. note:: The preview query types are for experimental workloads only and
+       are only supported by MongoDB versions before 9.0. Use
+       :attr:`QueryType.SUFFIX` instead.
+
     .. versionadded:: 4.15
     """
 
     SUBSTRINGPREVIEW = "substringPreview"
     """**BETA** - Used to encrypt a value for a substringPreview query.
 
+    .. note:: The preview query types are for experimental workloads only and
+       are only supported by MongoDB versions before 9.0. Use
+       :attr:`QueryType.SUBSTRING` instead.
+
     .. versionadded:: 4.15
     """
 
 
+def _resolve_string_opts(
+    string_opts: Optional[StringOpts], text_opts: Optional[StringOpts]
+) -> Optional[StringOpts]:
+    """Resolve the deprecated text_opts alias for string_opts."""
+    if text_opts is None:
+        return string_opts
+    if string_opts is not None:
+        raise ConfigurationError("Cannot set both string_opts and text_opts")
+    warnings.warn(
+        "The text_opts parameter is deprecated. Use string_opts instead.",
+        DeprecationWarning,
+        stacklevel=3,
+    )
+    return text_opts
+
+
 def _create_mongocrypt_options(**kwargs: Any) -> MongoCryptOptions:
     # For compat with pymongocrypt <1.13, avoid setting the default key_expiration_ms.
     if kwargs.get("key_expiration_ms") is None:
@@ -910,7 +973,7 @@ def _encrypt_helper(
         contention_factor: Optional[int] = None,
         range_opts: Optional[RangeOpts] = None,
         is_expression: bool = False,
-        text_opts: Optional[TextOpts] = None,
+        string_opts: Optional[StringOpts] = None,
     ) -> Any:
         self._check_closed()
         if isinstance(key_id, uuid.UUID):
@@ -930,10 +993,10 @@ def _encrypt_helper(
                 range_opts.document,
                 codec_options=self._codec_options,
             )
-        text_opts_bytes = None
-        if text_opts:
-            text_opts_bytes = encode(
-                text_opts.document,
+        string_opts_bytes = None
+        if string_opts:
+            string_opts_bytes = encode(
+                string_opts.document,
                 codec_options=self._codec_options,
             )
         with _wrap_encryption_errors():
@@ -946,8 +1009,9 @@ def _encrypt_helper(
                 contention_factor=contention_factor,
                 range_opts=range_opts_bytes,
                 is_expression=is_expression,
+                # pymongocrypt still names this parameter text_opts.
                 # For compatibility with pymongocrypt < 1.16:
-                **{"text_opts": text_opts_bytes} if text_opts_bytes else {},
+                **{"text_opts": string_opts_bytes} if string_opts_bytes else {},
             )
             return decode(encrypted_doc)["v"]
 
@@ -960,7 +1024,8 @@ def encrypt(
         query_type: Optional[str] = None,
         contention_factor: Optional[int] = None,
         range_opts: Optional[RangeOpts] = None,
-        text_opts: Optional[TextOpts] = None,
+        string_opts: Optional[StringOpts] = None,
+        text_opts: Optional[StringOpts] = None,
     ) -> Binary:
         """Encrypt a BSON value with a given key and algorithm.
 
@@ -981,11 +1046,15 @@ def encrypt(
             used.
         :param range_opts: Index options for `range` queries. See
             :class:`RangeOpts` for some valid options.
-        :param text_opts: Index options for `textPreview` queries. See
-            :class:`TextOpts` for some valid options.
+        :param string_opts: Index options for `prefix`, `suffix`, and
+            `substring` queries. See :class:`StringOpts` for some valid options.
+        :param text_opts: **DEPRECATED** - Alias for `string_opts`.
 
         :return: The encrypted value, a :class:`~bson.binary.Binary` with subtype 6.
 
+        .. versionchanged:: 4.18
+           Added the `string_opts` parameter and deprecated `text_opts`.
+
         .. versionchanged:: 4.9
            Added the `text_opts` parameter.
 
@@ -1009,7 +1078,7 @@ def encrypt(
                 contention_factor=contention_factor,
                 range_opts=range_opts,
                 is_expression=False,
-                text_opts=text_opts,
+                string_opts=_resolve_string_opts(string_opts, text_opts),
             ),
         )
 
diff --git a/test/asynchronous/test_encryption.py b/test/asynchronous/test_encryption.py
index daeb18607a..585eef8c83 100644
--- a/test/asynchronous/test_encryption.py
+++ b/test/asynchronous/test_encryption.py
@@ -63,7 +63,13 @@
 from pymongo.asynchronous.helpers import anext
 from pymongo.asynchronous.mongo_client import AsyncMongoClient
 from pymongo.cursor_shared import CursorType
-from pymongo.encryption_options import _HAVE_PYMONGOCRYPT, AutoEncryptionOpts, RangeOpts, TextOpts
+from pymongo.encryption_options import (
+    _HAVE_PYMONGOCRYPT,
+    AutoEncryptionOpts,
+    RangeOpts,
+    StringOpts,
+    TextOpts,
+)
 from pymongo.errors import (
     AutoReconnect,
     BulkWriteError,
@@ -104,6 +110,7 @@
     camel_to_snake_args,
     is_greenthread_patched,
 )
+from test.version import Version
 
 _IS_SYNC = False
 
@@ -229,6 +236,37 @@ async def test_kwargs(self):
         self.assertEqual(get_client_opts(client).auto_encryption_opts, opts)
 
 
+class TestStringOptsDeprecation(AsyncPyMongoTestCase):
+    def test_text_opts_is_still_re_exported(self):
+        # TextOpts is deprecated, not removed, so it must stay importable from
+        # the encryption module for the deprecation period.
+        self.assertIs(encryption.TextOpts, TextOpts)
+
+    def test_text_opts_is_deprecated(self):
+        with self.assertWarns(DeprecationWarning):
+            opts = TextOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        self.assertIsInstance(opts, StringOpts)
+        self.assertEqual(
+            StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10}).document,
+            opts.document,
+        )
+
+    def test_resolve_string_opts(self):
+        string_opts = StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        self.assertIsNone(encryption._resolve_string_opts(None, None))
+        self.assertIs(encryption._resolve_string_opts(string_opts, None), string_opts)
+
+    def test_resolve_string_opts_text_opts_is_deprecated(self):
+        string_opts = StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        with self.assertWarns(DeprecationWarning):
+            self.assertIs(encryption._resolve_string_opts(None, string_opts), string_opts)
+
+    def test_resolve_string_opts_rejects_both(self):
+        string_opts = StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        with self.assertRaises(ConfigurationError):
+            encryption._resolve_string_opts(string_opts, string_opts)
+
+
 class AsyncEncryptionIntegrationTest(AsyncIntegrationTest):
     """Base class for encryption integration tests."""
 
@@ -3312,13 +3350,44 @@ async def test_collection_name_collision(self):
         self.assertIsInstance(exc.exception.encrypted_fields["fields"][0]["keyId"], Binary)
 
 
-# https://github.com/mongodb/specifications/blob/master/source/client-side-encryption/tests/README.md#27-text-explicit-encryption
-@unittest.skip("PYTHON-5799 need to add support for the new query type")
-class TestExplicitTextEncryptionProse(AsyncEncryptionIntegrationTest):
+def _libmongocrypt_at_least(*version):
+    """Return True if the installed libmongocrypt is at least `version`."""
+    from pymongocrypt import libmongocrypt_version
+
+    return Version.from_string(libmongocrypt_version()) >= Version(*version)
+
+
+# The minimum libmongocrypt version required by each string query type, declared
+# in one place so the test gates and the changelog agree. Support landed per
+# query type rather than all at once (see the libmongocrypt changelog):
+#   1.18.1 - fixes caseSensitive/diacriticSensitive handling for "textPreview".
+#   1.19.0 - the "string" algorithm replaces "textPreview"; prefix and suffix go
+#            stable; prefixPreview and suffixPreview are removed.
+#   1.19.1 - prefixPreview and suffixPreview are restored.
+#   1.20.0 - substring goes stable.
+_STRING_QUERY_MIN_LIBMONGOCRYPT = {
+    "prefix": (1, 19, 0),
+    "suffix": (1, 19, 0),
+    "substring": (1, 20, 0),
+    "prefixPreview": (1, 18, 1),
+    "suffixPreview": (1, 18, 1),
+    "substringPreview": (1, 18, 1),
+}
+
+# prefixPreview and suffixPreview were removed in 1.19.0 and restored in 1.19.1,
+# so that one release is a hole rather than a floor.
+_PREVIEW_REMOVED_IN = (1, 19, 0)
+
+
+# https://github.com/mongodb/specifications/blob/master/source/client-side-encryption/tests/README.md#27-string-explicit-encryption
+class TestStringExplicitEncryptionProse(AsyncEncryptionIntegrationTest):
+    # The GA collections require server 9.0+, the preview collections require
+    # server pre-9.0. Setup encrypts with the "String" algorithm on 9.0+ and the
+    # deprecated "textPreview" algorithm on earlier servers, since "String" was
+    # only introduced in libmongocrypt 1.19.0.
     @async_client_context.require_no_standalone
     @async_client_context.require_version_min(8, 2, -1)
-    @async_client_context.require_version_max(8, 99, 99)
-    @async_client_context.require_libmongocrypt_min(1, 15, 1)
+    @async_client_context.require_libmongocrypt_min(1, 18, 1)
     @async_client_context.require_pymongocrypt_min(1, 16, 0)
     async def asyncSetUp(self):
         await super().asyncSetUp()
@@ -3338,210 +3407,270 @@ async def asyncSetUp(self):
             self.client,
             OPTS,
         )
-        # Create a MongoClient named encryptedClient with these AutoEncryptionOpts.
-        opts = AutoEncryptionOpts(
-            self.kms_providers,
-            "keyvault.datakeys",
-            bypass_query_analysis=True,
+        # Create a MongoClient named explicitEncryptedClient with these AutoEncryptionOpts.
+        self.client_encrypted = await self.async_rs_or_single_client(
+            auto_encryption_opts=AutoEncryptionOpts(
+                self.kms_providers,
+                "keyvault.datakeys",
+                bypass_query_analysis=True,
+            )
         )
-        self.client_encrypted = await self.async_rs_or_single_client(auto_encryption_opts=opts)
-
-        # Using QE CreateCollection() and Collection.Drop(), drop and create the following collections with majority write concern:
-        # db.prefix-suffix using the encryptedFields option set to the contents of encryptedFields-prefix-suffix.json.
-        db = self.client_encrypted.db
-        await db.drop_collection("prefix-suffix")
-        encrypted_fields = json_data("etc", "data", "encryptedFields-prefix-suffix.json")
-        await self.client_encryption.create_encrypted_collection(
-            db, "prefix-suffix", kms_provider="local", encrypted_fields=encrypted_fields
+        # Create a MongoClient named autoEncryptedClient with these AutoEncryptionOpts.
+        self.client_auto_encrypted = await self.async_rs_or_single_client(
+            auto_encryption_opts=AutoEncryptionOpts(
+                self.kms_providers,
+                "keyvault.datakeys",
+            )
         )
-        # db.substring using the encryptedFields option set to the contents of encryptedFields-substring.json.
-        await db.drop_collection("substring")
-        encrypted_fields = json_data("etc", "data", "encryptedFields-substring.json")
-        await self.client_encryption.create_encrypted_collection(
-            db, "substring", kms_provider="local", encrypted_fields=encrypted_fields
+
+        # The GA query types ("prefix", "suffix", "substring") require server
+        # 9.0+, which in turn dropped the preview query types.
+        self.is_ga = async_client_context.version.at_least(9, 0, -1)
+        # The "String" algorithm was added in libmongocrypt 1.19.0. Servers
+        # before 9.0 are tested against libmongocrypt 1.18.x, where the preview
+        # query types are only usable via the deprecated "textPreview"
+        # algorithm, so pick whichever the running combination supports.
+        self.algorithm = (
+            Algorithm.STRING if _libmongocrypt_at_least(1, 19, 0) else Algorithm.TEXTPREVIEW
         )
 
+        # Using QE CreateCollection() and Collection.Drop(), drop and create the
+        # collections with majority write concern.
+        db = self.client_encrypted.db
+        if self.is_ga:
+            collections = ["prefix-suffix", "prefix-suffix-ci-di", "substring", "substring-ci-di"]
+        else:
+            collections = ["prefix-suffix-preview", "substring-preview"]
+        for name in collections:
+            await db.drop_collection(name)
+            await self.client_encryption.create_encrypted_collection(
+                db,
+                name,
+                kms_provider="local",
+                encrypted_fields=json_data("etc", "data", f"encryptedFields-{name}.json"),
+            )
+
         # Use clientEncryption to encrypt the string "foobarbaz" with the following EncryptOpts.
-        text_opts = TextOpts(
-            case_sensitive=True,
-            diacritic_sensitive=True,
-            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
-            suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
-        )
         encrypted_value = await self.client_encryption.encrypt(
             "foobarbaz",
             key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
+            algorithm=self.algorithm,
             contention_factor=0,
-            text_opts=text_opts,
+            string_opts=StringOpts(
+                case_sensitive=True,
+                diacritic_sensitive=True,
+                prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+                suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+            ),
         )
-        # Use encryptedClient to insert the following document into db.prefix-suffix with majority write concern.
-        coll = self.client_encrypted.db["prefix-suffix"].with_options(
-            write_concern=WriteConcern(w="majority")
+        # Use explicitEncryptedClient to insert the following document into
+        # db.prefix-suffix (if created) and db.prefix-suffix-preview (if created)
+        # with majority write concern.
+        await self._insert(
+            "prefix-suffix" if self.is_ga else "prefix-suffix-preview",
+            {"_id": 0, "encryptedText": encrypted_value},
         )
-        await coll.insert_one({"_id": 0, "encryptedText": encrypted_value})
 
         # Use clientEncryption to encrypt the string "foobarbaz" with the following EncryptOpts.
-        text_opts = TextOpts(
-            case_sensitive=True,
-            diacritic_sensitive=True,
-            substring=dict(strMaxLength=10, strMaxQueryLength=10, strMinQueryLength=2),
-        )
         encrypted_value = await self.client_encryption.encrypt(
             "foobarbaz",
             key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
+            algorithm=self.algorithm,
             contention_factor=0,
-            text_opts=text_opts,
+            string_opts=StringOpts(
+                case_sensitive=True,
+                diacritic_sensitive=True,
+                substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
+            ),
         )
-        # Use encryptedClient to insert the following document into db.substring with majority write concern.
-        coll = self.client_encrypted.db["substring"].with_options(
-            write_concern=WriteConcern(w="majority")
+        # Use explicitEncryptedClient to insert the following document into
+        # db.substring (if created) and db.substring-preview (if created) with
+        # majority write concern.
+        await self._insert(
+            "substring" if self.is_ga else "substring-preview",
+            {"_id": 0, "encryptedText": encrypted_value},
+        )
+
+    async def _insert(self, collection, document, client=None):
+        """Insert a document with majority write concern."""
+        client = client or self.client_encrypted
+        coll = client.db[collection].with_options(write_concern=WriteConcern(w="majority"))
+        await coll.insert_one(document)
+
+    def _require_query_type(self, query_type):
+        """Skip unless the installed libmongocrypt supports `query_type`."""
+        required = _STRING_QUERY_MIN_LIBMONGOCRYPT[query_type]
+        if not _libmongocrypt_at_least(*required):
+            raise unittest.SkipTest(
+                f"queryType={query_type} requires libmongocrypt {'.'.join(map(str, required))}+"
+            )
+        if query_type in ("prefixPreview", "suffixPreview") and (
+            _libmongocrypt_at_least(*_PREVIEW_REMOVED_IN) and not _libmongocrypt_at_least(1, 19, 1)
+        ):
+            raise unittest.SkipTest(f"queryType={query_type} was removed in libmongocrypt 1.19.0")
+
+    def _params(self, kind):
+        """Return the (query_type, collection) pair to run a case against.
+
+        Each case runs against the GA query type on server 9.0+ and against the
+        preview query type on earlier servers, skipping when the installed
+        libmongocrypt is too old for the applicable variant.
+        """
+        base = "substring" if kind == "substring" else "prefix-suffix"
+        if self.is_ga:
+            query_type, collection = kind, base
+        else:
+            query_type, collection = f"{kind}Preview", f"{base}-preview"
+        self._require_query_type(query_type)
+        return query_type, collection
+
+    def _require_ga(self, *query_types):
+        """Skip a case that only applies to the GA query types.
+
+        Gates on each query type the case exercises, since substring support
+        landed in a later libmongocrypt than prefix and suffix.
+        """
+        if not self.is_ga:
+            raise unittest.SkipTest("requires server 9.0+")
+        for query_type in query_types:
+            self._require_query_type(query_type)
+
+    async def _encrypt(self, value, query_type=None, **string_opts):
+        return await self.client_encryption.encrypt(
+            value,
+            key_id=self.key1_id,
+            algorithm=self.algorithm,
+            query_type=query_type,
+            contention_factor=0,
+            string_opts=StringOpts(**string_opts),
         )
-        await coll.insert_one({"_id": 0, "encryptedText": encrypted_value})
+
+    async def _find(self, collection, filter):
+        value = await self.client_encrypted.db[collection].find_one(filter)
+        if value is not None:
+            value.pop("__safeContent__", None)
+        return value
 
     async def test_01_can_find_a_document_by_prefix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "foo" with the following EncryptOpts.
-        text_opts = TextOpts(
+        query_type, collection = self._params("prefix")
+        # Use clientEncryption.encrypt() to encrypt the string "foo".
+        encrypted_value = await self._encrypt(
+            "foo",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = await self.client_encryption.encrypt(
-            "foo",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.PREFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter.
-        value = await self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            collection,
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
         )
         # Assert the following document is returned.
-        expected = {"_id": 0, "encryptedText": "foobarbaz"}
-        value.pop("__safeContent__", None)
-        self.assertEqual(value, expected)
+        self.assertEqual(value, {"_id": 0, "encryptedText": "foobarbaz"})
 
     async def test_02_can_find_a_document_by_suffix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "baz" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("suffix")
+        # Use clientEncryption.encrypt() to encrypt the string "baz".
+        encrypted_value = await self._encrypt(
+            "baz",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = await self.client_encryption.encrypt(
-            "baz",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUFFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter:
-        value = await self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            collection,
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
         )
         # Assert the following document is returned.
-        expected = {"_id": 0, "encryptedText": "foobarbaz"}
-        value.pop("__safeContent__", None)
-        self.assertEqual(value, expected)
+        self.assertEqual(value, {"_id": 0, "encryptedText": "foobarbaz"})
 
     async def test_03_no_document_found_by_prefix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "baz" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("prefix")
+        # Use clientEncryption.encrypt() to encrypt the string "baz".
+        encrypted_value = await self._encrypt(
+            "baz",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = await self.client_encryption.encrypt(
-            "baz",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.PREFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter:
-        value = await self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            collection,
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
         )
         # Assert that no documents are returned.
         self.assertIsNone(value)
 
     async def test_04_no_document_found_by_suffix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "foo" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("suffix")
+        # Use clientEncryption.encrypt() to encrypt the string "foo".
+        encrypted_value = await self._encrypt(
+            "foo",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = await self.client_encryption.encrypt(
-            "foo",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUFFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter:
-        value = await self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            collection,
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
         )
         # Assert that no documents are returned.
         self.assertIsNone(value)
 
     async def test_05_can_find_a_document_by_substring(self):
-        # Use clientEncryption.encrypt() to encrypt the string "bar" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("substring")
+        # Use clientEncryption.encrypt() to encrypt the string "bar".
+        encrypted_value = await self._encrypt(
+            "bar",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
-            substring=dict(strMaxLength=10, strMaxQueryLength=10, strMinQueryLength=2),
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
         )
-        encrypted_value = await self.client_encryption.encrypt(
-            "bar",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUBSTRINGPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.substring collection with the following filter:
-        value = await self.client_encrypted.db["substring"].find_one(
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            collection,
             {
                 "$expr": {
                     "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
                 }
-            }
+            },
         )
-        # Assert the following document is returned:
-        expected = {"_id": 0, "encryptedText": "foobarbaz"}
-        value.pop("__safeContent__", None)
-        self.assertEqual(value, expected)
+        # Assert the following document is returned.
+        self.assertEqual(value, {"_id": 0, "encryptedText": "foobarbaz"})
 
     async def test_06_no_document_found_by_substring(self):
-        # Use clientEncryption.encrypt() to encrypt the string "qux" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("substring")
+        # Use clientEncryption.encrypt() to encrypt the string "qux".
+        encrypted_value = await self._encrypt(
+            "qux",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
-            substring=dict(strMaxLength=10, strMaxQueryLength=10, strMinQueryLength=2),
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
         )
-        encrypted_value = await self.client_encryption.encrypt(
-            "qux",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUBSTRINGPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.substring collection with the following filter:
-        value = await self.client_encrypted.db["substring"].find_one(
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            collection,
             {
                 "$expr": {
                     "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
                 }
-            }
+            },
         )
         # Assert that no documents are returned.
         self.assertIsNone(value)
@@ -3549,25 +3678,164 @@ async def test_06_no_document_found_by_substring(self):
     async def test_07_contentionFactor_is_required(self):
         from pymongocrypt.errors import MongoCryptError
 
-        # Use clientEncryption.encrypt() to encrypt the string "foo" with the following EncryptOpts:
-        text_opts = TextOpts(
-            case_sensitive=True,
-            diacritic_sensitive=True,
-            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
-        )
+        self._require_ga("prefix")
+        # Use clientEncryption.encrypt() to encrypt the string "foo" without contentionFactor.
         with self.assertRaises(EncryptionError) as ctx:
             await self.client_encryption.encrypt(
                 "foo",
                 key_id=self.key1_id,
-                algorithm=Algorithm.TEXTPREVIEW,
-                query_type=QueryType.PREFIXPREVIEW,
-                text_opts=text_opts,
+                algorithm=Algorithm.STRING,
+                query_type=QueryType.PREFIX,
+                string_opts=StringOpts(
+                    case_sensitive=True,
+                    diacritic_sensitive=True,
+                    prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+                ),
             )
-        # Expect an error from libmongocrypt with a message containing the string: "contention factor is required for textPreview algorithm".
+        # Expect an error from libmongocrypt with a message containing the
+        # string: "contention factor is required for string algorithm".
         self.assertIsInstance(ctx.exception.cause, MongoCryptError)
-        self.assertEqual(
-            str(ctx.exception), "contention factor is required for textPreview algorithm"
+        self.assertIn("contention factor is required for string algorithm", str(ctx.exception))
+
+    async def test_08_case_insensitive_prefix_and_suffix(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("prefix", "suffix")
+        # Use autoEncryptedClient to insert the following document.
+        await self._insert(
+            "prefix-suffix-ci-di", {"encryptedText": "BingQiLin"}, self.client_auto_encrypted
+        )
+        # Use clientEncryption.encrypt() to encrypt the string "bing".
+        encrypted_value = await self._encrypt(
+            "bing",
+            query_type=QueryType.PREFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            "prefix-suffix-ci-di",
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "BingQiLin")
+
+        # Use clientEncryption.encrypt() to encrypt the string "lin".
+        encrypted_value = await self._encrypt(
+            "lin",
+            query_type=QueryType.SUFFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            "prefix-suffix-ci-di",
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "BingQiLin")
+
+    async def test_09_diacritic_insensitive_prefix_and_suffix(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("prefix", "suffix")
+        # Use autoEncryptedClient to insert the following document.
+        await self._insert(
+            "prefix-suffix-ci-di", {"encryptedText": "cafébarbäz"}, self.client_auto_encrypted
+        )
+        # Use clientEncryption.encrypt() to encrypt the string "cafe".
+        encrypted_value = await self._encrypt(
+            "cafe",
+            query_type=QueryType.PREFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            "prefix-suffix-ci-di",
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "cafébarbäz")
+
+        # Use clientEncryption.encrypt() to encrypt the string "baz".
+        encrypted_value = await self._encrypt(
+            "baz",
+            query_type=QueryType.SUFFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            "prefix-suffix-ci-di",
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
         )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "cafébarbäz")
+
+    async def test_10_case_insensitive_substring(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("substring")
+        # Use autoEncryptedClient to insert the following document.
+        await self._insert(
+            "substring-ci-di", {"encryptedText": "FooBarBaz"}, self.client_auto_encrypted
+        )
+        # Use clientEncryption.encrypt() to encrypt the string "bar".
+        encrypted_value = await self._encrypt(
+            "bar",
+            query_type=QueryType.SUBSTRING,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            "substring-ci-di",
+            {
+                "$expr": {
+                    "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "FooBarBaz")
+
+    async def test_11_diacritic_insensitive_substring(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("substring")
+        # Use autoEncryptedClient to insert the following document.
+        await self._insert(
+            "substring-ci-di", {"encryptedText": "foocafébaz"}, self.client_auto_encrypted
+        )
+        # Use clientEncryption.encrypt() to encrypt the string "cafe".
+        encrypted_value = await self._encrypt(
+            "cafe",
+            query_type=QueryType.SUBSTRING,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = await self._find(
+            "substring-ci-di",
+            {
+                "$expr": {
+                    "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "foocafébaz")
 
 
 def start_mongocryptd(port) -> None:
diff --git a/test/test_encryption.py b/test/test_encryption.py
index 744db01b1b..1cbf388b24 100644
--- a/test/test_encryption.py
+++ b/test/test_encryption.py
@@ -59,7 +59,13 @@
 from bson.son import SON
 from pymongo import ReadPreference
 from pymongo.cursor_shared import CursorType
-from pymongo.encryption_options import _HAVE_PYMONGOCRYPT, AutoEncryptionOpts, RangeOpts, TextOpts
+from pymongo.encryption_options import (
+    _HAVE_PYMONGOCRYPT,
+    AutoEncryptionOpts,
+    RangeOpts,
+    StringOpts,
+    TextOpts,
+)
 from pymongo.errors import (
     AutoReconnect,
     BulkWriteError,
@@ -104,6 +110,7 @@
     is_greenthread_patched,
     wait_until,
 )
+from test.version import Version
 
 _IS_SYNC = True
 
@@ -229,6 +236,37 @@ def test_kwargs(self):
         self.assertEqual(get_client_opts(client).auto_encryption_opts, opts)
 
 
+class TestStringOptsDeprecation(PyMongoTestCase):
+    def test_text_opts_is_still_re_exported(self):
+        # TextOpts is deprecated, not removed, so it must stay importable from
+        # the encryption module for the deprecation period.
+        self.assertIs(encryption.TextOpts, TextOpts)
+
+    def test_text_opts_is_deprecated(self):
+        with self.assertWarns(DeprecationWarning):
+            opts = TextOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        self.assertIsInstance(opts, StringOpts)
+        self.assertEqual(
+            StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10}).document,
+            opts.document,
+        )
+
+    def test_resolve_string_opts(self):
+        string_opts = StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        self.assertIsNone(encryption._resolve_string_opts(None, None))
+        self.assertIs(encryption._resolve_string_opts(string_opts, None), string_opts)
+
+    def test_resolve_string_opts_text_opts_is_deprecated(self):
+        string_opts = StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        with self.assertWarns(DeprecationWarning):
+            self.assertIs(encryption._resolve_string_opts(None, string_opts), string_opts)
+
+    def test_resolve_string_opts_rejects_both(self):
+        string_opts = StringOpts(prefix={"strMinQueryLength": 2, "strMaxQueryLength": 10})
+        with self.assertRaises(ConfigurationError):
+            encryption._resolve_string_opts(string_opts, string_opts)
+
+
 class EncryptionIntegrationTest(IntegrationTest):
     """Base class for encryption integration tests."""
 
@@ -3294,13 +3332,44 @@ def test_collection_name_collision(self):
         self.assertIsInstance(exc.exception.encrypted_fields["fields"][0]["keyId"], Binary)
 
 
-# https://github.com/mongodb/specifications/blob/master/source/client-side-encryption/tests/README.md#27-text-explicit-encryption
-@unittest.skip("PYTHON-5799 need to add support for the new query type")
-class TestExplicitTextEncryptionProse(EncryptionIntegrationTest):
+def _libmongocrypt_at_least(*version):
+    """Return True if the installed libmongocrypt is at least `version`."""
+    from pymongocrypt import libmongocrypt_version
+
+    return Version.from_string(libmongocrypt_version()) >= Version(*version)
+
+
+# The minimum libmongocrypt version required by each string query type, declared
+# in one place so the test gates and the changelog agree. Support landed per
+# query type rather than all at once (see the libmongocrypt changelog):
+#   1.18.1 - fixes caseSensitive/diacriticSensitive handling for "textPreview".
+#   1.19.0 - the "string" algorithm replaces "textPreview"; prefix and suffix go
+#            stable; prefixPreview and suffixPreview are removed.
+#   1.19.1 - prefixPreview and suffixPreview are restored.
+#   1.20.0 - substring goes stable.
+_STRING_QUERY_MIN_LIBMONGOCRYPT = {
+    "prefix": (1, 19, 0),
+    "suffix": (1, 19, 0),
+    "substring": (1, 20, 0),
+    "prefixPreview": (1, 18, 1),
+    "suffixPreview": (1, 18, 1),
+    "substringPreview": (1, 18, 1),
+}
+
+# prefixPreview and suffixPreview were removed in 1.19.0 and restored in 1.19.1,
+# so that one release is a hole rather than a floor.
+_PREVIEW_REMOVED_IN = (1, 19, 0)
+
+
+# https://github.com/mongodb/specifications/blob/master/source/client-side-encryption/tests/README.md#27-string-explicit-encryption
+class TestStringExplicitEncryptionProse(EncryptionIntegrationTest):
+    # The GA collections require server 9.0+, the preview collections require
+    # server pre-9.0. Setup encrypts with the "String" algorithm on 9.0+ and the
+    # deprecated "textPreview" algorithm on earlier servers, since "String" was
+    # only introduced in libmongocrypt 1.19.0.
     @client_context.require_no_standalone
     @client_context.require_version_min(8, 2, -1)
-    @client_context.require_version_max(8, 99, 99)
-    @client_context.require_libmongocrypt_min(1, 15, 1)
+    @client_context.require_libmongocrypt_min(1, 18, 1)
     @client_context.require_pymongocrypt_min(1, 16, 0)
     def setUp(self):
         super().setUp()
@@ -3320,210 +3389,270 @@ def setUp(self):
             self.client,
             OPTS,
         )
-        # Create a MongoClient named encryptedClient with these AutoEncryptionOpts.
-        opts = AutoEncryptionOpts(
-            self.kms_providers,
-            "keyvault.datakeys",
-            bypass_query_analysis=True,
+        # Create a MongoClient named explicitEncryptedClient with these AutoEncryptionOpts.
+        self.client_encrypted = self.rs_or_single_client(
+            auto_encryption_opts=AutoEncryptionOpts(
+                self.kms_providers,
+                "keyvault.datakeys",
+                bypass_query_analysis=True,
+            )
         )
-        self.client_encrypted = self.rs_or_single_client(auto_encryption_opts=opts)
-
-        # Using QE CreateCollection() and Collection.Drop(), drop and create the following collections with majority write concern:
-        # db.prefix-suffix using the encryptedFields option set to the contents of encryptedFields-prefix-suffix.json.
-        db = self.client_encrypted.db
-        db.drop_collection("prefix-suffix")
-        encrypted_fields = json_data("etc", "data", "encryptedFields-prefix-suffix.json")
-        self.client_encryption.create_encrypted_collection(
-            db, "prefix-suffix", kms_provider="local", encrypted_fields=encrypted_fields
+        # Create a MongoClient named autoEncryptedClient with these AutoEncryptionOpts.
+        self.client_auto_encrypted = self.rs_or_single_client(
+            auto_encryption_opts=AutoEncryptionOpts(
+                self.kms_providers,
+                "keyvault.datakeys",
+            )
         )
-        # db.substring using the encryptedFields option set to the contents of encryptedFields-substring.json.
-        db.drop_collection("substring")
-        encrypted_fields = json_data("etc", "data", "encryptedFields-substring.json")
-        self.client_encryption.create_encrypted_collection(
-            db, "substring", kms_provider="local", encrypted_fields=encrypted_fields
+
+        # The GA query types ("prefix", "suffix", "substring") require server
+        # 9.0+, which in turn dropped the preview query types.
+        self.is_ga = client_context.version.at_least(9, 0, -1)
+        # The "String" algorithm was added in libmongocrypt 1.19.0. Servers
+        # before 9.0 are tested against libmongocrypt 1.18.x, where the preview
+        # query types are only usable via the deprecated "textPreview"
+        # algorithm, so pick whichever the running combination supports.
+        self.algorithm = (
+            Algorithm.STRING if _libmongocrypt_at_least(1, 19, 0) else Algorithm.TEXTPREVIEW
         )
 
+        # Using QE CreateCollection() and Collection.Drop(), drop and create the
+        # collections with majority write concern.
+        db = self.client_encrypted.db
+        if self.is_ga:
+            collections = ["prefix-suffix", "prefix-suffix-ci-di", "substring", "substring-ci-di"]
+        else:
+            collections = ["prefix-suffix-preview", "substring-preview"]
+        for name in collections:
+            db.drop_collection(name)
+            self.client_encryption.create_encrypted_collection(
+                db,
+                name,
+                kms_provider="local",
+                encrypted_fields=json_data("etc", "data", f"encryptedFields-{name}.json"),
+            )
+
         # Use clientEncryption to encrypt the string "foobarbaz" with the following EncryptOpts.
-        text_opts = TextOpts(
-            case_sensitive=True,
-            diacritic_sensitive=True,
-            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
-            suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
-        )
         encrypted_value = self.client_encryption.encrypt(
             "foobarbaz",
             key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
+            algorithm=self.algorithm,
             contention_factor=0,
-            text_opts=text_opts,
+            string_opts=StringOpts(
+                case_sensitive=True,
+                diacritic_sensitive=True,
+                prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+                suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+            ),
         )
-        # Use encryptedClient to insert the following document into db.prefix-suffix with majority write concern.
-        coll = self.client_encrypted.db["prefix-suffix"].with_options(
-            write_concern=WriteConcern(w="majority")
+        # Use explicitEncryptedClient to insert the following document into
+        # db.prefix-suffix (if created) and db.prefix-suffix-preview (if created)
+        # with majority write concern.
+        self._insert(
+            "prefix-suffix" if self.is_ga else "prefix-suffix-preview",
+            {"_id": 0, "encryptedText": encrypted_value},
         )
-        coll.insert_one({"_id": 0, "encryptedText": encrypted_value})
 
         # Use clientEncryption to encrypt the string "foobarbaz" with the following EncryptOpts.
-        text_opts = TextOpts(
-            case_sensitive=True,
-            diacritic_sensitive=True,
-            substring=dict(strMaxLength=10, strMaxQueryLength=10, strMinQueryLength=2),
-        )
         encrypted_value = self.client_encryption.encrypt(
             "foobarbaz",
             key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
+            algorithm=self.algorithm,
             contention_factor=0,
-            text_opts=text_opts,
+            string_opts=StringOpts(
+                case_sensitive=True,
+                diacritic_sensitive=True,
+                substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
+            ),
         )
-        # Use encryptedClient to insert the following document into db.substring with majority write concern.
-        coll = self.client_encrypted.db["substring"].with_options(
-            write_concern=WriteConcern(w="majority")
+        # Use explicitEncryptedClient to insert the following document into
+        # db.substring (if created) and db.substring-preview (if created) with
+        # majority write concern.
+        self._insert(
+            "substring" if self.is_ga else "substring-preview",
+            {"_id": 0, "encryptedText": encrypted_value},
+        )
+
+    def _insert(self, collection, document, client=None):
+        """Insert a document with majority write concern."""
+        client = client or self.client_encrypted
+        coll = client.db[collection].with_options(write_concern=WriteConcern(w="majority"))
+        coll.insert_one(document)
+
+    def _require_query_type(self, query_type):
+        """Skip unless the installed libmongocrypt supports `query_type`."""
+        required = _STRING_QUERY_MIN_LIBMONGOCRYPT[query_type]
+        if not _libmongocrypt_at_least(*required):
+            raise unittest.SkipTest(
+                f"queryType={query_type} requires libmongocrypt {'.'.join(map(str, required))}+"
+            )
+        if query_type in ("prefixPreview", "suffixPreview") and (
+            _libmongocrypt_at_least(*_PREVIEW_REMOVED_IN) and not _libmongocrypt_at_least(1, 19, 1)
+        ):
+            raise unittest.SkipTest(f"queryType={query_type} was removed in libmongocrypt 1.19.0")
+
+    def _params(self, kind):
+        """Return the (query_type, collection) pair to run a case against.
+
+        Each case runs against the GA query type on server 9.0+ and against the
+        preview query type on earlier servers, skipping when the installed
+        libmongocrypt is too old for the applicable variant.
+        """
+        base = "substring" if kind == "substring" else "prefix-suffix"
+        if self.is_ga:
+            query_type, collection = kind, base
+        else:
+            query_type, collection = f"{kind}Preview", f"{base}-preview"
+        self._require_query_type(query_type)
+        return query_type, collection
+
+    def _require_ga(self, *query_types):
+        """Skip a case that only applies to the GA query types.
+
+        Gates on each query type the case exercises, since substring support
+        landed in a later libmongocrypt than prefix and suffix.
+        """
+        if not self.is_ga:
+            raise unittest.SkipTest("requires server 9.0+")
+        for query_type in query_types:
+            self._require_query_type(query_type)
+
+    def _encrypt(self, value, query_type=None, **string_opts):
+        return self.client_encryption.encrypt(
+            value,
+            key_id=self.key1_id,
+            algorithm=self.algorithm,
+            query_type=query_type,
+            contention_factor=0,
+            string_opts=StringOpts(**string_opts),
         )
-        coll.insert_one({"_id": 0, "encryptedText": encrypted_value})
+
+    def _find(self, collection, filter):
+        value = self.client_encrypted.db[collection].find_one(filter)
+        if value is not None:
+            value.pop("__safeContent__", None)
+        return value
 
     def test_01_can_find_a_document_by_prefix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "foo" with the following EncryptOpts.
-        text_opts = TextOpts(
+        query_type, collection = self._params("prefix")
+        # Use clientEncryption.encrypt() to encrypt the string "foo".
+        encrypted_value = self._encrypt(
+            "foo",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = self.client_encryption.encrypt(
-            "foo",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.PREFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter.
-        value = self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            collection,
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
         )
         # Assert the following document is returned.
-        expected = {"_id": 0, "encryptedText": "foobarbaz"}
-        value.pop("__safeContent__", None)
-        self.assertEqual(value, expected)
+        self.assertEqual(value, {"_id": 0, "encryptedText": "foobarbaz"})
 
     def test_02_can_find_a_document_by_suffix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "baz" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("suffix")
+        # Use clientEncryption.encrypt() to encrypt the string "baz".
+        encrypted_value = self._encrypt(
+            "baz",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = self.client_encryption.encrypt(
-            "baz",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUFFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter:
-        value = self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            collection,
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
         )
         # Assert the following document is returned.
-        expected = {"_id": 0, "encryptedText": "foobarbaz"}
-        value.pop("__safeContent__", None)
-        self.assertEqual(value, expected)
+        self.assertEqual(value, {"_id": 0, "encryptedText": "foobarbaz"})
 
     def test_03_no_document_found_by_prefix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "baz" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("prefix")
+        # Use clientEncryption.encrypt() to encrypt the string "baz".
+        encrypted_value = self._encrypt(
+            "baz",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = self.client_encryption.encrypt(
-            "baz",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.PREFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter:
-        value = self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            collection,
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
         )
         # Assert that no documents are returned.
         self.assertIsNone(value)
 
     def test_04_no_document_found_by_suffix(self):
-        # Use clientEncryption.encrypt() to encrypt the string "foo" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("suffix")
+        # Use clientEncryption.encrypt() to encrypt the string "foo".
+        encrypted_value = self._encrypt(
+            "foo",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
             suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
-        encrypted_value = self.client_encryption.encrypt(
-            "foo",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUFFIXPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.prefix-suffix collection with the following filter:
-        value = self.client_encrypted.db["prefix-suffix"].find_one(
-            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}}
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            collection,
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
         )
         # Assert that no documents are returned.
         self.assertIsNone(value)
 
     def test_05_can_find_a_document_by_substring(self):
-        # Use clientEncryption.encrypt() to encrypt the string "bar" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("substring")
+        # Use clientEncryption.encrypt() to encrypt the string "bar".
+        encrypted_value = self._encrypt(
+            "bar",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
-            substring=dict(strMaxLength=10, strMaxQueryLength=10, strMinQueryLength=2),
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
         )
-        encrypted_value = self.client_encryption.encrypt(
-            "bar",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUBSTRINGPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.substring collection with the following filter:
-        value = self.client_encrypted.db["substring"].find_one(
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            collection,
             {
                 "$expr": {
                     "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
                 }
-            }
+            },
         )
-        # Assert the following document is returned:
-        expected = {"_id": 0, "encryptedText": "foobarbaz"}
-        value.pop("__safeContent__", None)
-        self.assertEqual(value, expected)
+        # Assert the following document is returned.
+        self.assertEqual(value, {"_id": 0, "encryptedText": "foobarbaz"})
 
     def test_06_no_document_found_by_substring(self):
-        # Use clientEncryption.encrypt() to encrypt the string "qux" with the following EncryptOpts:
-        text_opts = TextOpts(
+        query_type, collection = self._params("substring")
+        # Use clientEncryption.encrypt() to encrypt the string "qux".
+        encrypted_value = self._encrypt(
+            "qux",
+            query_type=query_type,
             case_sensitive=True,
             diacritic_sensitive=True,
-            substring=dict(strMaxLength=10, strMaxQueryLength=10, strMinQueryLength=2),
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
         )
-        encrypted_value = self.client_encryption.encrypt(
-            "qux",
-            key_id=self.key1_id,
-            algorithm=Algorithm.TEXTPREVIEW,
-            query_type=QueryType.SUBSTRINGPREVIEW,
-            contention_factor=0,
-            text_opts=text_opts,
-        )
-        # Use encryptedClient to run a "find" operation on the db.substring collection with the following filter:
-        value = self.client_encrypted.db["substring"].find_one(
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            collection,
             {
                 "$expr": {
                     "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
                 }
-            }
+            },
         )
         # Assert that no documents are returned.
         self.assertIsNone(value)
@@ -3531,25 +3660,160 @@ def test_06_no_document_found_by_substring(self):
     def test_07_contentionFactor_is_required(self):
         from pymongocrypt.errors import MongoCryptError
 
-        # Use clientEncryption.encrypt() to encrypt the string "foo" with the following EncryptOpts:
-        text_opts = TextOpts(
-            case_sensitive=True,
-            diacritic_sensitive=True,
-            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
-        )
+        self._require_ga("prefix")
+        # Use clientEncryption.encrypt() to encrypt the string "foo" without contentionFactor.
         with self.assertRaises(EncryptionError) as ctx:
             self.client_encryption.encrypt(
                 "foo",
                 key_id=self.key1_id,
-                algorithm=Algorithm.TEXTPREVIEW,
-                query_type=QueryType.PREFIXPREVIEW,
-                text_opts=text_opts,
+                algorithm=Algorithm.STRING,
+                query_type=QueryType.PREFIX,
+                string_opts=StringOpts(
+                    case_sensitive=True,
+                    diacritic_sensitive=True,
+                    prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+                ),
             )
-        # Expect an error from libmongocrypt with a message containing the string: "contention factor is required for textPreview algorithm".
+        # Expect an error from libmongocrypt with a message containing the
+        # string: "contention factor is required for string algorithm".
         self.assertIsInstance(ctx.exception.cause, MongoCryptError)
-        self.assertEqual(
-            str(ctx.exception), "contention factor is required for textPreview algorithm"
+        self.assertIn("contention factor is required for string algorithm", str(ctx.exception))
+
+    def test_08_case_insensitive_prefix_and_suffix(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("prefix", "suffix")
+        # Use autoEncryptedClient to insert the following document.
+        self._insert(
+            "prefix-suffix-ci-di", {"encryptedText": "BingQiLin"}, self.client_auto_encrypted
+        )
+        # Use clientEncryption.encrypt() to encrypt the string "bing".
+        encrypted_value = self._encrypt(
+            "bing",
+            query_type=QueryType.PREFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
         )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            "prefix-suffix-ci-di",
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "BingQiLin")
+
+        # Use clientEncryption.encrypt() to encrypt the string "lin".
+        encrypted_value = self._encrypt(
+            "lin",
+            query_type=QueryType.SUFFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            "prefix-suffix-ci-di",
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "BingQiLin")
+
+    def test_09_diacritic_insensitive_prefix_and_suffix(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("prefix", "suffix")
+        # Use autoEncryptedClient to insert the following document.
+        self._insert(
+            "prefix-suffix-ci-di", {"encryptedText": "cafébarbäz"}, self.client_auto_encrypted
+        )
+        # Use clientEncryption.encrypt() to encrypt the string "cafe".
+        encrypted_value = self._encrypt(
+            "cafe",
+            query_type=QueryType.PREFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            prefix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            "prefix-suffix-ci-di",
+            {
+                "$expr": {
+                    "$encStrStartsWith": {"input": "$encryptedText", "prefix": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "cafébarbäz")
+
+        # Use clientEncryption.encrypt() to encrypt the string "baz".
+        encrypted_value = self._encrypt(
+            "baz",
+            query_type=QueryType.SUFFIX,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            suffix=dict(strMaxQueryLength=10, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            "prefix-suffix-ci-di",
+            {"$expr": {"$encStrEndsWith": {"input": "$encryptedText", "suffix": encrypted_value}}},
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "cafébarbäz")
+
+    def test_10_case_insensitive_substring(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("substring")
+        # Use autoEncryptedClient to insert the following document.
+        self._insert("substring-ci-di", {"encryptedText": "FooBarBaz"}, self.client_auto_encrypted)
+        # Use clientEncryption.encrypt() to encrypt the string "bar".
+        encrypted_value = self._encrypt(
+            "bar",
+            query_type=QueryType.SUBSTRING,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            "substring-ci-di",
+            {
+                "$expr": {
+                    "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "FooBarBaz")
+
+    def test_11_diacritic_insensitive_substring(self):
+        # This is a regression test for DRIVERS-3470.
+        self._require_ga("substring")
+        # Use autoEncryptedClient to insert the following document.
+        self._insert("substring-ci-di", {"encryptedText": "foocafébaz"}, self.client_auto_encrypted)
+        # Use clientEncryption.encrypt() to encrypt the string "cafe".
+        encrypted_value = self._encrypt(
+            "cafe",
+            query_type=QueryType.SUBSTRING,
+            case_sensitive=False,
+            diacritic_sensitive=False,
+            substring=dict(strMaxLength=10, strMaxQueryLength=6, strMinQueryLength=2),
+        )
+        # Use explicitEncryptedClient to run a "find" operation.
+        value = self._find(
+            "substring-ci-di",
+            {
+                "$expr": {
+                    "$encStrContains": {"input": "$encryptedText", "substring": encrypted_value}
+                }
+            },
+        )
+        # Assert the following document is returned.
+        self.assertEqual(value["encryptedText"], "foocafébaz")
 
 
 def start_mongocryptd(port) -> None: