4.0 error handling improvements (#366)

* added new errors

ResultConsumedError

TransactionNestingError

* added unit tests for exceptions

testing ServiceUnavailable and Nesting BoltProtocolError into a
ServiceUnavailable error.

* added exceptions and unit tests

ConfigurationError
AuthConfigurationError
CertificateConfigurationError

* updated error propagation

The BoltHandshakeError is wrapped in a ServiceUnavailable error.

Removed exposure of ServiceUnavailable in neo4j/__init__.py

* raises BoltHandshake error instead of ServiceUnavailable

* fixed integration test code

* improved speed for stub tests

* improved speed and fixed http error with xfail

Author: martin-neotech

neo4j/__init__.py

@@ -73,7 +73,7 @@
     SessionConfig,
     unit_of_work,
 )
-from neo4j.exceptions import ServiceUnavailable
+
 
 log = getLogger("neo4j")
 
@@ -94,12 +94,10 @@ def driver(cls, uri, *, auth=None, acquire_timeout=None, **config):
         """
         parsed = urlparse(uri)
         if parsed.scheme == "bolt":
-            return cls.bolt_driver(parsed.netloc, auth=auth, acquire_timeout=acquire_timeout,
-                                   **config)
+            return cls.bolt_driver(parsed.netloc, auth=auth, acquire_timeout=acquire_timeout, **config)
         elif parsed.scheme == "neo4j" or parsed.scheme == "bolt+routing":
             rc = cls._parse_routing_context(parsed.query)
-            return cls.neo4j_driver(parsed.netloc, auth=auth, routing_context=rc,
-                                    acquire_timeout=acquire_timeout, **config)
+            return cls.neo4j_driver(parsed.netloc, auth=auth, routing_context=rc, acquire_timeout=acquire_timeout, **config)
         else:
             raise ValueError("Unknown URI scheme {!r}".format(parsed.scheme))
 
@@ -108,16 +106,27 @@ def bolt_driver(cls, target, *, auth=None, acquire_timeout=None, **config):
         """ Create a driver for direct Bolt server access that uses
         socket I/O and thread-based concurrency.
         """
-        return BoltDriver.open(target, auth=auth, acquire_timeout=acquire_timeout, **config)
+        from neo4j._exceptions import BoltHandshakeError
+
+        try:
+            return BoltDriver.open(target, auth=auth, acquire_timeout=acquire_timeout, **config)
+        except BoltHandshakeError as error:
+            from neo4j.exceptions import ServiceUnavailable
+            raise ServiceUnavailable(str(error)) from error
 
     @classmethod
     def neo4j_driver(cls, *targets, auth=None, routing_context=None, acquire_timeout=None,
                      **config):
         """ Create a driver for routing-capable Neo4j service access
         that uses socket I/O and thread-based concurrency.
         """
-        return Neo4jDriver.open(*targets, auth=auth, routing_context=routing_context,
-                                acquire_timeout=acquire_timeout, **config)
+        from neo4j._exceptions import BoltHandshakeError
+
+        try:
+            return Neo4jDriver.open(*targets, auth=auth, routing_context=routing_context, acquire_timeout=acquire_timeout, **config)
+        except BoltHandshakeError as error:
+            from neo4j.exceptions import ServiceUnavailable
+            raise ServiceUnavailable(str(error)) from error
 
     @classmethod
     def _parse_routing_context(cls, query):
@@ -340,14 +349,23 @@ def get_routing_table(self):
         return self._pool.routing_table
 
     def verify_connectivity(self, **config):
+        """
+        :raise ServiceUnavailable: raised if the server does not support routing or if routing support is broken.
+        """
         # TODO: Improve and update Stub Test Server to be able to test.
         return self._verify_routing_connectivity()
 
     def _verify_routing_connectivity(self):
+        from neo4j.exceptions import ServiceUnavailable
+        from neo4j._exceptions import BoltHandshakeError
+
         table = self.get_routing_table()
         routing_info = {}
         for ix in list(table.routers):
-            routing_info[ix] = self._pool.fetch_routing_info(table.routers[0])
+            try:
+                routing_info[ix] = self._pool.fetch_routing_info(table.routers[0])
+            except BoltHandshakeError as error:
+                routing_info[ix] = None
 
         for key, val in routing_info.items():
             if val is not None:

neo4j/exceptions.py

@@ -39,6 +39,7 @@
 
 + DriverError
   + TransactionError
+    + TransactionNestingError
   + SessionExpired
   + ServiceUnavailable
     + RoutingServiceUnavailable
@@ -47,7 +48,7 @@
   + ConfigurationError
     + AuthConfigurationError
     + CertificateConfigurationError
-
+  + ResultConsumedError
 
 Connector API Errors
 ====================
@@ -65,6 +66,10 @@
 
 """
 
