api/jobs: Create a couple tests for runJob and createDbIndexes

packages/api/src/app-router.ts

@@ -57,21 +57,16 @@ const PROM_BUNDLE_OPTS: promBundle.Opts = {
 const isTest =
   process.env.NODE_ENV === "test" || process.env.NODE_ENV === "development";
 
-export async function initClients(params: CliArgs, serviceName = "api") {
+export async function initDb(params: CliArgs, serviceName = "api") {
   const {
     postgresUrl,
     postgresReplicaUrl,
     postgresConnPoolSize: pgPoolSize,
     postgresJobsConnPoolSize: pgJobsPoolSize,
     postgresCreateTables: createTablesOnDb,
-    defaultCacheTtl,
     ownRegion,
-    stripeSecretKey,
-    amqpUrl,
-    amqpTasksExchange,
   } = params;
 
-  // Storage init
   const appName = ownRegion ? `${ownRegion}-${serviceName}` : serviceName;
   const pgBaseParams: PostgresParams = {
     postgresUrl,
@@ -83,11 +78,25 @@ export async function initClients(params: CliArgs, serviceName = "api") {
     { ...pgBaseParams, poolMaxSize: pgPoolSize },
     { ...pgBaseParams, poolMaxSize: pgJobsPoolSize, appName: `${appName}-jobs` }
   );
+  return { db, jobsDb, store };
+}
+
+export async function initClients(params: CliArgs, serviceName = "api") {
+  const {
+    defaultCacheTtl,
+    ownRegion,
+    stripeSecretKey,
+    amqpUrl,
+    amqpTasksExchange,
+  } = params;
+
+  const { db, jobsDb, store } = await initDb(params, serviceName);
   if (defaultCacheTtl > 0) {
     cache.init({ stdTTL: defaultCacheTtl });
   }
 
   // RabbitMQ
+  const appName = ownRegion ? `${ownRegion}-${serviceName}` : serviceName;
   const queue: Queue = amqpUrl
     ? await RabbitQueue.connect(amqpUrl, appName, amqpTasksExchange)
     : new NoopQueue();

packages/api/src/jobs/create-db-tables.test.ts

@@ -0,0 +1,29 @@
+import { db } from "../store";
+import params from "../test-params";
+import createDbTables from "./create-db-tables";
+
+describe("create-db-tables", () => {
+  it("should create tables and indexes in the DB", async () => {
+    // make sure it forces the creation of the tables, even with a bad config
+    await createDbTables({ ...params, postgresCreateTables: false });
+
+    const res = await db.query(
+      "SELECT indexname FROM pg_indexes WHERE indexname NOT LIKE '%_pkey'"
+    );
+    const indexes = res.rows?.map((r: any) => r.indexname).sort();
+
+    // test just a couple indexes from each table or this would be too long
+    const testIndexes = [
+      "stream_isActive",
+      "task_inputAssetId",
+      "users_email",
+      "webhook_log_userId",
+      "session_parentId",
+      "asset_source_url",
+      "api_token_projectId",
+    ];
+    for (const index of testIndexes) {
+      expect(indexes).toContain(index);
+    }
+  });
+});

packages/api/src/jobs/create-db-tables.ts

@@ -1,9 +1,9 @@
-import { initClients } from "../app-router";
+import { initDb } from "../app-router";
 import { CliArgs } from "../parse-cli";
 
 export default async function createDbTables(config: CliArgs) {
-  await initClients(
-    { ...config, createDbTables: true },
+  await initDb(
+    { ...config, postgresCreateTables: true },
     "create-db-tables-job"
   );
 }

packages/api/src/jobs/index.ts

@@ -7,7 +7,7 @@ import updateUsage from "./update-usage";
 
 type JobFunc = (config: CliArgs) => Promise<void | { logContext?: string }>;
 
-const jobFuncs: Record<JobType, JobFunc> = {
+export const jobFuncs: Record<JobType, JobFunc> = {
   "active-cleanup": activeCleanup,
   "create-db-tables": createDbTables,
   "update-usage": updateUsage,

packages/api/src/jobs/jobs.test.ts

@@ -0,0 +1,41 @@
+import { jobFuncs, runJob } from ".";
+
+describe("runJob", () => {
+  let mockExit: jest.SpyInstance;
+  let jobMockFn: jest.Mock;
+
+  beforeAll(() => {
+    mockExit = jest
+      .spyOn(process, "exit")
+      .mockImplementation((() => {}) as any);
+  });
+
+  beforeEach(() => {
+    jobFuncs["create-db-tables"] = jobMockFn = jest.fn(() => Promise.resolve());
+  });
+
+  it("calls the corresponding job function", async () => {
+    await runJob("create-db-tables", { jobTimeoutSec: 1 } as any);
+    expect(jobMockFn).toHaveBeenCalledTimes(1);
+  });
+
+  it("exits with status 0 on success", async () => {
+    await runJob("create-db-tables", { jobTimeoutSec: 1 } as any);
+    expect(process.exit).toHaveBeenCalledWith(0);
+  });
+
+  it("respects timeout configuration", async () => {
+    jobMockFn.mockImplementation(() => new Promise(() => {})); // never resolved
+    const startTime = Date.now();
+    await runJob("create-db-tables", { jobTimeoutSec: 1 } as any);
+    const elapsedTime = Date.now() - startTime;
+    expect(process.exit).toHaveBeenCalledWith(1);
+    expect(elapsedTime).toBeGreaterThanOrEqual(1000);
+  });
+
+  it("catches errors and exits with status 1", async () => {
+    jobMockFn.mockRejectedValue(new Error("job failed"));
+    await runJob("create-db-tables", { jobTimeoutSec: 1 } as any);
+    expect(process.exit).toHaveBeenCalledWith(1);
+  });
+});

packages/api/src/test-params.ts

@@ -0,0 +1,67 @@
+/**
+ * This file is imported from all the integration tests. It boots up a server based on the provided argv.
+ */
+import fs from "fs-extra";
+import os from "os";
+import path from "path";
+import { v4 as uuid } from "uuid";
+
+import argParser from "./parse-cli";
+
+export const dbPath = path.resolve(os.tmpdir(), "livepeer", uuid());
+
+const clientId = "EXPECTED_AUDIENCE";
+const trustedDomain = "livepeer.org";
+const jwtAudience = "livepeer";
+const jwtSecret = "secret";
+// enable to test SendGrid integration
+const supportAddr: [string, string] = ["Livepeer Team", "angie@livepeer.org"];
+const sendgridTemplateId = "iamanid";
+const sendgridApiKey = "SG. iamanapikey";
+
+fs.ensureDirSync(dbPath);
+
+const params = argParser();
+// Secret code used for back-door DB access in test env
+
+// Some overrides... we want to run on a random port for parallel reasons
+export const testId = `test_${Date.now()}`;
+
+delete params.port;
+params.dbPath = dbPath;
+params.clientId = clientId;
+params.trustedDomain = trustedDomain;
+params.jwtAudience = jwtAudience;
+params.jwtSecret = jwtSecret;
+params.supportAddr = supportAddr;
+params.sendgridTemplateId = sendgridTemplateId;
+params.sendgridApiKey = sendgridApiKey;
+params.postgresUrl = `postgresql://postgres@127.0.0.1/${testId}`;
+params.recordObjectStoreId = "mock_store";
+params.vodObjectStoreId = "mock_vod_store";
+params.trustedIpfsGateways = [
+  "https://ipfs.example.com/ipfs/",
+  /https:\/\/.+\.ipfs-provider.io\/ipfs\//,
+];
+params.ingest = [
+  {
+    ingest: "rtmp://test/live",
+    playback: "https://test/hls",
+    playbackDirect: "https://test/hls",
+    base: "https://test",
+    baseDirect: "https://test",
+    origin: "http://test",
+  },
+];
+params.amqpUrl = `amqp://127.0.0.1:5672/${testId}`;
+if (!params.insecureTestToken) {
+  params.insecureTestToken = uuid();
+}
+params.listen = true;
+params.requireEmailVerification = true;
+params.livekitHost = "livekit";
+params.frontend = false;
+
+console.log(`test run parameters: ${JSON.stringify(params)}`);
+
+export default params;

packages/api/src/test-server.ts

@@ -2,71 +2,14 @@
  * This file is imported from all the integration tests. It boots up a server based on the provided argv.
  */
 import fs from "fs-extra";
-import { v4 as uuid } from "uuid";
-import path from "path";
-import os from "os";
 
 import makeApp, { AppServer } from "./index";
-import argParser from "./parse-cli";
 import { rabbitMgmt, startAuxTestServer } from "./test-helpers";
-
-const dbPath = path.resolve(os.tmpdir(), "livepeer", uuid());
-const clientId = "EXPECTED_AUDIENCE";
-const trustedDomain = "livepeer.org";
-const jwtAudience = "livepeer";
-const jwtSecret = "secret";
-// enable to test SendGrid integration
-const supportAddr: [string, string] = ["Livepeer Team", "angie@livepeer.org"];
-const sendgridTemplateId = "iamanid";
-const sendgridApiKey = "SG. iamanapikey";
-
-fs.ensureDirSync(dbPath);
-
-const params = argParser();
-// Secret code used for back-door DB access in test env
-
-// Some overrides... we want to run on a random port for parallel reasons
-const testId = `test_${Date.now()}`;
-delete params.port;
-params.dbPath = dbPath;
-params.clientId = clientId;
-params.trustedDomain = trustedDomain;
-params.jwtAudience = jwtAudience;
-params.jwtSecret = jwtSecret;
-params.supportAddr = supportAddr;
-params.sendgridTemplateId = sendgridTemplateId;
-params.sendgridApiKey = sendgridApiKey;
-params.postgresUrl = `postgresql://postgres@127.0.0.1/${testId}`;
-params.recordObjectStoreId = "mock_store";
-params.vodObjectStoreId = "mock_vod_store";
-params.trustedIpfsGateways = [
-  "https://ipfs.example.com/ipfs/",
-  /https:\/\/.+\.ipfs-provider.io\/ipfs\//,
-];
-params.ingest = [
-  {
-    ingest: "rtmp://test/live",
-    playback: "https://test/hls",
-    playbackDirect: "https://test/hls",
-    base: "https://test",
-    baseDirect: "https://test",
-    origin: "http://test",
-  },
-];
-params.amqpUrl = `amqp://127.0.0.1:5672/${testId}`;
-if (!params.insecureTestToken) {
-  params.insecureTestToken = uuid();
-}
-params.listen = true;
-params.requireEmailVerification = true;
-params.livekitHost = "livekit";
-params.frontend = false;
+import params, { dbPath, testId } from "./test-params";
 
 let server: AppServer & { host?: string };
 let catalystServer;
 
-console.log(`test run parameters: ${JSON.stringify(params)}`);
-
 async function setupServer() {
   await rabbitMgmt.createVhost(testId);