test(qe): driver adapters: parameterize test kit to test any driver a…

…dapter that we support (#4265)

This PR changes the executor for driver adapters to instantiate a different driver adapter based on environment configuration.

From the chunk of documentation added to the connector kit README:

#### Running tests through driver adapters

The query engine is able to delegate query execution to javascript through [driver adapters](query-engine/driver-adapters/js/README.md).
This means that instead of drivers being implemented in Rust, it's a layer of adapters over NodeJs drivers the code that actually communicates with the databases. 

To run tests through a driver adapters, you should also configure the following environment variables:

* `NODE_TEST_EXECUTOR`: tells the query engine test kit to use an external process to run the queries, this is a node process running
a program that will read the queries to run from STDIN, and return responses to STDOUT. The connector kit follows a protocol over JSON RPC for this communication. 
* `DRIVER_ADAPTER`: tells the test executor to use a particular driver adapter. Set to `neon`, `planetscale` or any other supported adapter.
* `DRIVER_ADAPTER_URL_OVERRIDE`: it overrides the schema URL for the database to use one understood by the driver adapter (ex. neon, planetscale)
 

Example:

```shell
export NODE_TEST_EXECUTOR="$WORKSPACE_ROOT/query-engine/driver-adapters/js/connector-test-kit-executor/script/start_node.sh"
export DRIVER_ADAPTER=neon
export DRIVER_ADAPTER_URL_OVERRIDE ="postgres://USER:PASSWORD@DATABASExxxx"
````

Closes https://github.com/prisma/team-orm/issues/364

.envrc

@@ -22,6 +22,12 @@ export QE_LOG_LEVEL=debug # Set it to "trace" to enable query-graph debugging lo
 # export PRISMA_RENDER_DOT_FILE=1 # Uncomment to enable rendering a dot file of the Query Graph from an executed query.
 # export FMT_SQL=1 # Uncomment it to enable logging formatted SQL queries
 
+### Uncomment to run driver adapters tests. See query-engine-driver-adapters.yml workflow for how tests run in CI.
+# export EXTERNAL_TEST_EXECUTOR="$(pwd)/query-engine/driver-adapters/js/connector-test-kit-executor/script/start_node.sh"
+# export DRIVER_ADAPTER=pg # Set to pg, neon or planetscale
+# export PRISMA_DISABLE_QUAINT_EXECUTORS=1 # Disable quaint executors for driver adapters
+# export DRIVER_ADAPTER_URL_OVERRIDE ="postgres://USER:PASSWORD@DATABASExxxx" # Override the database url for the driver adapter tests
+
 # Mongo image requires additional wait time on arm arch for some reason.
 if uname -a | grep -q 'arm64'; then
     export INIT_WAIT_SEC="10"

.github/workflows/query-engine-driver-adapters.yml

@@ -28,20 +28,24 @@ jobs:
             version: "13"
             driver_adapter: "pg"
         node_version: ["18"]
-
     env:
       LOG_LEVEL: "info"
       LOG_QUERIES: "y"
       RUST_LOG: "info"
       RUST_LOG_FORMAT: "devel"
       RUST_BACKTRACE: "1"
+      PRISMA_DISABLE_QUAINT_EXECUTORS: "1"
       CLICOLOR_FORCE: "1"
       CLOSED_TX_CLEANUP: "2"
       SIMPLE_TEST_MODE: "1"
       QUERY_BATCH_SIZE: "10"
       TEST_CONNECTOR: ${{ matrix.database.connector }}
       TEST_CONNECTOR_VERSION: ${{ matrix.database.version }}
-      TEST_DRIVER_ADAPTER: ${{ matrix.database.driver_adapter }}
+      WORKSPACE_ROOT: ${{ github.workspace }}
+      # Driver adapter testing specific env vars
+      EXTERNAL_TEST_EXECUTOR: "${{ github.workspace }}/query-engine/driver-adapters/js/connector-test-kit-executor/script/start_node.sh"
+      DRIVER_ADAPTER: ${{ matrix.database.driver_adapter }}
+      DRIVER_ADAPTER_URL_OVERRIDE: ${{ matrix.database.driver_adapter_url }}
 
     runs-on: buildjet-16vcpu-ubuntu-2004
     steps:
@@ -77,7 +81,7 @@ jobs:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
 
-      - name: "Start ${{ matrix.database.name }} (${{ matrix.engine_protocol }})"
+      - name: "Start ${{ matrix.database.name }}"
         run: make start-${{ matrix.database.name }}
 
       - uses: dtolnay/rust-toolchain@stable
@@ -90,8 +94,5 @@ jobs:
 
       - name: "Run tests"
         run:  cargo test --package query-engine-tests -- --test-threads=1
-        env:
-          CLICOLOR_FORCE: 1
-          WORKSPACE_ROOT: ${{ github.workspace }}
-          NODE_TEST_EXECUTOR: "${{ github.workspace }}/query-engine/driver-adapters/js/connector-test-kit-executor/script/start_node.sh"
+
 

query-engine/connector-test-kit-rs/README.md

@@ -1,6 +1,5 @@
 # Query Engine Test Kit - A Full Guide
-The test kit is a (currently incomplete) port of the Scala test kit, located in `../connector-test-kit`.
-It's fully focused on integration testing the query engine through request-response assertions.
+The test kit is focused on integration testing the query engine through request-response assertions.
 
 ## Test organization
 
@@ -35,8 +34,10 @@ Contains the main bulk of logic to make tests run, which is mostly invisible to
 Tests are executed in the context of *one* _connector_ (with version) and _runner_. Some tests may only be specified to run for a subset of connectors or versions, in which case they will be skipped. Testing all connectors at once is not supported, however, for example, CI will run all the different connectors and versions concurrently in separate runs.
 
 ### Configuration
+
 Tests must be configured to run There's a set of env vars that is always useful to have and an optional one.
 Always useful to have:
+
 ```shell
 export WORKSPACE_ROOT=/path/to/engines/repository/root
 ```
@@ -54,6 +55,7 @@ As previously stated, the above can be omitted in favor of the `.test_config` co
     "version": "10"
 }
 ```
+
 The config file must be either in the current working folder from which you invoke a test run or in `$WORKSPACE_ROOT`.
 It's recommended to use the file-based config as it's easier to switch between providers with an open IDE (reloading env vars would usually require reloading the IDE).
 The workspace root makefile contains a series of convenience commands to setup different connector test configs, e.g. `make dev-postgres10` sets up the correct test config file for the tests to pick up.
@@ -62,7 +64,29 @@ On the note of docker containers: Most connectors require an endpoint to run aga
 
 If you choose to set up the databases yourself, please note that the connection strings used in the tests (found in the files in `<repo_root>/query-engine/connector-test-kit-rs/query-tests-setup/src/connector_tag/`) to set up user, password and database for the test user.
 
+#### Running tests through driver adapters
+
+The query engine is able to delegate query execution to javascript through [driver adapters](query-engine/driver-adapters/js/README.md).
+This means that instead of drivers being implemented in Rust, it's a layer of adapters over NodeJs drivers the code that actually communicates with the databases. 
+
+To run tests through a driver adapters, you should also configure the following environment variables:
+
+* `EXTERNAL_TEST_EXECUTOR`: tells the query engine test kit to use an external process to run the queries, this is a node process running
+a program that will read the queries to run from STDIN, and return responses to STDOUT. The connector kit follows a protocol over JSON RPC for this communication. 
+* `DRIVER_ADAPTER`: tells the test executor to use a particular driver adapter. Set to `neon`, `planetscale` or any other supported adapter.
+* `DRIVER_ADAPTER_URL_OVERRIDE`: it overrides the schema URL for the database to use one understood by the driver adapter (ex. neon, planetscale)
+
+
+Example:
+
+```shell
+export EXTERNAL_TEST_EXECUTOR="$WORKSPACE_ROOT/query-engine/driver-adapters/js/connector-test-kit-executor/script/start_node.sh"
+export DRIVER_ADAPTER=neon
+export DRIVER_ADAPTER_URL_OVERRIDE ="postgres://USER:PASSWORD@DATABASExxxx"
+````
+
 ### Running
+
 Note that by default tests run concurrently.
 
 - VSCode should automatically detect tests and display `run test`.

query-engine/connector-test-kit-rs/query-tests-setup/src/connector_tag/js.rs

@@ -1,7 +1,7 @@
-mod node_process;
+mod external_process;
 
 use super::*;
-use node_process::*;
+use external_process::*;
 use serde::de::DeserializeOwned;
 use std::{collections::HashMap, sync::atomic::AtomicU64};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
@@ -10,5 +10,5 @@ pub(crate) async fn executor_process_request<T: DeserializeOwned>(
     method: &str,
     params: serde_json::Value,
 ) -> Result<T, Box<dyn std::error::Error + Send + Sync>> {
-    NODE_PROCESS.request(method, params).await
+    EXTERNAL_PROCESS.request(method, params).await
 }

query-engine/connector-test-kit-rs/query-tests-setup/src/connector_tag/js/node_process.rs → query-engine/connector-test-kit-rs/query-tests-setup/src/connector_tag/js/external_process.rs

@@ -69,7 +69,7 @@ impl ExecutorProcess {
     }
 }
 
-pub(super) static NODE_PROCESS: Lazy<ExecutorProcess> =
+pub(super) static EXTERNAL_PROCESS: Lazy<ExecutorProcess> =
     Lazy::new(|| match std::thread::spawn(ExecutorProcess::new).join() {
         Ok(Ok(process)) => process,
         Ok(Err(err)) => exit_with_message(1, &format!("Failed to start node process. Details: {err}")),
@@ -87,7 +87,10 @@ fn start_rpc_thread(mut receiver: mpsc::Receiver<ReqImpl>) -> Result<()> {
 
     let env_var = match crate::EXTERNAL_TEST_EXECUTOR.as_ref() {
         Some(env_var) => env_var,
-        None => exit_with_message(1, "start_rpc_thread() error: NODE_TEST_EXECUTOR env var is not defined"),
+        None => exit_with_message(
+            1,
+            "start_rpc_thread() error: EXTERNAL_TEST_EXECUTOR env var is not defined",
+        ),
     };
 
     tokio::runtime::Builder::new_current_thread()

query-engine/connector-test-kit-rs/query-tests-setup/src/lib.rs

@@ -42,8 +42,7 @@ pub static ENV_LOG_LEVEL: Lazy<String> = Lazy::new(|| std::env::var("LOG_LEVEL")
 pub static ENGINE_PROTOCOL: Lazy<String> =
     Lazy::new(|| std::env::var("PRISMA_ENGINE_PROTOCOL").unwrap_or_else(|_| "graphql".to_owned()));
 
-// TODO: rename env var to EXTERNAL_TEST_EXECUTOR
-static EXTERNAL_TEST_EXECUTOR: Lazy<Option<String>> = Lazy::new(|| std::env::var("NODE_TEST_EXECUTOR").ok());
+static EXTERNAL_TEST_EXECUTOR: Lazy<Option<String>> = Lazy::new(|| std::env::var("EXTERNAL_TEST_EXECUTOR").ok());
 
 /// Teardown of a test setup.
 async fn teardown_project(datamodel: &str, db_schemas: &[&str], schema_id: Option<usize>) -> TestResult<()> {

query-engine/driver-adapters/js/connector-test-kit-executor/package.json

@@ -13,9 +13,12 @@
   "sideEffects": false,
   "license": "Apache-2.0",
   "dependencies": {
+    "@neondatabase/serverless": "^0.6.0",
+    "@prisma/adapter-neon": "workspace:*",
     "@prisma/adapter-pg": "workspace:*",
     "@prisma/driver-adapter-utils": "workspace:*",
+    "@types/pg": "^8.10.2",
     "pg": "^8.11.3",
-    "@types/pg": "^8.10.2"
+    "undici": "^5.23.0"
   }
 }

query-engine/driver-adapters/js/connector-test-kit-executor/src/index.ts

@@ -1,10 +1,22 @@
-import pgDriver from 'pg'
-import * as pg from '@prisma/adapter-pg'
 import * as qe from './qe'
 import * as engines from './engines/Library'
 import * as readline from 'node:readline'
 import * as jsonRpc from './jsonRpc'
-import {bindAdapter, ErrorCapturingDriverAdapter} from "@prisma/driver-adapter-utils";
+
+// pg dependencies
+import pgDriver from 'pg'
+import * as prismaPg from '@prisma/adapter-pg'
+
+// neon dependencies
+import { Pool as NeonPool, neonConfig } from '@neondatabase/serverless'
+import { WebSocket } from 'undici'
+import * as prismaNeon from '@prisma/adapter-neon'
+neonConfig.webSocketConstructor = WebSocket
+
+import {bindAdapter, DriverAdapter, ErrorCapturingDriverAdapter} from "@prisma/driver-adapter-utils";
+
+const SUPPORTED_ADAPTERS: Record<string, (_ : string) => Promise<DriverAdapter>>
+    = {pg: pgAdapter, neon: neonAdapter};
 
 async function main(): Promise<void> {
     const iface = readline.createInterface({
@@ -94,22 +106,21 @@ async function handleRequest(method: string, params: unknown): Promise<unknown>
                 schemaId: number,
                 options: unknown
             }
+
             console.error("Got `startTx", params)
-            const { schemaId, options } = params as StartTxPayload
+            const {schemaId, options} = params as StartTxPayload
             const result = await schemas[schemaId].startTransaction(JSON.stringify(options), "")
             return JSON.parse(result)
-
-
-
         }
 
         case 'commitTx': {
             interface CommitTxPayload {
                 schemaId: number,
                 txId: string,
             }
+
             console.error("Got `commitTx", params)
-            const { schemaId, txId } = params as CommitTxPayload
+            const {schemaId, txId} = params as CommitTxPayload
             const result = await schemas[schemaId].commitTransaction(txId, '{}')
             return JSON.parse(result)
         }