+CLASSIFICATION_CLIENT = "ClientError"
+CLASSIFICATION_TRANSIENT = "TransientError"
+CLASSIFICATION_DATABASE = "DatabaseError"
+
 
 class Neo4jError(Exception):
     """ Raised when the Cypher engine returns an error to the client.
@@ -79,12 +84,12 @@ class Neo4jError(Exception):
 
     @classmethod
     def hydrate(cls, message=None, code=None, **metadata):
-        message = message or "An unknown error occurred."
+        message = message or "An unknown error occurred"
         code = code or "Neo.DatabaseError.General.UnknownError"
         try:
             _, classification, category, title = code.split(".")
         except ValueError:
-            classification = "DatabaseError"
+            classification = CLASSIFICATION_DATABASE
             category = "General"
             title = "UnknownError"
 
@@ -101,19 +106,19 @@ def hydrate(cls, message=None, code=None, **metadata):
 
     @classmethod
     def _extract_error_class(cls, classification, code):
-        if classification == "ClientError":
+        if classification == CLASSIFICATION_CLIENT:
             try:
                 return client_errors[code]
             except KeyError:
                 return ClientError
 
-        elif classification == "TransientError":
+        elif classification == CLASSIFICATION_TRANSIENT:
             try:
                 return transient_errors[code]
             except KeyError:
                 return TransientError
 
-        elif classification == "DatabaseError":
+        elif classification == CLASSIFICATION_DATABASE:
             return DatabaseError
 
         else:
@@ -238,6 +243,15 @@ def __init__(self, transaction, *args, **kwargs):
         self.transaction = transaction
 
 
+class TransactionNestingError(DriverError):
+    """ Raised when transactions are nested incorrectly.
+    """
+
+    def __init__(self, transaction, *args, **kwargs):
+        super(TransactionError, self).__init__(*args, **kwargs)
+        self.transaction = transaction
+
+
 class ServiceUnavailable(DriverError):
     """ Raised when no database service is available.
     """
@@ -256,3 +270,23 @@ class WriteServiceUnavailable(ServiceUnavailable):
 class ReadServiceUnavailable(ServiceUnavailable):
     """ Raised when no read service is available.
     """
+
+
+class ResultConsumedError(DriverError):
+    """ Raised when trying to access records after the records have been consumed.
+    """
+
+
+class ConfigurationError(DriverError):
+    """ Raised when there is an error concerning a configuration.
+    """
+
+
+class AuthConfigurationError(ConfigurationError):
+    """ Raised when there is an error with the authentication configuration.
+    """
+
+
+class CertificateConfigurationError(ConfigurationError):
+    """ Raised when there is an error with the authentication configuration.
+    """

neo4j/io/__init__.py

@@ -154,6 +154,8 @@ def ping(cls, address, *, timeout=None, **config):
             s, protocol_version = connect(address, timeout=timeout, config=config)
         except ServiceUnavailable:
             return None
+        except BoltHandshakeError as e:
+            return None
         else:
             s.close()
             return protocol_version
@@ -167,6 +169,8 @@ def open(cls, address, *, auth=None, timeout=None, **config):
         :param timeout:
         :param config:
         :return:
+        :raise BoltHandshakeError: raised if the Bolt Protocol can not negotiate a protocol version.
+        :raise ServiceUnavailable: raised if there was a connection issue.
         """
         config = PoolConfig.consume(config)
         s, config.protocol_version, handshake, data = connect(address, timeout=timeout, config=config)
@@ -839,8 +843,7 @@ def _handshake(s, resolved_address):
         # response, the server has closed the connection
         log.debug("[#%04X]  S: <CLOSE>", local_port)
         s.close()
-        raise ServiceUnavailable("Connection to %r closed without handshake "
-                                 "response" % (resolved_address,))
+        raise BoltHandshakeError("Connection to {address} closed without handshake response".format(address=resolved_address), address=resolved_address, request_data=handshake, response_data=None)
     if data_size != 4:
         # Some garbled data has been received
         log.debug("[#%04X]  S: @*#!", local_port)

tests/integration/conftest.py

@@ -24,14 +24,18 @@
 from os.path import dirname, join
 from threading import RLock
 
-from pytest import fixture, skip
+import pytest
+import urllib
 
