feat: automated test isolation

db-service/lib/common/DatabaseIsolation.js

@@ -0,0 +1,195 @@
+const cds = require('@sap/cds')
+
+function getIsolate() {
+  const { options = {} } = cds
+  const fts = cds.requires.toggles && cds.resolve(cds.env.features.folders)
+  const src = [options.from || '*', ...(fts || [])]
+
+  const isolation = process.env.TRAVIS_JOB_ID || process.env.GITHUB_RUN_ID || require('os').userInfo().username || 'test_db'
+  const srchash = hash([cds.root, ...src.flat()].join('/'))
+
+  return {
+    src,
+    // Create one database for each overall test execution
+    database: 'D' + hash(isolation),
+    // Create one tenant for each model source definition
+    tenant: 'T' + srchash,
+    // Track source definition hash
+    source: srchash,
+  }
+}
+
+const hash = str => {
+  const { createHash } = require('crypto')
+  const hash = createHash('sha1')
+  hash.update(str)
+  return hash.digest('hex')
+}
+
+async function beforeWrite(dbs, isolate) {
+  const { ten } = dbs
+  const modified = isolate.modified = {}
+  ten.before(['*'], async (req) => {
+    if (
+      !req.query ||
+      req.query?.SELECT ||
+      (typeof req.query === 'string' && /^(BEGIN|COMMIT|ROLLBACK|SELECT)/i.test(req.query))
+    ) return // Ignore reading requests
+    if (req.target) modified[req.target.name] = true
+    if (req.tx._isolating) return req.tx._isolating
+    if (ten._writeable) return
+
+    // Add modification tracking for deep-queries internal calls
+    for (const fn of ['onSIMPLE', 'onUPDATE', 'onINSERT']) {
+      const org = ten[fn]
+      ten[fn] = function (req) {
+        if (req.query?.target) modified[req.query.target.name] = true
+        return org.apply(this, arguments)
+      }
+    }
+
+    ten._writeable = true
+    return (req.tx._isolating = req.tx.commit()
+      .then(() => getWriteTenant(dbs, isolate))
+      .then(() => req.tx.begin()))
+  })
+}
+
+async function deploy(dbs, isolate) {
+  console.log('DEPLOYING:', isolate.tenant)
+  const { ten } = dbs
+  await ten.tx(async () => {
+    try {
+      const src = isolate.src
+      const { options = {} } = cds
+      const m = await cds.load(src, options).then(cds.minify)
+      // options.schema_evolution = 'auto'
+      await cds.deploy(m).to(ten, options)
+    } catch (err) {
+      if (err.code === 'MODEL_NOT_FOUND' || err.code === 288) return
+      throw err
+    }
+  })
+}
+
+async function getReadTenant(dbs, isolate) {
+  const { dat, ten } = dbs
+  const { schemas } = dat.entities()
+  const deployTimeout = 120 // seconds
+
+  let isnew = false
+  try {
+    await dat.run(cds.ql.CREATE('schemas')).catch(() => { })
+    await dat.tx(async tx => {
+      await tx.run(DELETE.from(schemas).where`tenant=${isolate.tenant} and available=${false} and seconds_between(started, $now) > ${deployTimeout}`)
+      // If insert works the schema does not yet exist and this client has won the race and can deploy the contents
+      await tx.run(INSERT({ tenant: isolate.tenant, source: isolate.source, available: false, started: new Date() }).into(schemas))
+      isnew = true
+    })
+  } catch (err) {
+    const query = cds.ql`SELECT FROM ${schemas} {
+      (SELECT count(1) FROM ${schemas} WHERE tenant=${isolate.tenant} and available=${false} and seconds_between(started, $now) < ${deployTimeout}) as progress,
+      (SELECT count(1) FROM ${schemas} WHERE tenant=${isolate.tenant} and available=${true}) as available,
+    }`
+    // If the schema already exists wait for the row to be updated with available=true
+    await dat.tx(async tx => {
+      let available = 0
+      let progress = 0
+      while (progress && !available) [{ progress, available }] = await tx.run(query)
+    })
+  }
+
+  await ten.database(isolate)
+  await ten.tenant(isolate)
+
+  if (isnew) {
+    let err
+    await deploy(dbs, isolate).catch(e => { err = e })
+    await dat.tx(async tx => {
+      if (err) {
+        await tx.run(DELETE(schemas).where`tenant=${isolate.tenant}`)
+      } else {
+        await tx.run(UPDATE(schemas).where`tenant=${isolate.tenant}`.with({ available: true }))
+      }
+    })
+    if (err) throw err
+  }
+}
+
+async function getWriteTenant(dbs, isolate) {
+  const { ten, dat } = dbs
+  const { schemas } = dat.entities()
+  // await this.database(isolate)
+
+  let isnew = false
+  await dat.tx(async tx => {
+    const available = await tx.run(SELECT.from(schemas).where`tenant!=${isolate.tenant} and source=${isolate.source} and available=${true}`.forUpdate().limit(1))
+    if (available.length) {
+      const tenant = isolate.tenant = available[0].tenant
+      await tx.run(UPDATE(schemas).where`tenant=${tenant}`.with({ available: false, started: new Date() }))
+    } else {
+      isolate.tenant = 'T' + cds.utils.uuid()
+      await tx.run(INSERT({ tenant: isolate.tenant, source: isolate.source, available: false, started: new Date() }).into(schemas))
+      isnew = true
+    }
+  })
+
+  console.log('USING:', isolate.tenant)
+
+  await ten.database(isolate)
+  await ten.tenant(isolate)
+
+  if (isnew) await deploy(dbs, isolate)
+
+  // Release schema for follow up test runs
+  cds.on('shutdown', async () => {
+    try {
+      try {
+        // Clean tenant entities
+        await ten.tx(async tx => {
+          await tx.begin()
+          for (const entity in isolate.modified) {
+            const query = DELETE(entity).where`true=true`
+            if (!query.target._unresolved) await tx.onSIMPLE({ query }) // Skip deep delete
+          }
+          // UPSERT all data sources again
+          await cds.deploy.data(tx, tx.model, { schema_evolution: 'auto' })
+        })
+
+        await dat.run(UPDATE(schemas).where`tenant=${isolate.tenant}`.with({ available: true }))
+      } catch (err) {
+        // Try to cleanup broken tenant isolation
+        await ten.tenant(isolate, true)
+        // Remove cleaned up schema
+        await dat.run(DELETE(schemas).where`tenant=${isolate.tenant}`)
+      }
+    } catch (err) {
+      // if an shutdown handler throws an error it goes into an infinite loop
+      console.error(err)
+    }
+  })
+}
+
+module.exports = async function (db) {
+  const isolate = getIsolate()
+  // Just deploy when the database doesn't have isolation implementations available
+  if (typeof db.database !== 'function' || typeof db.tenant !== 'function') return deploy({ ten: db }, isolate)
+
+  const dbs = {
+    ten: db,
+    sys: await cds.connect.to('db_sys', { ...cds.requires.db, isolate: false }),
+    dat: await cds.connect.to('db_dat', {
+      ...cds.requires.db, isolate: false,
+      model: await cds.load(cds.utils.path.join(__dirname, 'database'))
+    }),
+  }
+
+  await dbs.dat.database(isolate)
+
+  await getReadTenant(dbs, isolate)
+
+  await db.database(isolate)
+  await db.tenant(isolate)
+
+  beforeWrite(dbs, isolate)
+}

