Fix RxSession in Testkit Backend (#980)

Testkit environment variables without `TEST_` prefix are not forward by the Testkit to the
scripts in the driver. So we need change the session type variable expected by the `backend.py`
to  be `TEST_SESSION_TYPE` and then remove the prefix before redirect to the backend.

This PR also fix the `SessionRun` and `TransactionRun` to not initiate the stream.

**Warn**: `RxSession.beginTransaction` and `RxSession.execute*` were tweaked. The `Observable<RxTransaction>` only emits the `RxTransaction` when the begin is executed with success. The begin still defer the execution to the point observer is subscribed. 

Co-authored-by: Florent Biville <445792+fbiville@users.noreply.github.com>

Author: bigmontz

packages/core/src/internal/util.ts

@@ -24,7 +24,10 @@ import { stringify } from '../json'
 
 const ENCRYPTION_ON: EncryptionLevel = 'ENCRYPTION_ON'
 const ENCRYPTION_OFF: EncryptionLevel = 'ENCRYPTION_OFF'
-
+// eslint-disable-next-line @typescript-eslint/naming-convention
+const __isBrokenObject__ = '__isBrokenObject__'
+// eslint-disable-next-line @typescript-eslint/naming-convention
+const __reason__ = '__reason__'
 /**
  * Verifies if the object is null or empty
  * @param obj The subject object
@@ -237,7 +240,16 @@ function createBrokenObject<T extends object> (error: Error, object: any = {}):
   }
 
   return new Proxy(object, {
-    get: fail,
+    get: (_: T, p: string | Symbol): any => {
+      if (p === __isBrokenObject__) {
+        return true
+      } else if (p === __reason__) {
+        return error
+      } else if (p === 'toJSON') {
+        return undefined
+      }
+      fail()
+    },
     set: fail,
     apply: fail,
     construct: fail,
@@ -253,6 +265,27 @@ function createBrokenObject<T extends object> (error: Error, object: any = {}):
   })
 }
 
+/**
+ * Verifies if it is a Broken Object
+ * @param {any} object The object
+ * @returns {boolean} If it was created with createBrokenObject
+ */
+function isBrokenObject (object: any): boolean {
+  return object !== null && typeof object === 'object' && object[__isBrokenObject__] === true
+}
+
+/**
+ * Returns if the reason the object is broken.
+ *
+ * This method should only be called with instances create with {@link createBrokenObject}
+ *
+ * @param {any} object The object
+ * @returns {Error} The reason the object is broken
+ */
+function getBrokenObjectReason (object: any): Error {
+  return object[__reason__]
+}
+
 export {
   isEmptyObjectOrNull,
   isObject,
@@ -265,5 +298,7 @@ export {
   validateQueryAndParameters,
   ENCRYPTION_ON,
   ENCRYPTION_OFF,
-  createBrokenObject
+  createBrokenObject,
+  isBrokenObject,
+  getBrokenObjectReason
 }

packages/core/src/json.ts

@@ -17,14 +17,25 @@
  * limitations under the License.
  */
 
+import { isBrokenObject, getBrokenObjectReason } from './internal/util'
+
 /**
  * Custom version on JSON.stringify that can handle values that normally don't support serialization, such as BigInt.
  * @private
  * @param val A JavaScript value, usually an object or array, to be converted.
  * @returns A JSON string representing the given value.
  */
 export function stringify (val: any): string {
-  return JSON.stringify(val, (_, value) =>
-    typeof value === 'bigint' ? `${value}n` : value
-  )
+  return JSON.stringify(val, (_, value) => {
+    if (isBrokenObject(value)) {
+      return {
+        __isBrokenObject__: true,
+        __reason__: getBrokenObjectReason(value)
+      }
+    }
+    if (typeof value === 'bigint') {
+      return `${value}n`
+    }
+    return value
+  })
 }

packages/core/test/__snapshots__/json.test.ts.snap

@@ -0,0 +1,7 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`json .stringify should handle objects created with createBrokenObject 1`] = `"{\\"__isBrokenObject__\\":true,\\"__reason__\\":{\\"code\\":\\"N/A\\",\\"name\\":\\"Neo4jError\\",\\"retriable\\":false}}"`;
+
+exports[`json .stringify should handle objects created with createBrokenObject in list 1`] = `"[{\\"__isBrokenObject__\\":true,\\"__reason__\\":{\\"code\\":\\"N/A\\",\\"name\\":\\"Neo4jError\\",\\"retriable\\":false}}]"`;
+
+exports[`json .stringify should handle objects created with createBrokenObject inside other object 1`] = `"{\\"number\\":1,\\"broken\\":{\\"__isBrokenObject__\\":true,\\"__reason__\\":{\\"code\\":\\"N/A\\",\\"name\\":\\"Neo4jError\\",\\"retriable\\":false}}}"`;

packages/core/test/internal/util.test.ts

@@ -16,6 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+import { newError } from '../../src'
 import Integer, { int } from '../../src/integer'
 import {
   isEmptyObjectOrNull,
@@ -28,7 +29,10 @@ import {
   assertValidDate,
   validateQueryAndParameters,
   ENCRYPTION_ON,
-  ENCRYPTION_OFF
+  ENCRYPTION_OFF,
+  createBrokenObject,
+  isBrokenObject,
+  getBrokenObjectReason
 } from '../../src/internal/util'
 
 /* eslint-disable no-new-wrappers */
@@ -250,4 +254,47 @@ describe('Util', () => {
     expect(ENCRYPTION_ON).toBe('ENCRYPTION_ON'))
   test('should ENCRYPTION_OFF toBe "ENCRYPTION_OFF"', () =>
     expect(ENCRYPTION_OFF).toBe('ENCRYPTION_OFF'))
+
+  describe('isBrokenObject', () => {
+    it('should return true when object created with createBrokenObject', () => {
+      const object = createBrokenObject(newError('error'), {})
+
+      expect(isBrokenObject(object)).toBe(true)
+    })
+
+    it('should return false for regular objects', () => {
+      const object = {}
+
+      expect(isBrokenObject(object)).toBe(false)
+    })
+
+    it('should return false for non-objects', () => {
+      expect(isBrokenObject(null)).toBe(false)
+      expect(isBrokenObject(undefined)).toBe(false)
+      expect(isBrokenObject(1)).toBe(false)
+      expect(isBrokenObject(() => {})).toBe(false)
+      expect(isBrokenObject('string')).toBe(false)
+    })
+  })
+
+  describe('getBrokenObjectReason', () => {
+    it('should return the reason the object is broken', () => {
+      const reason = newError('error')
+      const object = createBrokenObject(reason, {})
+
+      expect(getBrokenObjectReason(object)).toBe(reason)
+    })
+  })
+
+  describe('createBrokenObject', () => {
+    describe('toJSON', () => {
+      it('should return undefined', () => {
+        const reason = newError('error')
+        const object = createBrokenObject(reason, {})
+
+        // @ts-expect-error
+        expect(object.toJSON).toBeUndefined()
+      })
+    })
+  })
 })

packages/core/test/json.test.ts

@@ -0,0 +1,49 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { json, newError } from '../src'
+import { createBrokenObject } from '../src/internal/util'
+
+describe('json', () => {
+  describe('.stringify', () => {
+    it('should handle objects created with createBrokenObject', () => {
+      const reason = newError('some error')
+      const broken = createBrokenObject(reason, { })
+
+      expect(json.stringify(broken)).toMatchSnapshot()
+    })
+
+    it('should handle objects created with createBrokenObject in list', () => {
+      const reason = newError('some error')
+      const broken = createBrokenObject(reason, { })
+
+      expect(json.stringify([broken])).toMatchSnapshot()
+    })
+
+    it('should handle objects created with createBrokenObject inside other object', () => {
+      const reason = newError('some error')
+      const broken = createBrokenObject(reason, { })
+
+      expect(json.stringify({
+        number: 1,
+        broken
+      })).toMatchSnapshot()
+    })
+  })
+})

packages/neo4j-driver/src/result-rx.js

@@ -35,8 +35,9 @@ export default class RxResult {
    * @constructor
    * @protected
    * @param {Observable<Result>} result - An observable of single Result instance to relay requests.
+   * @param {number} state - The streaming state
    */
-  constructor (result) {
+  constructor (result, state) {
     const replayedResult = result.pipe(publishReplay(1), refCount())
 
     this._result = replayedResult
@@ -48,7 +49,7 @@ export default class RxResult {
     this._records = undefined
     this._controls = new StreamControl()
     this._summary = new ReplaySubject()
-    this._state = States.READY
+    this._state = state || States.READY
   }
 
   /**
@@ -184,6 +185,31 @@ export default class RxResult {
     }
   }
 
+  /**
+   * Create a {@link Observable} for the current {@link RxResult}
+   *
+   *
+   * @package
+   * @experimental
+   * @since 5.0
+   * @return {Observable<RxResult>}
+   */
+  _toObservable () {
+    function wrap (result) {
+      return new Observable(observer => {
+        observer.next(result)
+        observer.complete()
+      })
+    }
+    return new Observable(observer => {
+      this._result.subscribe({
+        complete: () => observer.complete(),
+        next: result => observer.next(new RxResult(wrap(result)), this._state),
+        error: e => observer.error(e)
+      })
+    })
+  }
+
   _setupRecordsStream (result) {
     if (this._records) {
       return this._records

packages/neo4j-driver/src/session-rx.js

@@ -205,12 +205,14 @@ export default class RxSession {
 
     return new Observable(observer => {
       try {
-        observer.next(
-          new RxTransaction(
-            this._session._beginTransaction(accessMode, txConfig)
-          )
-        )
-        observer.complete()
+        this._session._beginTransaction(accessMode, txConfig)
+          .then(tx => {
+            observer.next(
+              new RxTransaction(tx)
+            )
+            observer.complete()
+          })
+          .catch(err => observer.error(err))
       } catch (err) {
         observer.error(err)
       }

packages/neo4j-driver/test/index.test.js

@@ -152,6 +152,10 @@ describe('#unit index', () => {
 
       function subject () {
         const driver = neo4j.driver('bolt://localhost')
+        driver._createSession = () => ({
+          _mode: 'READ',
+          _beginTransaction: async () => new Transaction({})
+        })
         return driver.rxSession().beginTransaction().toPromise()
       }
     })
@@ -202,6 +206,10 @@ describe('#unit index', () => {
 
       async function subject () {
         const driver = neo4j.driver('bolt://localhost')
+        driver._createSession = () => ({
+          _mode: 'READ',
+          _beginTransaction: async () => new Transaction({})
+        })
         const tx = await driver.rxSession().beginTransaction().toPromise()
         return InternalRxManagedTransaction.fromTransaction(tx)
       }

packages/testkit-backend/src/feature/async.js

@@ -1,7 +1,6 @@
 const features = [
   'Feature:API:Result.List',
   'Feature:API:Result.Peek',
-  'Optimization:EagerTransactionBegin',
   'Optimization:PullPipelining'
 ]
 

packages/testkit-backend/src/feature/common.js

@@ -33,7 +33,7 @@ const features = [
   'Feature:API:ConnectionAcquisitionTimeout',
   'Feature:API:Driver:GetServerInfo',
   'Feature:API:Driver.VerifyConnectivity',
-  'Feature:API:Result.Peek',
+  'Optimization:EagerTransactionBegin',
   'Optimization:ImplicitDefaultArguments',
   'Optimization:MinimalResets',
   ...SUPPORTED_TLS