feat: integration test coverage of InMemoryMongoCollection

Author: Julusian

meteor/__mocks__/integration-global-setup.js

@@ -0,0 +1,23 @@
+const { MongoMemoryReplSet } = require('mongodb-memory-server')
+
+/**
+ * Jest globalSetup for the `integration` project: boot ONE in-memory MongoDB replica set, shared by every
+ * `*.integration.test.ts` file in the run. A replica set (not a standalone) is required because the
+ * change-stream observe engine relies on change streams.
+ *
+ * The instance handle is stashed on `globalThis` so {@link file://./integration-global-teardown.js} can stop
+ * it, and the connection string is published via `process.env.MONGO_URL`. Env vars set here are inherited by
+ * the forked jest workers, so both the test files' own `MongoClient`s and the production `getMongoDb()`
+ * (see `server/collections/mongoConnection.ts`) resolve to this same instance.
+ */
+module.exports = async () => {
+	process.env.TZ = 'UTC'
+
+	// eslint-disable-next-line n/file-extension-in-import -- Node's dynamic import() of a local ESM file needs the explicit extension
+	const { DEV_MONGO_VERSION } = await import('../scripts/dev-mongo.mjs')
+	const replSet = await MongoMemoryReplSet.create({ replSet: { count: 1 }, binary: { version: DEV_MONGO_VERSION } })
+	globalThis.__MONGO_REPLSET__ = replSet
+
+	// `parseMongoConnectionString` wants the database name in the URL path.
+	process.env.MONGO_URL = replSet.getUri('meteor-integration-test')
+}

meteor/__mocks__/integration-global-teardown.js

@@ -0,0 +1,11 @@
+/**
+ * Jest globalTeardown for the `integration` project: stop the shared MongoDB replica set started in
+ * {@link file://./integration-global-setup.js}.
+ */
+module.exports = async () => {
+	const replSet = globalThis.__MONGO_REPLSET__
+	if (replSet) {
+		await replSet.stop()
+		globalThis.__MONGO_REPLSET__ = undefined
+	}
+}

meteor/jest.config.js

