New test framework, extracted from #474

.github/workflows/ci.yml

@@ -40,7 +40,9 @@ jobs:
         with:
           node-version: ${{ matrix.node-version }}
       - run: yarn --frozen-lockfile
-      - run: yarn node --experimental-vm-modules node_modules/.bin/jest -i --ci
+      - run: yarn prepack
+      - run: yarn test:setupdb
+      - run: yarn test:only --ci
 
   lint:
     runs-on: ubuntu-latest
@@ -89,8 +91,9 @@ jobs:
         with:
           node-version: ${{ matrix.node-version }}
       - run: yarn --frozen-lockfile
-      # - run: yarn lint # No need to lint altschema
-      - run: yarn node --experimental-vm-modules node_modules/.bin/jest -i --ci
+      - run: yarn prepack
+      - run: yarn test:setupdb
+      - run: yarn test:only --ci
 
   database_updated:
     runs-on: ubuntu-latest

.gitignore

@@ -1,3 +1,4 @@
+/.git
 dist
 npm-debug.log*
 yarn-debug.log*

__tests__/events.test.ts

@@ -6,7 +6,7 @@ import deferred, { Deferred } from "../src/deferred";
 import { Task, TaskList, WorkerSharedOptions } from "../src/interfaces";
 import {
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
-  jobCount,
+  expectJobCount,
   reset,
   sleep,
   sleepUntil,
@@ -132,7 +132,7 @@ test("emits the expected events", () =>
       expect(eventCount("worker:release")).toEqual(CONCURRENCY);
       expect(eventCount("worker:stop")).toEqual(CONCURRENCY);
       expect(eventCount("pool:release")).toEqual(1);
-      expect(await jobCount(pgPool)).toEqual(0);
+      await expectJobCount(pgPool, 0);
     } finally {
       Object.values(jobPromises).forEach((p) => p?.resolve());
     }

__tests__/forbiddenFlags.test.ts

@@ -5,20 +5,14 @@ import {
   TaskList,
   WorkerSharedOptions,
 } from "../src/index";
-import {
-  getJobs,
-  reset,
-  TEST_CONNECTION_STRING,
-  withPgClient,
-  withPgPool,
-} from "./helpers";
+import { getJobs, reset, withPgClient, withPgPool } from "./helpers";
 
 const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
 
 const options: WorkerSharedOptions = {};
 
 test("supports the flags API", () =>
-  withPgClient(async (pgClient) => {
+  withPgClient(async (pgClient, { TEST_CONNECTION_STRING }) => {
     await reset(pgClient, options);
 
     // Schedule a job

__tests__/helpers.ts

@@ -1,6 +1,11 @@
+import { randomBytes } from "crypto";
 import { EventEmitter } from "events";
+import {
+  setupFakeTimers as realSetupFakeTimers,
+  sleep,
+  sleepUntil as baseSleepUntil,
+} from "jest-time-helpers";
 import * as pg from "pg";
-import { parse } from "pg-connection-string";
 
 import defer from "../src/deferred";
 import {
@@ -16,6 +21,8 @@ import { processSharedOptions } from "../src/lib";
 import { _allWorkerPools } from "../src/main";
 import { migrate } from "../src/migrate";
 
+export { DAY, HOUR, MINUTE, SECOND, sleep, WEEK } from "jest-time-helpers";
+
 declare global {
   namespace GraphileWorker {
     interface Tasks {
@@ -25,39 +32,65 @@ declare global {
   }
 }
 
-export {
-  DAY,
-  HOUR,
-  MINUTE,
-  SECOND,
-  sleep,
-  sleepUntil,
-  WEEK,
-} from "jest-time-helpers";
-import {
-  setupFakeTimers as realSetupFakeTimers,
-  sleepUntil,
-} from "jest-time-helpers";
-
 let fakeTimers: ReturnType<typeof realSetupFakeTimers> | null = null;
 export function setupFakeTimers() {
   fakeTimers = realSetupFakeTimers();
   return fakeTimers;
 }
 
+export function sleepUntil(condition: () => boolean, ms?: number) {
+  // Bump the default timeout from 2000ms for CI
+  return baseSleepUntil(condition, ms ?? 5000);
+}
+
 // Sometimes CI's clock can get interrupted (it is shared infra!) so this
 // extends the default timeout just in case.
-jest.setTimeout(15000);
+jest.setTimeout(20000);
 
 // process.env.GRAPHILE_LOGGER_DEBUG = "1";
 
-export const TEST_CONNECTION_STRING =
-  process.env.TEST_CONNECTION_STRING || "postgres:///graphile_worker_test";
+async function createTestDatabase() {
+  const id = randomBytes(8).toString("hex");
+  const PGDATABASE = `graphile_worker_test_${id}`;
+  {
+    const client = new pg.Client({ connectionString: `postgres:///template1` });
+    await client.connect();
+    await client.query(
+      `create database ${pg.Client.prototype.escapeIdentifier(
+        PGDATABASE,
+      )} with template = graphile_worker_testtemplate;`,
+    );
+    await client.end();
+  }
+  const TEST_CONNECTION_STRING = `postgres:///${PGDATABASE}`;
+  const PGHOST = process.env.PGHOST;
+  async function release() {
+    const client = new pg.Client({ connectionString: `postgres:///template1` });
+    await client.connect();
+    await client.query(
+      `drop database ${pg.Client.prototype.escapeIdentifier(PGDATABASE)};`,
+    );
+    await client.end();
+  }
 
-const parsed = parse(TEST_CONNECTION_STRING);
+  return {
+    TEST_CONNECTION_STRING,
+    PGHOST,
+    PGDATABASE,
+    release,
+  };
+}
+
+export let databaseDetails: Awaited<
+  ReturnType<typeof createTestDatabase>
+> | null = null;
 
-export const PGHOST = parsed.host || process.env.PGHOST;
-export const PGDATABASE = parsed.database || undefined;
+beforeAll(async () => {
+  databaseDetails = await createTestDatabase();
+});
+afterAll(async () => {
+  databaseDetails?.release();
+});
 
 export const GRAPHILE_WORKER_SCHEMA =
   process.env.GRAPHILE_WORKER_SCHEMA || "graphile_worker";
@@ -67,6 +100,7 @@ export const ESCAPED_GRAPHILE_WORKER_SCHEMA =
 export async function withPgPool<T>(
   cb: (pool: pg.Pool) => Promise<T>,
 ): Promise<T> {
+  const { TEST_CONNECTION_STRING } = databaseDetails!;
   const pool = new pg.Pool({
     connectionString: TEST_CONNECTION_STRING,
     max: 100,
@@ -85,12 +119,17 @@ afterEach(() => {
 });
 
 export async function withPgClient<T>(
-  cb: (client: pg.PoolClient) => Promise<T>,
+  cb: (
+    client: pg.PoolClient,
+    extra: {
+      TEST_CONNECTION_STRING: string;
+    },
+  ) => Promise<T>,
 ): Promise<T> {
   return withPgPool(async (pool) => {
     const client = await pool.connect();
     try {
-      return await cb(client);
+      return await cb(client, databaseDetails!);
     } finally {
       client.release();
     }
@@ -156,7 +195,18 @@ async function _reset(
   }
 }
 
-export async function jobCount(
+/**
+ * Counts the number of jobs currently in DB.
+ *
+ * If you have a pool, you may hit race conditions with this method, instead
+ * use `expectJobCount()` which will try multiple times to give time for
+ * multiple clients to synchronize.
+ */
+export async function jobCount(pgClient: pg.PoolClient): Promise<number> {
+  return _jobCount(pgClient);
+}
+
+async function _jobCount(
   pgPoolOrClient: pg.Pool | pg.PoolClient,
 ): Promise<number> {
   const {
@@ -327,3 +377,28 @@ export function withOptions<T>(
     }),
   );
 }
+
+/**
+ * Wait for the job count to match the expected count, handles
+ * issues with different connections to the database not
+ * reflecting the same data by retrying.
+ */
+export async function expectJobCount(
+  // NOTE: if you have a pgClient then you shouldn't need to
+  // use this - just call `jobCount()` directly since you're
+  // in the same client
+  pool: pg.Pool,
+  expectedCount: number,
+) {
+  let count: number = Infinity;
+  for (let i = 0; i < 8; i++) {
+    if (i > 0) {
+      await sleep(i * 50);
+    }
+    count = await _jobCount(pool);
+    if (count === expectedCount) {
+      break;
+    }
+  }
+  expect(count).toEqual(expectedCount);
+}

__tests__/jobsView.test.ts

@@ -3,14 +3,13 @@ import {
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
   makeSelectionOfJobs,
   reset,
-  TEST_CONNECTION_STRING,
   withPgClient,
 } from "./helpers";
 
 const options: WorkerSharedOptions = {};
 
 test("jobs view renders jobs", () =>
-  withPgClient(async (pgClient) => {
+  withPgClient(async (pgClient, { TEST_CONNECTION_STRING }) => {
     await reset(pgClient, options);
 
     const utils = await makeWorkerUtils({

__tests__/main.runTaskList.test.ts

@@ -6,7 +6,7 @@ import { Task, TaskList, WorkerSharedOptions } from "../src/interfaces";
 import { runTaskList } from "../src/main";
 import {
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
-  jobCount,
+  expectJobCount,
   reset,
   sleep,
   sleepUntil,
@@ -70,7 +70,7 @@ test("main will execute jobs as they come up, and exits cleanly", () =>
       await sleep(1);
       expect(finished).toBeTruthy();
       await workerPool.promise;
-      expect(await jobCount(pgPool)).toEqual(0);
+      await expectJobCount(pgPool, 0);
       expect(process.listeners("SIGTERM")).toHaveLength(0);
     } finally {
       Object.values(jobPromises).forEach((p) => p?.resolve());

__tests__/runner.helpers.getTaskName.test.ts

@@ -3,9 +3,9 @@ import { Pool, PoolClient } from "pg";
 import { DbJobSpec, Runner, RunnerOptions } from "../src/interfaces";
 import { run } from "../src/runner";
 import {
+  databaseDetails,
   ESCAPED_GRAPHILE_WORKER_SCHEMA,
   sleepUntil,
-  TEST_CONNECTION_STRING,
   withPgClient,
 } from "./helpers";
 
@@ -15,7 +15,7 @@ let runner: Runner | null = null;
 const JOB_COUNT = 10;
 beforeAll(() => {
   pgPool = new Pool({
-    connectionString: TEST_CONNECTION_STRING,
+    connectionString: databaseDetails!.TEST_CONNECTION_STRING,
     max: JOB_COUNT * 2 + 5,
   });
 });