@@ -119,8 +130,9 @@ async function handleRequest(method: string, params: unknown): Promise<unknown>
                 schemaId: number,
                 txId: string,
             }
+
             console.error("Got `rollbackTx", params)
-            const { schemaId, txId } = params as RollbackTxPayload
+            const {schemaId, txId} = params as RollbackTxPayload
             const result = await schemas[schemaId].rollbackTransaction(txId, '{}')
             return JSON.parse(result)
         }
@@ -132,15 +144,15 @@ async function handleRequest(method: string, params: unknown): Promise<unknown>
             const castParams = params as TeardownPayload;
             await schemas[castParams.schemaId].disconnect("")
             delete schemas[castParams.schemaId]
+            delete adapters[castParams.schemaId]
             delete queryLogs[castParams.schemaId]
             return {}
-
         }
-
         case 'getLogs': {
             interface GetLogsPayload {
                 schemaId: number
             }
+
             const castParams = params as GetLogsPayload
             return queryLogs[castParams.schemaId] ?? []
         }
@@ -170,10 +182,39 @@ function respondOk(requestId: number, payload: unknown) {
 }
 
 async function initQe(url: string, prismaSchema: string, logCallback: qe.QueryLogCallback): Promise<[engines.QueryEngineInstance, ErrorCapturingDriverAdapter]> {
-    const pool = new pgDriver.Pool({ connectionString: url })
-    const adapter = bindAdapter(new pg.PrismaPg(pool))
-    const engineInstance = qe.initQueryEngine(adapter, prismaSchema, logCallback)
-    return [engineInstance, adapter];
+    const adapter = await adapterFromEnv(url) as DriverAdapter
+    const errorCapturingAdapter = bindAdapter(adapter)
+    const engineInstance = qe.initQueryEngine(errorCapturingAdapter, prismaSchema, logCallback)
+    return [engineInstance, errorCapturingAdapter];
+}
+
+async function adapterFromEnv(url: string): Promise<DriverAdapter> {
+    const adapter = process.env.DRIVER_ADAPTER ?? ''
+
+    if (adapter == '') {
+        throw new Error("DRIVER_ADAPTER is not defined or empty.")
+    }
+
+    if (!(adapter in SUPPORTED_ADAPTERS)) {
+        throw new Error(`Unsupported driver adapter: ${adapter}`)
+    }
+
+    return await SUPPORTED_ADAPTERS[adapter](url);
+}
+
+async function pgAdapter(url: string): Promise<DriverAdapter> {
+    const pool = new pgDriver.Pool({connectionString: url})
+    return new prismaPg.PrismaPg(pool)
+}
+
+async function neonAdapter(_: string): Promise<DriverAdapter> {
+    const connectionString = process.env.DRIVER_ADAPTER_URL_OVERRIDE ?? ''
+    if (connectionString == '') {
+        throw new Error("DRIVER_ADAPTER_URL_OVERRIDE is not defined or empty, but its required for neon adapter.");
+    }
+
+    const pool = new NeonPool({ connectionString })
+    return new prismaNeon.PrismaNeon(pool)
 }
 
 main().catch(console.error)

