Skip to content

Commit b564842

Browse files
JulusianJulusian
committed
feat: integration test coverage of InMemoryMongoCollection
1 parent aeff950 commit b564842

20 files changed

Lines changed: 1461 additions & 193 deletions

‎meteor/__mocks__/integration-global-setup.js‎

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

‎meteor/__mocks__/integration-global-teardown.js‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
/**
2+
* Jest globalTeardown for the `integration` project: stop the shared MongoDB replica set started in
3+
* {@link file://./integration-global-setup.js}.
4+
*/
5+
module.exports = async () => {
6+
const replSet = globalThis.__MONGO_REPLSET__
7+
if (replSet) {
8+
await replSet.stop()
9+
globalThis.__MONGO_REPLSET__ = undefined
10+
}
11+
}

‎meteor/jest.config.js‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,14 +34,32 @@ const commonConfig = {
3434

3535
module.exports = {
3636
projects: [
37+
// Fast unit tests (the default inner loop). Integration tests (`*.integration.test.ts`, which spin up a
38+
// real MongoDB) are excluded here and live in the `integration` project below.
3739
Object.assign({}, commonConfig, {
40+
displayName: 'unit',
3841
testMatch: [
3942
'<rootDir>/server/__tests__/**/*.(spec|test).(ts|js)',
4043
'<rootDir>/server/**/__tests__/**/*.(spec|test).(ts|js)',
4144
'!.meteor/*.*',
4245
],
46+
testPathIgnorePatterns: ['/node_modules/', '\\.integration\\.test\\.(ts|js)$'],
4347
testEnvironment: 'node',
4448
}),
49+
// Real-MongoDB integration tests. A single in-memory replica set is booted once for the whole project
50+
// (see __mocks__/integration-global-setup.js) and reused across every file. Run on its own with
51+
// `jest --selectProjects integration`; `yarn unit` runs it alongside the `unit` project.
52+
Object.assign({}, commonConfig, {
53+
displayName: 'integration',
54+
testMatch: ['<rootDir>/server/**/*.integration.test.(ts|js)'],
55+
testEnvironment: 'node',
56+
globalSetup: './__mocks__/integration-global-setup.js',
57+
globalTeardown: './__mocks__/integration-global-teardown.js',
58+
// These tests wait on real change-stream / replica-set I/O, which is slower under CI load than the
59+
// 5s jest default. Give them some headroom (the reconnect test, which waits on real backoff delays,
60+
// overrides this inline).
61+
testTimeout: 20000,
62+
}),
4563
],
4664
coverageProvider: 'v8',
4765
coverageThreshold: {

‎meteor/scripts/dev-mongo.mjs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,12 @@ import { MongoClient } from 'mongodb'
66
// Use the same replica-set name Meteor uses for its dev mongod, so the same data dir works on both this
77
// branch (our dev-mongo) and an older branch (Meteor's bundled mongod)
88
const REPLSET_NAME = 'meteor'
9+
10+
/**
11+
* The mongod version we run everywhere: the dev instance (scripts/run.mjs) and the jest integration
12+
*/
13+
export const DEV_MONGO_VERSION = '7.0.16'
14+
915
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
1016

1117
/**

‎meteor/server/__tests__/integration/_integrationDb.ts‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/**
2+
* Shared helpers for the `integration` jest project: connect to the single MongoDB replica set booted once
3+
* by `__mocks__/integration-global-setup.js` (published via `process.env.MONGO_URL`), and hand out a fresh,
4+
* uniquely-named collection per test so files don't need to reboot mongo or clean up after each other.
5+
*/
6+
import { MongoClient, Collection, Document } from 'mongodb'
7+
import { randomBytes } from 'crypto'
8+
9+
/** Open a client against the shared replica set. Each test file should do this once in `beforeAll`. */
10+
export async function connectSharedMongoClient(): Promise<MongoClient> {
11+
const url = process.env.MONGO_URL
12+
if (!url) {
13+
throw new Error('MONGO_URL is not set - the integration globalSetup (shared replset) did not run')
14+
}
15+
// Match the production driver options (see server/collections/mongoConnection.ts).
16+
const client = new MongoClient(url, { ignoreUndefined: true })
17+
await client.connect()
18+
return client
19+
}
20+
21+
let collectionCounter = 0
22+
23+
/**
24+
* Create a fresh, uniquely-named collection (optionally seeded), for test isolation without rebooting mongo.
25+
* The name carries a random token so it is unique across all test files/workers sharing the one replset
26+
* (jest reuses worker processes, so a per-module counter alone would collide between files).
27+
*/
28+
export async function freshCollection<TDoc extends Document>(
29+
client: MongoClient,
30+
seed: TDoc[] = []
31+
): Promise<Collection<TDoc>> {
32+
const name = `itest_${randomBytes(6).toString('hex')}_${collectionCounter++}`
33+
const collection = client.db().collection<TDoc>(name)
34+
if (seed.length) await collection.insertMany(seed as any[])
35+
return collection
36+
}

‎meteor/server/__tests__/integration/_observeHelpers.ts‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/**
2+
* Helpers shared by the change-stream integration tests (the moved engine test and the observe-parity test).
3+
* Extracted from the original `engine.integration.test.ts` so both can drive the real change-stream
4+
* multiplexer the same way.
5+
*/
6+
import { MongoClient, Collection, ChangeStreamDocument } from 'mongodb'
7+
import { protectString, ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
8+
import { MongoQuery, MongoFieldSpecifier } from '@sofie-automation/corelib/dist/mongo'
9+
import { CollectionChangeFeed, createCollectionChangeStream } from '../../collections/changeStream/collectionChangeFeed'
10+
import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../../collections/changeStream/observeMultiplexer'
11+
12+
export interface TestDoc {
13+
_id: ProtectedString<any>
14+
name?: string
15+
val?: number
16+
studioId?: string
17+
}
18+
19+
export const id = (s: string): ProtectedString<any> => protectString<any>(s)
20+
21+
/**
22+
* Poll `predicate` until it is true or the timeout elapses (for awaiting async change-stream delivery).
23+
* Delivery is normally sub-second; the default allows for CI load while staying below the integration
24+
* project's jest `testTimeout`, so the poll (not jest) reports a genuine hang.
25+
*/
26+
export async function waitFor(predicate: () => boolean, timeoutMs = 15000): Promise<void> {
27+
const start = Date.now()
28+
while (!predicate()) {
29+
if (Date.now() - start > timeoutMs) throw new Error('waitFor timed out')
30+
await new Promise((r) => setTimeout(r, 20))
31+
}
32+
}
33+
34+
export interface Sink {
35+
added: Array<[any, any]>
36+
changed: Array<[any, any]>
37+
removed: any[]
38+
}
39+
40+
export function newSink(): Sink {
41+
return { added: [], changed: [], removed: [] }
42+
}
43+
44+
/** Build an {@link ObserveMultiplexer} backed by a real change stream + snapshot against `collection`. */
45+
export function makeMultiplexer(
46+
client: MongoClient,
47+
collection: Collection<any>,
48+
selector: MongoQuery<TestDoc>,
49+
projection: MongoFieldSpecifier<TestDoc> | undefined
50+
): ObserveMultiplexer<TestDoc> {
51+
const feed = new CollectionChangeFeed(
52+
collection.collectionName,
53+
() => createCollectionChangeStream(client, client.db(), collection.collectionName),
54+
() => undefined
55+
)
56+
const deps: ObserveMultiplexerDeps<TestDoc> = {
57+
snapshot: async () => collection.find(selector as any).toArray() as unknown as Promise<TestDoc[]>,
58+
subscribeFeed: (onChange, onReconnect) =>
59+
feed.subscribe(onChange as (c: ChangeStreamDocument<any>) => void, onReconnect),
60+
}
61+
return new ObserveMultiplexer<TestDoc>(selector, projection, deps, () => undefined)
62+
}
63+
64+
/** Subscribe an observeChanges sink to a multiplexer; teardown is driven by aborting `signal`. */
65+
export async function subscribeChanges(m: ObserveMultiplexer<TestDoc>, sink: Sink, signal: AbortSignal): Promise<void> {
66+
await m.addObserveChangesSubscriber(
67+
{
68+
added: (i: any, f: any) => sink.added.push([i, f]),
69+
changed: (i: any, f: any) => sink.changed.push([i, f]),
70+
removed: (i: any) => sink.removed.push(i),
71+
} as any,
72+
signal,
73+
false
74+
)
75+
}

meteor/server/collections/changeStream/__tests__/engine.integration.test.ts renamed to meteor/server/__tests__/integration/changeStreamEngine.integration.test.ts

Lines changed: 27 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -1,47 +1,31 @@
11
/**
22
* Integration tests for the change-stream observe engine against a REAL MongoDB.
33
*
4-
* These require a mongod binary (downloaded on first run by mongodb-memory-server) and start a single-node
5-
* replica set (change streams require a replica set). They are slower than the pure-unit tests and live in
6-
* the normal suite for now; if they become too slow they can be split into their own jest config.
4+
* Moved out of `server/collections/changeStream/__tests__/` into the shared `integration` jest project:
5+
* instead of booting its own `MongoMemoryReplSet` per file, it connects to the single replica set started
6+
* once by `__mocks__/integration-global-setup.js` (change streams require a replica set). Run via
7+
* `yarn unit:integration`.
78
*/
8-
import { MongoClient, Collection, ChangeStreamDocument } from 'mongodb'
9-
import { MongoMemoryReplSet } from 'mongodb-memory-server'
10-
import { protectString, ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
11-
import { MongoQuery, MongoFieldSpecifier } from '@sofie-automation/corelib/dist/mongo'
12-
import { CollectionChangeFeed, createCollectionChangeStream, ChangeStreamLike } from '../collectionChangeFeed'
13-
import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../observeMultiplexer'
14-
15-
interface TestDoc {
16-
_id: ProtectedString<any>
17-
name?: string
18-
val?: number
19-
studioId?: string
20-
}
21-
const id = (s: string) => protectString<any>(s)
22-
23-
async function waitFor(predicate: () => boolean, timeoutMs = 5000): Promise<void> {
24-
const start = Date.now()
25-
while (!predicate()) {
26-
if (Date.now() - start > timeoutMs) throw new Error('waitFor timed out')
27-
await new Promise((r) => setTimeout(r, 20))
28-
}
29-
}
9+
import { MongoClient, ChangeStreamDocument } from 'mongodb'
10+
import {
11+
CollectionChangeFeed,
12+
createCollectionChangeStream,
13+
ChangeStreamLike,
14+
} from '../../collections/changeStream/collectionChangeFeed'
15+
import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../../collections/changeStream/observeMultiplexer'
16+
import { connectSharedMongoClient, freshCollection } from './_integrationDb'
17+
import { TestDoc, id, waitFor, newSink, makeMultiplexer, subscribeChanges } from './_observeHelpers'
3018

3119
describe('change-stream engine (integration)', () => {
32-
let replSet: MongoMemoryReplSet
3320
let client: MongoClient
34-
let collectionCounter = 0
3521

3622
// Every observe in a test shares this one signal; afterEach aborts it (and beforeEach makes a fresh one),
3723
// which cascades to stopping the underlying change feed (otherwise a closed client would restart-loop forever).
3824
// Future: once we are using vitest, we can use the signal vitest provides, to ensure cleanup happens promptly during a timeout
3925
let observers: AbortController
4026

4127
beforeAll(async () => {
42-
replSet = await MongoMemoryReplSet.create({ replSet: { count: 1 } })
43-
client = new MongoClient(replSet.getUri(), { ignoreUndefined: true })
44-
await client.connect()
28+
client = await connectSharedMongoClient()
4529
}, 120000)
4630

4731
beforeEach(() => {
@@ -54,62 +38,13 @@ describe('change-stream engine (integration)', () => {
5438

5539
afterAll(async () => {
5640
await client?.close()
57-
await replSet?.stop()
5841
})
5942

60-
function makeMultiplexer(
61-
collection: Collection<any>,
62-
selector: MongoQuery<TestDoc>,
63-
projection: MongoFieldSpecifier<TestDoc> | undefined
64-
): ObserveMultiplexer<TestDoc> {
65-
const feed = new CollectionChangeFeed(
66-
collection.collectionName,
67-
() => createCollectionChangeStream(client, client.db(), collection.collectionName),
68-
() => undefined
69-
)
70-
const deps: ObserveMultiplexerDeps<TestDoc> = {
71-
snapshot: async () => collection.find(selector as any).toArray() as unknown as Promise<TestDoc[]>,
72-
subscribeFeed: (onChange, onReconnect) =>
73-
feed.subscribe(onChange as (c: ChangeStreamDocument<any>) => void, onReconnect),
74-
}
75-
return new ObserveMultiplexer<TestDoc>(selector, projection, deps, () => undefined)
76-
}
77-
78-
async function subscribeChanges(
79-
m: ObserveMultiplexer<TestDoc>,
80-
sink: {
81-
added: Array<[any, any]>
82-
changed: Array<[any, any]>
83-
removed: any[]
84-
}
85-
) {
86-
await m.addObserveChangesSubscriber(
87-
{
88-
added: (i: any, f: any) => sink.added.push([i, f]),
89-
changed: (i: any, f: any) => sink.changed.push([i, f]),
90-
removed: (i: any) => sink.removed.push(i),
91-
} as any,
92-
observers.signal,
93-
false
94-
)
95-
}
96-
97-
function newSink() {
98-
return { added: [] as Array<[any, any]>, changed: [] as Array<[any, any]>, removed: [] as any[] }
99-
}
100-
101-
async function freshCollection(seed: TestDoc[] = []): Promise<Collection<any>> {
102-
const name = `obs_test_${collectionCounter++}`
103-
const collection = client.db().collection(name)
104-
if (seed.length) await collection.insertMany(seed as any[])
105-
return collection
106-
}
107-
10843
test('initial snapshot, insert, update, delete', async () => {
109-
const collection = await freshCollection([{ _id: id('A'), name: 'a', val: 1 }])
44+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), name: 'a', val: 1 }])
11045
const sink = newSink()
111-
const m = makeMultiplexer(collection, {}, undefined)
112-
await subscribeChanges(m, sink)
46+
const m = makeMultiplexer(client, collection, {}, undefined)
47+
await subscribeChanges(m, sink, observers.signal)
11348

11449
expect(sink.added).toEqual([[id('A'), { name: 'a', val: 1 }]])
11550

@@ -127,10 +62,10 @@ describe('change-stream engine (integration)', () => {
12762
})
12863

12964
test('selector enter/leave and projection', async () => {
130-
const collection = await freshCollection()
65+
const collection = await freshCollection<TestDoc>(client)
13166
const sink = newSink()
132-
const m = makeMultiplexer(collection, { studioId: 's1' }, { studioId: 1, val: 1 })
133-
await subscribeChanges(m, sink)
67+
const m = makeMultiplexer(client, collection, { studioId: 's1' }, { studioId: 1, val: 1 })
68+
await subscribeChanges(m, sink, observers.signal)
13469

13570
// insert outside selector → nothing; then move it in → added
13671
await collection.insertOne({ _id: id('X'), studioId: 's2', val: 1, name: 'secret' } as any)
@@ -152,7 +87,7 @@ describe('change-stream engine (integration)', () => {
15287
})
15388

15489
test('two observers with identical args share one change stream', async () => {
155-
const collection = await freshCollection([{ _id: id('A'), val: 1 }])
90+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), val: 1 }])
15691
let watchCount = 0
15792

15893
const feed = new CollectionChangeFeed(
@@ -192,11 +127,11 @@ describe('change-stream engine (integration)', () => {
192127
})
193128

194129
test('observe() delivers full documents on add/change/remove', async () => {
195-
const collection = await freshCollection([{ _id: id('A'), val: 1 }])
130+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), val: 1 }])
196131
const added: any[] = []
197132
const changed: Array<[any, any]> = []
198133
const removed: any[] = []
199-
const m = makeMultiplexer(collection, {}, undefined)
134+
const m = makeMultiplexer(client, collection, {}, undefined)
200135
await m.addObserveSubscriber(
201136
{
202137
added: (d: any) => added.push(d),
@@ -222,10 +157,10 @@ describe('change-stream engine (integration)', () => {
222157
})
223158

224159
test('a replaceOne is observed as a changed transition', async () => {
225-
const collection = await freshCollection([{ _id: id('A'), name: 'a', val: 1 }])
160+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), name: 'a', val: 1 }])
226161
const sink = newSink()
227-
const m = makeMultiplexer(collection, {}, undefined)
228-
await subscribeChanges(m, sink)
162+
const m = makeMultiplexer(client, collection, {}, undefined)
163+
await subscribeChanges(m, sink, observers.signal)
229164
expect(sink.added).toEqual([[id('A'), { name: 'a', val: 1 }]])
230165

231166
await collection.replaceOne({ _id: id('A') }, { name: 'b', val: 1 } as any)
@@ -234,7 +169,7 @@ describe('change-stream engine (integration)', () => {
234169
})
235170

236171
test('recovers writes made while the change stream is down (reconnect re-sync)', async () => {
237-
const collection = await freshCollection([{ _id: id('A'), val: 1 }])
172+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), val: 1 }])
238173
const sink = newSink()
239174

240175
// A controllable stream: it wraps the REAL change stream, but lets the test inject a synthetic

0 commit comments

Comments
 (0)