Extend test for issue 3174 - add suite test cases for cursor and null…

… check for pg-cursor/handleDataRow.
brianc · Nov 14, 2024 · 563f977 · 563f977
1 parent 4da0065
commit 563f977
Show file tree

Hide file tree

Showing 2 changed files with 84 additions and 10 deletions.
diff --git a/packages/pg-cursor/index.js b/packages/pg-cursor/index.js
@@ -112,6 +112,12 @@ class Cursor extends EventEmitter {
   handleDataRow(msg) {
     const row = this._result.parseRow(msg.fields)
     this.emit('row', row, this._result)
+
+    if (this._rows == null) {
+      const error = new Error('Received unexpected dataRow message from backend.')
+      this.handleError(error)
+      return
+    }
     this._rows.push(row)
   }
 

diff --git a/packages/pg/test/integration/gh-issues/3174-tests.js b/packages/pg/test/integration/gh-issues/3174-tests.js
@@ -104,25 +104,89 @@ const testErrorBuffer = (bufferName, errorBuffer) => {
     if (!cli.native) {
       assert(errorHit)
       // further queries on the client should fail since its in an invalid state
-      await assert.rejects(() => client.query('SELECTR NOW()'), 'Further queries on the client should reject')
+      await assert.rejects(() => client.query('SELECT NOW()'), 'Further queries on the client should reject')
     }
+    await closeServer()
+  })
+
+  suite.testAsync(`Out of order ${bufferName} on extended query is catchable`, async () => {
+    const closeServer = await new Promise((resolve, reject) => {
+      return startMockServer(options.port, errorBuffer, (closeServer) => resolve(closeServer))
+    })
+    const client = new helper.Client(options)
+    await client.connect()
+
+    let errorHit = false
+    client.on('error', () => {
+      errorHit = true
+    })
+
+    await client.query('SELECT $1', ['foo'])
+    await delay(40)
+
+    // the native client only emits a notice message and keeps on its merry way
+    if (!cli.native) {
+      assert(errorHit)
+      // further queries on the client should fail since its in an invalid state
+      await assert.rejects(() => client.query('SELECT NOW()'), 'Further queries on the client should reject')
+    }
+
+    await client.end()
+
+    await closeServer()
+  })
+
+  suite.testAsync(`Out of order ${bufferName} on pool is catchable`, async () => {
+    const closeServer = await new Promise((resolve, reject) => {
+      return startMockServer(options.port, errorBuffer, (closeServer) => resolve(closeServer))
+    })
+    const pool = new helper.pg.Pool(options)
+
+    let errorHit = false
+    pool.on('error', () => {
+      errorHit = true
+    })
+
+    await pool.query('SELECT $1', ['foo'])
+    await delay(100)
+
+    if (!cli.native) {
+      assert(errorHit)
+      assert.strictEqual(pool.idleCount, 0, 'Pool should have no idle clients')
+      assert.strictEqual(pool.totalCount, 0, 'Pool should have no connected clients')
+    }
+
+    await pool.end()
+    await closeServer()
+  })
+
+  suite.testAsync(`Out of order ${bufferName} on simple query using cursors is catchable`, async () => {
+    const closeServer = await new Promise((resolve, reject) => {
+      return startMockServer(options.port, errorBuffer, (closeServer) => resolve(closeServer))
+    })
+    const client = new helper.Client(options)
+    await client.connect()
+
+    let errorHit = false
+    client.on('error', () => {
+      errorHit = true
+    })
 
-    // Same run but using cursor
     const cursor = await client.query(new Cursor('SELECT NOW()'))
     cursor.read(100, () => {})
     await cursor.close()
     await delay(50)
 
+    // the native client only emits a notice message and keeps on its merry way
     if (!cli.native) {
       assert(errorHit)
       // further queries on the client should fail since its in an invalid state
-      await assert.rejects(() => client.query('SELECTR NOW()'), 'Further queries on the client should reject')
+      await assert.rejects(() => client.query('SELECT NOW()'), 'Further queries on the client should reject')
     }
-
     await closeServer()
   })
 
-  suite.testAsync(`Out of order ${bufferName} on extended query is catchable`, async () => {
+  suite.testAsync(`Out of order ${bufferName} on extended query using cursors is catchable`, async () => {
     const closeServer = await new Promise((resolve, reject) => {
       return startMockServer(options.port, errorBuffer, (closeServer) => resolve(closeServer))
     })
@@ -134,22 +198,24 @@ const testErrorBuffer = (bufferName, errorBuffer) => {
       errorHit = true
     })
 
-    await client.query('SELECT $1', ['foo'])
-    await delay(40)
+    const cursor = await client.query(new Cursor('SELECT $1', ['foo']))
+    cursor.read(100, () => {})
+    await cursor.close()
+    await delay(50)
 
     // the native client only emits a notice message and keeps on its merry way
     if (!cli.native) {
       assert(errorHit)
       // further queries on the client should fail since its in an invalid state
-      await assert.rejects(() => client.query('SELECTR NOW()'), 'Further queries on the client should reject')
+      await assert.rejects(() => client.query('SELECT NOW()'), 'Further queries on the client should reject')
     }
 
     await client.end()
 
     await closeServer()
   })
 
-  suite.testAsync(`Out of order ${bufferName} on pool is catchable`, async () => {
+  suite.testAsync(`Out of order ${bufferName} on pool using cursors is catchable`, async () => {
     const closeServer = await new Promise((resolve, reject) => {
       return startMockServer(options.port, errorBuffer, (closeServer) => resolve(closeServer))
     })
@@ -160,7 +226,9 @@ const testErrorBuffer = (bufferName, errorBuffer) => {
       errorHit = true
     })
 
-    await pool.query('SELECT $1', ['foo'])
+    const cursor = await pool.query(new Cursor('SELECT $1', ['foo']))
+    cursor.read(100, () => {})
+    await cursor.close()
     await delay(100)
 
     if (!cli.native) {