db-service/lib/common/DatabaseService.js

@@ -7,8 +7,14 @@ const cds = require('@sap/cds')
 
 class DatabaseService extends cds.Service {
 
-  init() {
+  async init() {
     cds.on('shutdown', () => this.disconnect())
+    if (this.options.isolate !== false && (
+      Object.getOwnPropertyDescriptor(cds, 'test') || this.options.isolate
+    )) {
+      const isolation = require('./DatabaseIsolation')
+      await isolation(this)
+    }
     return super.init()
   }
 
@@ -47,7 +53,7 @@ class DatabaseService extends cds.Service {
    * transaction with `BEGIN`
    * @returns this
    */
-  async begin (min) {
+  async begin(min) {
     // We expect tx.begin() being called for an txed db service
     const ctx = this.context
 
@@ -129,9 +135,9 @@ class DatabaseService extends cds.Service {
   }
 
   // REVISIT: should happen automatically after a configurable time
-  async disconnect (tenant) {
+  async disconnect(tenant) {
     const tenants = tenant ? [tenant] : Object.keys(this.pools)
-    await Promise.all (tenants.map (async t => {
+    await Promise.all(tenants.map(async t => {
       const pool = this.pools[t]; if (!pool) return
       delete this.pools[t]
       await pool.drain()

db-service/lib/common/database.cds

@@ -0,0 +1,6 @@
+entity schemas {
+  key tenant    : String;
+      source    : String;
+      available : Boolean;
+      started   : Timestamp;
+}

hana/lib/HANAService.js

@@ -1304,7 +1304,7 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction})) AS NEW LE
 
   // Creates a new database using HDI container groups
   async database({ database }, clean = false) {
-    if (clean) {
+    if (this.options.credentials.__system__) {
       // Reset back to system credentials
       this.options.credentials = this.options.credentials.__system__
     }
@@ -1346,11 +1346,12 @@ SELECT ${mixing} FROM JSON_TABLE(SRC.JSON, '$' COLUMNS(${extraction})) AS NEW LE
   // This removes SCHEMA name conflicts when testing in the same system
   // Additionally this allows for deploying using the HDI procedures
   async tenant({ database, tenant }, clean = false) {
-    if (clean) {
+    if (this.options.credentials.__database__) {
       // Reset back to database credentials
       this.options.credentials = this.options.credentials.__database__
     }
 
+    tenant = tenant.replaceAll('-','')
     const creds = {
       containerGroup: database.toUpperCase(),
       usergroup: `${database}_USERS`.toUpperCase(),

hana/lib/scripts/container-tenant.sql

@@ -18,20 +18,24 @@ BEGIN SEQUENTIAL EXECUTION
 
   NO_PARAMS = SELECT * FROM _SYS_DI.T_NO_PARAMETERS;
 
-  IGNOREPARAMS =
-    SELECT 'IGNORE_DEPLOYED' AS KEY, 'TRUE' AS VALUE FROM DUMMY
-    UNION ALL
-    SELECT 'IGNORE_WORK' AS KEY, 'TRUE' AS VALUE FROM DUMMY;
-  CALL _SYS_DI#{{{GROUP}}}.DROP_CONTAINER(:SCHEMANAME, :IGNOREPARAMS, :RETURN_CODE, :REQUEST_ID, :MESSAGES);
-  ALL_MESSAGES = SELECT * FROM :MESSAGES;
-
   SELECT COUNT(*) INTO USER_EXISTS FROM SYS.USERS WHERE USER_NAME = :USERNAME;
-  IF :USER_EXISTS > 0 THEN
-    EXEC 'DROP USER ' || :USERNAME || ' CASCADE';
+  IF :CREATECONTAINER = FALSE THEN
+    IGNOREPARAMS =
+      SELECT 'IGNORE_DEPLOYED' AS KEY, 'TRUE' AS VALUE FROM DUMMY
+      UNION ALL
+      SELECT 'IGNORE_WORK' AS KEY, 'TRUE' AS VALUE FROM DUMMY;
+    CALL _SYS_DI#{{{GROUP}}}.DROP_CONTAINER(:SCHEMANAME, :IGNOREPARAMS, :RETURN_CODE, :REQUEST_ID, :MESSAGES);
+    ALL_MESSAGES = SELECT * FROM :MESSAGES;
+
+    IF :USER_EXISTS > 0 THEN
+      EXEC 'DROP USER ' || :USERNAME || ' CASCADE';
+    END IF;
   END IF;
 
   IF :CREATECONTAINER = TRUE THEN
-    EXEC 'CREATE USER ' || :USERNAME || ' PASSWORD ' || :USERPASS || ' NO FORCE_FIRST_PASSWORD_CHANGE SET USERGROUP "{{{GROUP}}}_USERS"';
+    IF :USER_EXISTS = 0 THEN
+      EXEC 'CREATE USER ' || :USERNAME || ' PASSWORD ' || :USERPASS || ' NO FORCE_FIRST_PASSWORD_CHANGE SET USERGROUP "{{{GROUP}}}_USERS"';
+    END IF;
     EXEC 'GRANT EXECUTE ON SYS.GET_INSUFFICIENT_PRIVILEGE_ERROR_DETAILS TO ' || :USERNAME;
 
     CALL _SYS_DI#{{{GROUP}}}.CREATE_CONTAINER(:SCHEMANAME, :NO_PARAMS, :RETURN_CODE, :REQUEST_ID, :MESSAGES);

hana/test/data/sap.capire.TestEntity.json

@@ -0,0 +1 @@
+[{"ID":1,"title":"1"},{"ID":2,"title":"2"},{"ID":3,"title":"3"},{"ID":4,"title":"4"}]