Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions neo4j/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -310,6 +310,23 @@ def verify_connectivity(self, **config):
"""
raise NotImplementedError

def supports_multi_db(self):
""" Check if the server or cluster supports multi-databases.
:return: Returns true if the server or cluster the driver connects to supports multi-databases, otherwise false.
:rtype: bool
"""
from neo4j.io._bolt4x0 import Bolt4x0

multi_database = False
cx = self._pool.acquire()

if cx.PROTOCOL_VERSION >= Bolt4x0.PROTOCOL_VERSION and cx.server_info.version_info() >= Version(4, 0, 0):
multi_database = True

self._pool.release(cx)

return multi_database


class BoltDriver(Direct, Driver):
""" A :class:`.BoltDriver` is created from a ``bolt`` URI and addresses
Expand Down
26 changes: 25 additions & 1 deletion tests/integration/test_bolt_driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,11 @@

import pytest

from neo4j import GraphDatabase
from neo4j import (
GraphDatabase,
BoltDriver,
Version,
)
from neo4j.exceptions import (
ServiceUnavailable,
AuthError,
Expand Down Expand Up @@ -115,3 +119,23 @@ def test_should_fail_on_incorrect_password(bolt_uri):
except ServiceUnavailable as error:
if isinstance(error.__cause__, BoltHandshakeError):
pytest.skip(error.args[0])


def test_supports_multi_db(bolt_uri, auth):
# python -m pytest tests/integration/test_bolt_driver.py -s -v -k test_supports_multi_db
driver = GraphDatabase.driver(bolt_uri, auth=auth)
assert isinstance(driver, BoltDriver)

with driver.session() as session:
result = session.run("RETURN 1")
value = result.single().value() # Consumes the result
summary = result.summary()
server_info = summary.server

result = driver.supports_multi_db()
driver.close()

if server_info.version_info() >= Version(4, 0, 0) and server_info.protocol_version >= Version(4, 0):
assert result is True
else:
assert result is False
33 changes: 32 additions & 1 deletion tests/integration/test_neo4j_driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,11 @@

import pytest

from neo4j import GraphDatabase
from neo4j import (
GraphDatabase,
Neo4jDriver,
Version,
)
from neo4j.exceptions import (
ServiceUnavailable,
)
Expand All @@ -45,3 +49,30 @@ def test_neo4j_uri(neo4j_uri, auth, target):
pytest.skip(error.args[0])
elif isinstance(error.__cause__, BoltHandshakeError):
pytest.skip(error.args[0])


def test_supports_multi_db(neo4j_uri, auth, target):
# python -m pytest tests/integration/test_neo4j_driver.py -s -v -k test_supports_multi_db
try:
driver = GraphDatabase.driver(neo4j_uri, auth=auth)
assert isinstance(driver, Neo4jDriver)
except ServiceUnavailable as error:
if error.args[0] == "Server does not support routing":
# This is because a single instance Neo4j 3.5 does not have dbms.routing.cluster.getRoutingTable() call
pytest.skip(error.args[0])
elif isinstance(error.__cause__, BoltHandshakeError):
pytest.skip(error.args[0])

with driver.session() as session:
result = session.run("RETURN 1")
value = result.single().value() # Consumes the result
summary = result.summary()
server_info = summary.server

result = driver.supports_multi_db()
driver.close()

if server_info.version_info() >= Version(4, 0, 0) and server_info.protocol_version >= Version(4, 0):
assert result is True
else:
assert result is False