-from neo4j import GraphDatabase
+from neo4j import (
+    GraphDatabase,
+)
 from neo4j.exceptions import ServiceUnavailable
 from neo4j._exceptions import BoltHandshakeError
 from neo4j.io import Bolt
 
 
+
 NEO4J_RELEASES = getenv("NEO4J_RELEASES", "snapshot-enterprise 3.5-enterprise").split()
 NEO4J_HOST = "localhost"
 NEO4J_PORTS = {
@@ -170,7 +174,7 @@ def stop(self, timeout=None):
     NEO4J_RELEASES = ["existing"]
 
 
-@fixture(scope="session", params=NEO4J_RELEASES)
+@pytest.fixture(scope="session", params=NEO4J_RELEASES)
 def service(request):
     global NEO4J_SERVICE
     if NEO4J_DEBUG:
@@ -182,16 +186,19 @@ def service(request):
         if existing_service:
             NEO4J_SERVICE = existing_service
         else:
-            NEO4J_SERVICE = Neo4jService(auth=NEO4J_AUTH, image=request.param,
-                                         n_cores=NEO4J_CORES, n_replicas=NEO4J_REPLICAS)
-            NEO4J_SERVICE.start(timeout=300)
+            try:
+                NEO4J_SERVICE = Neo4jService(auth=NEO4J_AUTH, image=request.param, n_cores=NEO4J_CORES, n_replicas=NEO4J_REPLICAS)
+                NEO4J_SERVICE.start(timeout=300)
+            except urllib.error.HTTPError as error:
+                # pytest.skip(str(error))
+                pytest.xfail(str(error) + " " + request.param)
         yield NEO4J_SERVICE
         if NEO4J_SERVICE is not None:
             NEO4J_SERVICE.stop(timeout=300)
             NEO4J_SERVICE = None
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def addresses(service):
     try:
         machines = service.cores()
@@ -209,7 +216,7 @@ def addresses(service):
 #     return [machine.address for machine in machines]
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def address(addresses):
     try:
         return addresses[0]
@@ -225,7 +232,7 @@ def address(addresses):
 #         return None
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def targets(addresses):
     return " ".join("{}:{}".format(address[0], address[1]) for address in addresses)
 
@@ -235,7 +242,7 @@ def targets(addresses):
 #     return " ".join("{}:{}".format(address[0], address[1]) for address in readonly_addresses)
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def target(address):
     return "{}:{}".format(address[0], address[1])
 
@@ -248,17 +255,17 @@ def target(address):
 #         return None
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def bolt_uri(service, target):
     return "bolt://" + target
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def neo4j_uri(service, target):
     return "neo4j://" + target
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def uri(bolt_uri):
     return bolt_uri
 
@@ -271,47 +278,48 @@ def uri(bolt_uri):
 #         return None
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def auth():
     return NEO4J_AUTH
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def bolt_driver(target, auth):
     try:
         driver = GraphDatabase.bolt_driver(target, auth=auth)
         try:
             yield driver
         finally:
             driver.close()
-    except BoltHandshakeError as error:
-        skip(error.args[0])
+    except ServiceUnavailable as error:
+        if isinstance(error.__cause__, BoltHandshakeError):
+            pytest.skip(error.args[0])
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def neo4j_driver(target, auth):
     try:
         driver = GraphDatabase.neo4j_driver(target, auth=auth)
     except ServiceUnavailable as error:
-        if error.args[0] == "Server does not support routing":
-            skip(error.args[0])
+        if isinstance(error.__cause__, BoltHandshakeError):
+            pytest.skip(error.args[0])
+        elif error.args[0] == "Server does not support routing":
+            pytest.skip(error.args[0])
         else:
             raise
-    except BoltHandshakeError as error:
-        skip(error.args[0])
     else:
         try:
             yield driver
         finally:
             driver.close()
 
 
-@fixture(scope="session")
+@pytest.fixture(scope="session")
 def driver(neo4j_driver):
     return neo4j_driver
 
 
-@fixture()
+@pytest.fixture()
 def session(bolt_driver):
     session = bolt_driver.session()
     try:
@@ -320,14 +328,14 @@ def session(bolt_driver):
         session.close()
 
 
-@fixture()
+@pytest.fixture()
 def protocol_version(session):
     result = session.run("RETURN 1")
     yield session._connection.protocol_version
     result.consume()
 
 
-@fixture
+@pytest.fixture()
 def cypher_eval(bolt_driver):
 
     def run_and_rollback(tx, cypher, **parameters):