feat: Add execution_state to each test (#58)

packages/tests/runner.js

@@ -5,26 +5,83 @@ const {tests, testRunQueue, resultQueue, testQueues, connection} = require('./in
 const pool = createPool(process.env.DATABASE_URL)
 
 tests.forEach(test => {
-  require('./testWorker')(test.name, test.image, connection, resultQueue)
+  const worker = require('./testWorker')(test.name, test.image, connection, resultQueue)
+
+  worker.on('failed', (job, _failedReason) => {
+    console.log("a test worker failed: ", {job_id: job.id}, job.data)
+    pool.connect(c => 
+      c.any(sql`
+        UPDATE test_results 
+        SET execution_status = 'aborted'
+        WHERE id=${job.data.test_result_id}
+      `)
+    )
+    // TODO: Report failedReason to sentry?
+  })
+
+  // "Finally, you should add an error listener to your worker"
+  // https://docs.bullmq.io/guide/workers
+  worker.on('error', err => {
+    console.log('worker error', err)
+    // FIXME: Sentry error reporting
+  })
 })
 
-new Worker(testRunQueue.name, async ({data: {arguments, test_run_id}}) => {
+const testRunner = new Worker(testRunQueue.name, async ({data: {arguments, test_run_id}}) => {
   console.log("TestRunQueue", {arguments, test_run_id})
-  await Promise.all(
-    testQueues.map(tq =>
-        tq.add(`${arguments.host}`, {arguments, test_run_id})
+
+  // Insert a row for each test we are about to run
+  const ids = await pool.connect(async c => {
+    const values = tests.map(t => [test_run_id, t.name, 'pending'])
+
+    const rows = await c.any(sql`
+      INSERT INTO test_results (test_run_id, test_name, execution_status)
+      SELECT * 
+      FROM ${sql.unnest(values, ['int8', 'text', 'execution_states'])}
+      RETURNING id, test_name
+    `)
+    const lookup = rows.reduce(
+      (acc, next) => ({...acc, [next.test_name]: next.id}),
+      {}
     )
+
+    return lookup
+  })
+
+  // Schedule each test and send its result_id (primary key on test_results) along
+  await Promise.all(
+    testQueues.map(tq => tq.add(`${arguments.host}`, {arguments, test_run_id, test_result_id: ids[tq.name]}))
   )
 }, {connection})
 
-new Worker(resultQueue.name, async ({data}) => {
+testRunner.on('failed', (job, failedReason) => {
+  console.log('testRunner failed!', {job_id: job.id}, failedReason)
+})
+
+testRunner.on('error', err => {
+  console.log('testRunner error!', err)
+  // FIXME: Sentry error reporting
+})
+
+const resultWorker = new Worker(resultQueue.name, async ({data}) => {
   console.log("ResultQueue", data)
-  const { test_run_id, test_name, test_output } = data
+  const { test_result_id, test_output } = data
 
   await pool.connect(async (connection) => {
     await connection.any(sql`
-      INSERT INTO test_results (test_run_id, test_name, test_output)
-      VALUES (${test_run_id}, ${test_name}, ${JSON.stringify(test_output)})
+      UPDATE test_results 
+      SET execution_status = 'completed',
+          test_output = ${JSON.stringify(test_output)}
+      WHERE test_results.id = ${test_result_id} 
     `)
   })
 }, {connection})
+
+resultWorker.on('failed', (job, failedReason) => {
+  console.log('resultWorker failed!', {job_id: job.id}, failedReason)
+})
+
+resultWorker.on('error', err => {
+  console.log('resultWorker error!', err)
+  // FIXME: Sentry error reporting
+})

packages/tests/testWorker.js

@@ -28,7 +28,7 @@ function extractResult(logs) {
 module.exports = (test_name, image, connection, resultQueue) => new Worker(test_name, async (job) => {
   console.log(`Starting ${test_name}`)
   await job.log(`Starting ${job.name}`)
-  const {id, data: { arguments, test_run_id }} = job
+  const {id, data: { arguments, test_run_id, test_result_id }} = job
   const podName = `${test_name}-${test_run_id}`
   const containerArguments = [
       arguments.host,
@@ -43,7 +43,7 @@ module.exports = (test_name, image, connection, resultQueue) => new Worker(test_
     await job.log(`Pod logs: ${logs}`)
 
     const test_output = extractResult(logs)
-    const output = { test_run_id, test_name, test_output }
+    const output = { test_run_id, test_result_id, test_name, test_output }
 
     await resultQueue.add(job.name, output);
   } catch (err) {
@@ -59,6 +59,7 @@ module.exports = (test_name, image, connection, resultQueue) => new Worker(test_
     console.log(`Error: ${err}`)
     throw err;
   } finally {
+    // await to avoid causing an unhandled promise rejection error
     await deleteTest(podName)
   }
 }, {connection})

packages/web/src/db/queries/tests.js

@@ -27,7 +27,8 @@ function getTestResultByID(test_run_id) {
   return sql`
     SELECT
       res.test_name,
-      res.test_output
+      res.test_output,
+      res.execution_status
     FROM test_results res
     WHERE
       res.test_run_id = ${test_run_id}

packages/web/src/db/schema.sql

@@ -67,6 +67,8 @@ CREATE TABLE account_domains (
 ------------------------------------
 CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
 
+CREATE TYPE execution_states AS ENUM ('pending', 'aborted', 'completed');
+
 CREATE TABLE test_runs (
 	id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
 	public_id uuid NOT NULL UNIQUE DEFAULT uuid_generate_v4(),
@@ -81,6 +83,7 @@ CREATE TABLE test_results (
 	test_run_id BIGINT NOT NULL REFERENCES test_runs(id),
 	test_name TEXT NOT NULL,
 	test_output JSONB NOT NULL DEFAULT '{}'::jsonb,
+	execution_status execution_states NOT NULL DEFAULT 'pending',
 
 	UNIQUE(test_run_id, test_name)
 );

packages/web/src/tests.js

@@ -75,7 +75,11 @@ async function showTest(ctx) {
   await ctx.dbPool.connect(async (connection) => {
     const test = await connection.one(getTestRunByPublicID(id))
     const result = await connection.any(getTestResultByID(test.id))
-    if (result.length === 0) { return ctx.render('tests/loading') }
+
+    if (result.length === 0 || result.some(test => test.execution_status === 'pending')) { 
+      return ctx.render('tests/loading') 
+    }
+
     const allTestsPassed = !result.some(t => !t.test_output.passed)
     const domain = await connection.one(getDomainByID(test.domain_id))
     const groups = buildGroups(result)