Bolt Handshake Manifest v1 (#1102)

* Refactor bolt socket
  Move handshake and connect method that can be automatically un-async-ed for
  less code duplication.
* minor tox config improvements: enable warning as error on py3.13, too
* Drop (undocumented) support for Bolt 4.1
* Tests: fix event loop shutdown for Python <= 3.8

Author: robsdedude

CHANGELOG.md

@@ -5,6 +5,7 @@ See also https://github.com/neo4j/neo4j-python-driver/wiki for a full changelog.
 ## NEXT RELEASE
 - Since the types of `Relationship`s are tied to the `Graph` object they belong to, fixing `pickle` support for graph types means that `Relationship`s with the same name will have a different type after `deepcopy`ing or pickling and unpickling them or their graph.
   For more details, see https://github.com/neo4j/neo4j-python-driver/pull/1133
+- Drop undocumented support for Bolt protocol versions 4.1.
 
 
 ## Version 5.27

pyproject.toml

@@ -122,7 +122,7 @@ use_parentheses = true
 
 [tool.pytest.ini_options]
 mock_use_standalone_module = true
-asyncio_mode = "auto"
+asyncio_mode = "strict"
 
 
 [tool.mypy]

src/neo4j/_async/io/__init__.py

@@ -32,6 +32,12 @@
 ]
 
 
