Add tx termination test based on failed tx.run

Author: injectives

tests/stub/tx_run/scripts/tx_successful_and_failing_streams.script

@@ -0,0 +1,18 @@
+!: BOLT 4.4
+
+A: HELLO {"{}": "*"}
+*: RESET
+C: BEGIN {"{}": "*"}
+S: SUCCESS {}
+C: RUN "RETURN 1 AS n" {"{}": "*"} {"{}": "*"}
+   PULL {"n": {"Z": "*"}, "[qid]": -1}
+S: SUCCESS {"fields": ["n"], "qid": 1}
+   RECORD [1]
+   RECORD [2]
+   SUCCESS {"has_more": true, "type": "r"}
+C: RUN "invalid" {"{}": "*"} {"{}": "*"}
+   PULL {"n": {"Z": "*"}, "[qid]": -1}
+S: FAILURE {"code": "Neo.ClientError.Statement.SyntaxError", "message": "Invalid input"}
+   IGNORED
+*: RESET
+?: GOODBYE

tests/stub/tx_run/test_tx_run.py

@@ -258,3 +258,27 @@ def test_failed_tx_run_allows_rollback(self):
 
     def test_failed_tx_run_allows_skipping_rollback(self):
         self._test_failed_tx_run(rollback=False)
+
+    @driver_feature(types.Feature.OPT_PULL_PIPELINING)
+    def test_should_terminate_tx_after_failed_run(self):
+        self._create_direct_driver()
+        self._server1.start(
+            path=self.script_path("tx_successful_and_failing_streams.script")
+        )
+        self._session = self._driver.session("r", fetch_size=2)
+        tx = self._session.begin_transaction()
+        res = tx.run("RETURN 1 AS n")
+        with self.assertRaises(types.DriverError) as exc:
+            tx.run("invalid")
+        self.assertEqual(exc.exception.code, "Neo.ClientError.Statement.SyntaxError")
+
+        # while already buffered records may be accessible, there must be no
+        # further PULLs and an exception must be raised
+        with self.assertRaises(types.DriverError):
+            for i in range(0, 3):
+                res.next()
+
+        tx.close()
+        self._session.close()
+        self._session = None
+        self._server1.done()