@@ -34,14 +34,32 @@ const commonConfig = {
 
 module.exports = {
 	projects: [
+		// Fast unit tests (the default inner loop). Integration tests (`*.integration.test.ts`, which spin up a
+		// real MongoDB) are excluded here and live in the `integration` project below.
 		Object.assign({}, commonConfig, {
+			displayName: 'unit',
 			testMatch: [
 				'<rootDir>/server/__tests__/**/*.(spec|test).(ts|js)',
 				'<rootDir>/server/**/__tests__/**/*.(spec|test).(ts|js)',
 				'!.meteor/*.*',
 			],
+			testPathIgnorePatterns: ['/node_modules/', '\\.integration\\.test\\.(ts|js)$'],
 			testEnvironment: 'node',
 		}),
+		// Real-MongoDB integration tests. A single in-memory replica set is booted once for the whole project
+		// (see __mocks__/integration-global-setup.js) and reused across every file. Run on its own with
+		// `jest --selectProjects integration`; `yarn unit` runs it alongside the `unit` project.
+		Object.assign({}, commonConfig, {
+			displayName: 'integration',
+			testMatch: ['<rootDir>/server/**/*.integration.test.(ts|js)'],
+			testEnvironment: 'node',
+			globalSetup: './__mocks__/integration-global-setup.js',
+			globalTeardown: './__mocks__/integration-global-teardown.js',
+			// These tests wait on real change-stream / replica-set I/O, which is slower under CI load than the
+			// 5s jest default. Give them some headroom (the reconnect test, which waits on real backoff delays,
+			// overrides this inline).
+			testTimeout: 20000,
+		}),
 	],
 	coverageProvider: 'v8',
 	coverageThreshold: {

meteor/scripts/dev-mongo.mjs

@@ -6,6 +6,12 @@ import { MongoClient } from 'mongodb'
 // Use the same replica-set name Meteor uses for its dev mongod, so the same data dir works on both this
 // branch (our dev-mongo) and an older branch (Meteor's bundled mongod)
 const REPLSET_NAME = 'meteor'
+
+/**
+ * The mongod version we run everywhere: the dev instance (scripts/run.mjs) and the jest integration
+ */
+export const DEV_MONGO_VERSION = '7.0.16'
+
 const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
 
 /**

meteor/server/__tests__/integration/_integrationDb.ts

@@ -0,0 +1,36 @@
+/**
+ * Shared helpers for the `integration` jest project: connect to the single MongoDB replica set booted once
+ * by `__mocks__/integration-global-setup.js` (published via `process.env.MONGO_URL`), and hand out a fresh,
+ * uniquely-named collection per test so files don't need to reboot mongo or clean up after each other.
+ */
+import { MongoClient, Collection, Document } from 'mongodb'
+import { randomBytes } from 'crypto'
+
+/** Open a client against the shared replica set. Each test file should do this once in `beforeAll`. */
+export async function connectSharedMongoClient(): Promise<MongoClient> {
+	const url = process.env.MONGO_URL
+	if (!url) {
+		throw new Error('MONGO_URL is not set - the integration globalSetup (shared replset) did not run')
+	}
+	// Match the production driver options (see server/collections/mongoConnection.ts).
+	const client = new MongoClient(url, { ignoreUndefined: true })
+	await client.connect()
+	return client
+}
+
+let collectionCounter = 0
+
+/**
+ * Create a fresh, uniquely-named collection (optionally seeded), for test isolation without rebooting mongo.
+ * The name carries a random token so it is unique across all test files/workers sharing the one replset
+ * (jest reuses worker processes, so a per-module counter alone would collide between files).
+ */
+export async function freshCollection<TDoc extends Document>(
+	client: MongoClient,
+	seed: TDoc[] = []
+): Promise<Collection<TDoc>> {
+	const name = `itest_${randomBytes(6).toString('hex')}_${collectionCounter++}`
+	const collection = client.db().collection<TDoc>(name)
+	if (seed.length) await collection.insertMany(seed as any[])
+	return collection
+}

meteor/server/__tests__/integration/_observeHelpers.ts

@@ -0,0 +1,77 @@
+/**
+ * Helpers shared by the change-stream integration tests (the moved engine test and the observe-parity test).
+ * Extracted from the original `engine.integration.test.ts` so both can drive the real change-stream
+ * multiplexer the same way.
+ */
+import { MongoClient, Collection, ChangeStreamDocument } from 'mongodb'
+import { protectString, ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
+import { MongoQuery, MongoFieldSpecifier } from '@sofie-automation/corelib/dist/mongo'
+import { CollectionChangeFeed, createCollectionChangeStream } from '../../collections/changeStream/collectionChangeFeed'
+import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../../collections/changeStream/observeMultiplexer'
+import type { ObserveViewShape } from '@sofie-automation/corelib/dist/memoryCollection/observeView'
+
+export interface TestDoc {
+	_id: ProtectedString<any>
+	name?: string
+	val?: number
+	studioId?: string
+}
+
+export const id = (s: string): ProtectedString<any> => protectString<any>(s)
+
+/**
+ * Poll `predicate` until it is true or the timeout elapses (for awaiting async change-stream delivery).
+ * Delivery is normally sub-second; the default allows for CI load while staying below the integration
+ * project's jest `testTimeout`, so the poll (not jest) reports a genuine hang.
+ */
+export async function waitFor(predicate: () => boolean, timeoutMs = 15000): Promise<void> {
+	const start = Date.now()
+	while (!predicate()) {
+		if (Date.now() - start > timeoutMs) throw new Error('waitFor timed out')
+		await new Promise((r) => setTimeout(r, 20))
+	}
+}
+
+export interface Sink {
+	added: Array<[any, any]>
+	changed: Array<[any, any]>
+	removed: any[]
+}
+
+export function newSink(): Sink {
+	return { added: [], changed: [], removed: [] }
+}
+
+/** Build an {@link ObserveMultiplexer} backed by a real change stream + snapshot against `collection`. */
+export function makeMultiplexer(
+	client: MongoClient,
+	collection: Collection<any>,
+	selector: MongoQuery<TestDoc>,
+	projection: MongoFieldSpecifier<TestDoc> | undefined,
+	shape?: ObserveViewShape<TestDoc>
+): ObserveMultiplexer<TestDoc> {
+	const feed = new CollectionChangeFeed(
+		collection.collectionName,
+		() => createCollectionChangeStream(client, client.db(), collection.collectionName),
+		() => undefined
+	)
+	const deps: ObserveMultiplexerDeps<TestDoc> = {
+		snapshot: async () => collection.find(selector as any).toArray() as unknown as Promise<TestDoc[]>,
+		subscribeFeed: (onChange, onReconnect) =>
+			feed.subscribe(onChange as (c: ChangeStreamDocument<any>) => void, onReconnect),
+	}
+	return new ObserveMultiplexer<TestDoc>(selector, projection, shape, deps, () => undefined)
+}
+
+/** Subscribe an observeChanges sink to a multiplexer; teardown is driven by aborting `signal`. */
+export async function subscribeChanges(m: ObserveMultiplexer<TestDoc>, sink: Sink, signal: AbortSignal): Promise<void> {
+	await m.addObserveChangesSubscriber(
+		{
+			added: (i: any, f: any) => sink.added.push([i, f]),
+			changed: (i: any, f: any) => sink.changed.push([i, f]),
+			removed: (i: any) => sink.removed.push(i),
+		} as any,
+		signal,
+		false
+	)
+}