+# [bolt-version-bump] search tag when changing bolt version support
+from . import (  # noqa - imports needed to register protocol handlers
+    _bolt3,
+    _bolt4,
+    _bolt5,
+)
 from ._bolt import AsyncBolt
 from ._common import (
     check_supported_server_product,

src/neo4j/_async/io/_bolt.py

@@ -23,7 +23,6 @@
 from logging import getLogger
 from time import monotonic
 
-from ..._async_compat.network import AsyncBoltSocket
 from ..._async_compat.util import AsyncUtil
 from ..._auth_management import to_auth_dict
 from ..._codec.hydration import (
@@ -51,6 +50,7 @@
     SessionExpired,
 )
 from ..config import AsyncPoolConfig
+from ._bolt_socket import AsyncBoltSocket
 from ._common import (
     AsyncInbox,
     AsyncOutbox,
@@ -59,6 +59,8 @@
 
 
 if t.TYPE_CHECKING:
+    import typing_extensions as te
+
     from ..._api import TelemetryAPI
 
 
@@ -134,6 +136,8 @@ class AsyncBolt:
     # results for it.
     most_recent_qid = None
 
+    SKIP_REGISTRATION = False
+
     def __init__(
         self,
         unresolved_address,
@@ -257,101 +261,36 @@ def assert_notification_filtering_support(self):
                 f"{self.server_info.agent!r}"
             )
 
-    # [bolt-version-bump] search tag when changing bolt version support
-    @classmethod
-    def protocol_handlers(cls, protocol_version=None):
-        """
-        Return a dictionary of available Bolt protocol handlers.
-
-        The handlers are keyed by version tuple. If an explicit protocol
-        version is provided, the dictionary will contain either zero or one
-        items, depending on whether that version is supported. If no protocol
-        version is provided, all available versions will be returned.
-
-        :param protocol_version: tuple identifying a specific protocol
-            version (e.g. (3, 5)) or None
-        :returns: dictionary of version tuple to handler class for all
-            relevant and supported protocol versions
-        :raise TypeError: if protocol version is not passed in a tuple
-        """
-        # Carry out Bolt subclass imports locally to avoid circular dependency
-        # issues.
-        from ._bolt3 import AsyncBolt3
-        from ._bolt4 import (
-            AsyncBolt4x1,
-            AsyncBolt4x2,
-            AsyncBolt4x3,
-            AsyncBolt4x4,
-        )
-        from ._bolt5 import (
-            AsyncBolt5x0,
-            AsyncBolt5x1,
-            AsyncBolt5x2,
-            AsyncBolt5x3,
-            AsyncBolt5x4,
-            AsyncBolt5x5,
-            AsyncBolt5x6,
-            AsyncBolt5x7,
-            AsyncBolt5x8,
-        )
-
-        handlers = {
-            AsyncBolt3.PROTOCOL_VERSION: AsyncBolt3,
-            # 4.0 unsupported because no space left in the handshake
-            AsyncBolt4x1.PROTOCOL_VERSION: AsyncBolt4x1,
-            AsyncBolt4x2.PROTOCOL_VERSION: AsyncBolt4x2,
-            AsyncBolt4x3.PROTOCOL_VERSION: AsyncBolt4x3,
-            AsyncBolt4x4.PROTOCOL_VERSION: AsyncBolt4x4,
-            AsyncBolt5x0.PROTOCOL_VERSION: AsyncBolt5x0,
-            AsyncBolt5x1.PROTOCOL_VERSION: AsyncBolt5x1,
-            AsyncBolt5x2.PROTOCOL_VERSION: AsyncBolt5x2,
-            AsyncBolt5x3.PROTOCOL_VERSION: AsyncBolt5x3,
-            AsyncBolt5x4.PROTOCOL_VERSION: AsyncBolt5x4,
-            AsyncBolt5x5.PROTOCOL_VERSION: AsyncBolt5x5,
-            AsyncBolt5x6.PROTOCOL_VERSION: AsyncBolt5x6,
-            AsyncBolt5x7.PROTOCOL_VERSION: AsyncBolt5x7,
-            AsyncBolt5x8.PROTOCOL_VERSION: AsyncBolt5x8,
-        }
+    protocol_handlers: t.ClassVar[dict[Version, type[AsyncBolt]]] = {}
 
+    def __init_subclass__(cls: type[te.Self], **kwargs: t.Any) -> None:
+        if cls.SKIP_REGISTRATION:
+            super().__init_subclass__(**kwargs)
+            return
+        protocol_version = cls.PROTOCOL_VERSION
         if protocol_version is None:
-            return handlers
-
-        if not isinstance(protocol_version, tuple):
-            raise TypeError("Protocol version must be specified as a tuple")
-
-        if protocol_version in handlers:
-            return {protocol_version: handlers[protocol_version]}
-
-        return {}
-
-    @classmethod
-    def version_list(cls, versions):
-        """
-        Return a list of supported protocol versions in order of preference.
-
-        The number of protocol versions (or ranges) returned is limited to 4.
-        """
-        # In fact, 4.3 is the fist version to support ranges. However, the
-        # range support got backported to 4.2. But even if the server is too
-        # old to have the backport, negotiating BOLT 4.1 is no problem as it's
-        # equivalent to 4.2
-        first_with_range_support = Version(4, 2)
-        result = []
-        for version in versions:
-            if (
-                result
-                and version >= first_with_range_support
-                and result[-1][0] == version[0]
-                and result[-1][1][1] == version[1] + 1
-            ):
-                # can use range to encompass this version
-                result[-1][1][1] = version[1]
-                continue
-            result.append(Version(version[0], [version[1], version[1]]))
-            if len(result) == 4:
-                break
-        return result
+            raise ValueError(
+                "AsyncBolt subclasses must define PROTOCOL_VERSION"
+            )
+        if not (
+            isinstance(protocol_version, Version)
+            and len(protocol_version) == 2
+            and all(isinstance(i, int) for i in protocol_version)
+        ):
+            raise TypeError(
+                "PROTOCOL_VERSION must be a 2-tuple of integers, not "
+                f"{protocol_version!r}"
+            )
+        if protocol_version in AsyncBolt.protocol_handlers:
+            cls_conflict = AsyncBolt.protocol_handlers[protocol_version]
+            raise TypeError(
+                f"Multiple classes for the same protocol version "
+                f"{protocol_version}: {cls}, {cls_conflict}"
+            )
+        cls.protocol_handlers[protocol_version] = cls
+        super().__init_subclass__(**kwargs)
 
+    # [bolt-version-bump] search tag when changing bolt version support
     @classmethod
     def get_handshake(cls):
         """
@@ -360,12 +299,9 @@ def get_handshake(cls):
         The length is 16 bytes as specified in the Bolt version negotiation.
         :returns: bytes
         """
-        supported_versions = sorted(
-            cls.protocol_handlers().keys(), reverse=True
+        return (
+            b"\x00\x00\x01\xff\x00\x08\x08\x05\x00\x02\x04\x04\x00\x00\x00\x03"
         )
-        offered_versions = cls.version_list(supported_versions)
-        versions_bytes = (v.to_bytes() for v in offered_versions)
-        return b"".join(versions_bytes).ljust(16, b"\x00")
 
     @classmethod
     async def ping(cls, address, *, deadline=None, pool_config=None):
@@ -400,7 +336,6 @@ async def ping(cls, address, *, deadline=None, pool_config=None):
             await AsyncBoltSocket.close_socket(s)
             return protocol_version
 
-    # [bolt-version-bump] search tag when changing bolt version support
     @classmethod
     async def open(
         cls,
@@ -441,71 +376,17 @@ async def open(
         )
 
         pool_config.protocol_version = protocol_version
-
-        # Carry out Bolt subclass imports locally to avoid circular dependency
-        # issues.
-
-        # avoid new lines after imports for better readability and conciseness
-        # fmt: off
-        if protocol_version == (5, 8):
-            from ._bolt5 import AsyncBolt5x8
-            bolt_cls = AsyncBolt5x8
-        elif protocol_version == (5, 7):
-            from ._bolt5 import AsyncBolt5x7
-            bolt_cls = AsyncBolt5x7
-        elif protocol_version == (5, 6):
-            from ._bolt5 import AsyncBolt5x6
-            bolt_cls = AsyncBolt5x6
-        elif protocol_version == (5, 5):
-            from ._bolt5 import AsyncBolt5x5
-            bolt_cls = AsyncBolt5x5
-        elif protocol_version == (5, 4):
-            from ._bolt5 import AsyncBolt5x4
-            bolt_cls = AsyncBolt5x4
-        elif protocol_version == (5, 3):
-            from ._bolt5 import AsyncBolt5x3
-            bolt_cls = AsyncBolt5x3
-        elif protocol_version == (5, 2):
-            from ._bolt5 import AsyncBolt5x2
-            bolt_cls = AsyncBolt5x2
-        elif protocol_version == (5, 1):
-            from ._bolt5 import AsyncBolt5x1
-            bolt_cls = AsyncBolt5x1
-        elif protocol_version == (5, 0):
-            from ._bolt5 import AsyncBolt5x0
-            bolt_cls = AsyncBolt5x0
-        elif protocol_version == (4, 4):
-            from ._bolt4 import AsyncBolt4x4
-            bolt_cls = AsyncBolt4x4
-        elif protocol_version == (4, 3):
-            from ._bolt4 import AsyncBolt4x3
-            bolt_cls = AsyncBolt4x3
-        elif protocol_version == (4, 2):
-            from ._bolt4 import AsyncBolt4x2
-            bolt_cls = AsyncBolt4x2
-        elif protocol_version == (4, 1):
-            from ._bolt4 import AsyncBolt4x1
-            bolt_cls = AsyncBolt4x1
-        # Implementation for 4.0 exists, but there was no space left in the
-        # handshake to offer this version to the server. Hence, the server
-        # should never request us to speak bolt 4.0.
-        # elif protocol_version == (4, 0):
-        #     from ._bolt4 import AsyncBolt4x0
-        #     bolt_cls = AsyncBolt4x0
-        elif protocol_version == (3, 0):
-            from ._bolt3 import AsyncBolt3
-            bolt_cls = AsyncBolt3
-        # fmt: on
-        else:
+        protocol_handlers = AsyncBolt.protocol_handlers
+        bolt_cls = protocol_handlers.get(protocol_version)
+        if bolt_cls is None:
             log.debug("[#%04X]  C: <CLOSE>", s.getsockname()[1])
             await AsyncBoltSocket.close_socket(s)
 
-            supported_versions = cls.protocol_handlers().keys()
             # TODO: 6.0 - raise public DriverError subclass instead
             raise BoltHandshakeError(
                 "The neo4j server does not support communication with this "
                 "driver. This driver has support for Bolt protocols "
-                f"{tuple(map(str, supported_versions))}.",
+                f"{tuple(map(str, AsyncBolt.protocol_handlers.keys()))}.",
                 address=address,
                 request_data=handshake,
                 response_data=data,

src/neo4j/_async/io/_bolt4.py

@@ -74,6 +74,8 @@ class AsyncBolt4x0(AsyncBolt):
 
     supports_notification_filtering = False
 
+    SKIP_REGISTRATION = True
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._server_state_manager = ServerStateManager(
@@ -550,6 +552,8 @@ class AsyncBolt4x2(AsyncBolt4x1):
 
     PROTOCOL_VERSION = Version(4, 2)
 
+    SKIP_REGISTRATION = False
+
 
 class AsyncBolt4x3(AsyncBolt4x2):
     """