query-engine/driver-adapters/js/connector-test-kit-executor/src/qe.ts

@@ -28,14 +28,18 @@ export function initQueryEngine(adapter: ErrorCapturingDriverAdapter, datamodel:
         ignoreEnvVarErrors: false,
     }
 
+
     const logCallback = (event: any) => {
         const parsed = JSON.parse(event)
         if (parsed.is_query) {
             queryLogCallback(parsed.query)
         }
-        console.error("[nodejs] ", parsed)
+
+        const level = process.env.LOG_LEVEL ?? ''
+        if (level.toLowerCase() == 'debug') {
+            console.error("[nodejs] ", parsed)
+        }
     }
-    const engine = new QueryEngine(queryEngineOptions, logCallback, adapter)
 
-    return engine
+    return new QueryEngine(queryEngineOptions, logCallback, adapter)
 }

query-engine/driver-adapters/js/smoke-test-js/src/client/neon.http.test.ts

@@ -4,7 +4,7 @@ import { PrismaNeonHTTP } from '@prisma/adapter-neon'
 import { smokeTestClient } from './client'
 
 describe('neon with @prisma/client', async () => {
-  const connectionString = `${process.env.JS_NEON_DATABASE_URL as string}`
+  const connectionString = process.env.JS_NEON_DATABASE_URL ?? ''
 
   const connection = neon(connectionString, {
     arrayMode: false,

query-engine/driver-adapters/js/smoke-test-js/src/client/neon.ws.test.ts

@@ -7,7 +7,7 @@ import { smokeTestClient } from './client'
 neonConfig.webSocketConstructor = WebSocket
 
 describe('neon with @prisma/client', async () => {
-  const connectionString = `${process.env.JS_NEON_DATABASE_URL as string}`
+  const connectionString = process.env.JS_NEON_DATABASE_URL ?? ''
 
   const pool = new Pool({ connectionString })
   const adapter = new PrismaNeon(pool)

query-engine/driver-adapters/js/smoke-test-js/src/client/pg.test.ts

@@ -4,10 +4,10 @@ import { PrismaPg } from '@prisma/adapter-pg'
 import { smokeTestClient } from './client'
 
 describe('pg with @prisma/client', async () => {
-  const connectionString = `${process.env.JS_PG_DATABASE_URL as string}`
+  const connectionString = process.env.JS_PG_DATABASE_URL ?? ''
 
   const pool = new pg.Pool({ connectionString })
   const adapter = new PrismaPg(pool)
-  
+
   smokeTestClient(adapter)
 })

query-engine/driver-adapters/js/smoke-test-js/src/client/planetscale.test.ts

@@ -4,10 +4,10 @@ import { describe } from 'node:test'
 import { smokeTestClient } from './client'
 
 describe('planetscale with @prisma/client', async () => {
-  const connectionString = `${process.env.JS_PLANETSCALE_DATABASE_URL as string}`
+  const connectionString = process.env.JS_PLANETSCALE_DATABASE_URL ?? ''
 
   const connnection = connect({ url: connectionString })
   const adapter = new PrismaPlanetScale(connnection)
-  
+
   smokeTestClient(adapter)
 })