Add tests for server failing on RESET

Author: robsdedude

.pre-commit-config.yaml

@@ -25,7 +25,7 @@ repos:
           - flake8-bugbear
           - flake8-builtins
           - flake8-docstrings
-          - flake8-picky-parentheses
+          - flake8-picky-parentheses >= 0.5.3
           - flake8-quotes
           - pep8-naming
   - repo: https://github.com/pycqa/isort

tests/stub/retry/scripts/reset_fails_after_pull.script

@@ -0,0 +1,58 @@
+!: BOLT 4.4
+!: ALLOW RESTART
+!: PY invalid_responses = round = 0
+
+A: HELLO {"{}": "*"}
+
+IF: round == 0
+{{
+    PY: round += 1
+    *: RESET
+
+    {?
+        # Optionally a transaction
+        C: BEGIN {"{}": "*"}
+        S: SUCCESS {}
+    ?}
+    C: RUN {"U": "*"} {"{}": "*"} {"{}": "*"}
+    S: SUCCESS {"t_first": 2, "fields": ["n"], "qid": 0}
+    C: PULL {"n": {"Z": "*"}}
+    S: FAILURE {"code": "Neo.TransientError.Statement.RemoteExecutionTransientError", "message": "Remote execution failed with code N/A and message 'Failed to obtain connection towards READ server. Known routing table is: Ttl 1703005857129, currentTime 1703005847132, routers [], writers [], readers [], database 'neo4j''"}
+    {*
+        C: RESET
+        #INVALID_RESPONSE#
+        PY: invalid_responses += 1
+    *}
+}}
+ELIF: round == 1
+{{
+    PY: round += 1
+    *: RESET
+
+    {{
+        C: BEGIN {"{}": "*"}
+        S: SUCCESS {}
+        C: RUN {"U": "*"} {"{}": "*"} {"{}": "*"}
+        S: SUCCESS {"t_first": 2, "fields": ["n"], "qid": 0}
+        C: PULL {"n": {"Z": "*"}}
+        S: RECORD [1]
+        S: SUCCESS {"type": "r"}
+        C: COMMIT
+        S: SUCCESS {}
+    ----
+        C: RUN {"U": "*"} {"{}": "*"} {"{}": "*"}
+        S: SUCCESS {"t_first": 2, "fields": ["n"], "qid": 0}
+        C: PULL {"n": {"Z": "*"}}
+        S: RECORD [1]
+        S: SUCCESS {"type": "r"}
+    }}
+
+    *: RESET
+}}
+ELSE:
+{{
+    PY: round += 1
+    S: <EXIT>
+}}
+
+?: GOODBYE

tests/stub/retry/test_retry.py

@@ -222,3 +222,105 @@ def once(tx):
         for failure in failures:
             with self.subTest(failure=failure):
                 _test()
+
+    def test_reset_fails_after_pull(self):
+        def _test(invalid_response_, api_):
+            def check_exception(exc):
+                self.assertEqual(
+                    exc.exception.code,
+                    "Neo.TransientError.Statement."
+                    "RemoteExecutionTransientError"
+                )
+                if self.driver_supports_features(
+                    types.Feature.API_RETRYABLE_EXCEPTION
+                ):
+                    self.assertTrue(exc.exception.retryable)
+
+            def api_call(session_):
+                if api_ == "session":
+                    with self.assertRaises(types.DriverError) as exc:
+                        result = session_.run("RETURN 1 AS n")
+                        list(result)
+                    check_exception(exc)
+                elif api_ == "explicit_tx":
+                    tx = session_.begin_transaction()
+                    try:
+                        with self.assertRaises(types.DriverError) as exc:
+                            result = tx.run("RETURN 1 AS n")
+                            list(result)
+                        check_exception(exc)
+                    finally:
+                        tx.close()
+                elif api_ == "managed_tx":
+                    run = 0
+
+                    def work(tx):
+                        nonlocal run
+                        run += 1
+                        if run == 1:
+                            with self.assertRaises(types.DriverError) as exc:
+                                result = tx.run("RETURN 1 AS n")
+                                list(result)
+                            check_exception(exc)
+                            raise exc.exception
+                        else:
+                            result = tx.run("RETURN 1 AS n")
+                            return list(result)
+
+                    records = session_.execute_write(work)
+                    assert len(records) == 1
+                    self.assertEqual(records, [
+                        types.Record(values=[types.CypherInt(1)])
+                    ])
+                else:
+                    raise ValueError(f"Unknown API: {api_}")
+
+            self._server.start(
+                path=self.script_path("reset_fails_after_pull.script"),
+                vars_={
+                    "#INVALID_RESPONSE#": invalid_response_,
+                }
+            )
+            auth = types.AuthorizationToken("basic", principal="",
+                                            credentials="")
+            driver = Driver(self._backend,
+                            "bolt://%s" % self._server.address, auth)
+            try:
+                session = driver.session("r")
+                try:
+                    api_call(session)
+
+                finally:
+                    session.close()
+                # driver should've killed the misbehaving connection
+                try:
+                    self.assertEqual(
+                        self._server.count_responses("<HANGUP>"), 1
+                    )
+                finally:
+                    self._server._dump()
+            finally:
+                driver.close()
+            self._server.done()
+
+        invalid_responses = (
+            (
+                'S: FAILURE {"code": "Neo.ClientError.General.Unknown", '
+                '"message": "The driver should ignore this error!"}'
+            ),
+            "S: IGNORED",
+            (
+                "# MIXED   \n"
+                "IF: invalid_responses <= 1\n"
+                '    S: FAILURE {"code": "Neo.ClientError.General.Unknown", '
+                '"message": "The driver should ignore this error!"}\n'
+                "ELSE:\n"
+                "    S: IGNORED\n"
+            )
+        )
+        for invalid_response in invalid_responses:
+            for api in ("session", "explicit_tx", "managed_tx"):
+                with self.subTest(response=invalid_response[2:10].strip(),
+                                  api=api):
+                    _test(invalid_response, api)
+                self._server.reset()