Skip to content

Commit ce5fb67

Browse files
JulusianJulusian
committed
feat: integration test coverage of InMemoryMongoCollection
1 parent 005a1be commit ce5fb67

16 files changed

Lines changed: 1425 additions & 190 deletions

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
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+
const replSet = await MongoMemoryReplSet.create({ replSet: { count: 1 } })
17+
globalThis.__MONGO_REPLSET__ = replSet
18+
19+
// `parseMongoConnectionString` wants the database name in the URL path.
20+
process.env.MONGO_URL = replSet.getUri('meteor-integration-test')
21+
}

‎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: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,14 +34,28 @@ 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+
}),
4559
],
4660
coverageProvider: 'v8',
4761
coverageThreshold: {

‎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: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
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 { Meteor } from 'meteor/meteor'
8+
import { protectString, ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
9+
import { MongoQuery, MongoFieldSpecifier } from '@sofie-automation/corelib/dist/mongo'
10+
import { CollectionChangeFeed, createCollectionChangeStream } from '../../collections/changeStream/collectionChangeFeed'
11+
import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../../collections/changeStream/observeMultiplexer'
12+
13+
export interface TestDoc {
14+
_id: ProtectedString<any>
15+
name?: string
16+
val?: number
17+
studioId?: string
18+
}
19+
20+
export const id = (s: string): ProtectedString<any> => protectString<any>(s)
21+
22+
/** Poll `predicate` until it is true or the timeout elapses (for awaiting async change-stream delivery). */
23+
export 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+
}
30+
31+
export interface Sink {
32+
added: Array<[any, any]>
33+
changed: Array<[any, any]>
34+
removed: any[]
35+
}
36+
37+
export function newSink(): Sink {
38+
return { added: [], changed: [], removed: [] }
39+
}
40+
41+
/** Build an {@link ObserveMultiplexer} backed by a real change stream + snapshot against `collection`. */
42+
export function makeMultiplexer(
43+
client: MongoClient,
44+
collection: Collection<any>,
45+
selector: MongoQuery<TestDoc>,
46+
projection: MongoFieldSpecifier<TestDoc> | undefined
47+
): ObserveMultiplexer<TestDoc> {
48+
const feed = new CollectionChangeFeed(
49+
collection.collectionName,
50+
() => createCollectionChangeStream(client, client.db(), collection.collectionName),
51+
() => undefined
52+
)
53+
const deps: ObserveMultiplexerDeps<TestDoc> = {
54+
snapshot: async () => collection.find(selector as any).toArray() as unknown as Promise<TestDoc[]>,
55+
subscribeFeed: (onChange, onReconnect) =>
56+
feed.subscribe(onChange as (c: ChangeStreamDocument<any>) => void, onReconnect),
57+
}
58+
return new ObserveMultiplexer<TestDoc>(selector, projection, deps, () => undefined)
59+
}
60+
61+
/** Subscribe an observeChanges sink to a multiplexer, tracking the handle in `openHandles` for teardown. */
62+
export async function subscribeChanges(
63+
m: ObserveMultiplexer<TestDoc>,
64+
sink: Sink,
65+
openHandles: Meteor.LiveQueryHandle[]
66+
): Promise<Meteor.LiveQueryHandle> {
67+
const handle = await m.addObserveChangesSubscriber(
68+
{
69+
added: (i: any, f: any) => sink.added.push([i, f]),
70+
changed: (i: any, f: any) => sink.changed.push([i, f]),
71+
removed: (i: any) => sink.removed.push(i),
72+
} as any,
73+
false
74+
)
75+
openHandles.push(handle)
76+
return handle
77+
}

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

Lines changed: 27 additions & 93 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'
9+
import { MongoClient, ChangeStreamDocument } from 'mongodb'
1010
import { Meteor } from 'meteor/meteor'
11-
import { protectString, ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
12-
import { MongoQuery, MongoFieldSpecifier } from '@sofie-automation/corelib/dist/mongo'
13-
import { CollectionChangeFeed, createCollectionChangeStream, ChangeStreamLike } from '../collectionChangeFeed'
14-
import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../observeMultiplexer'
15-
16-
interface TestDoc {
17-
_id: ProtectedString<any>
18-
name?: string
19-
val?: number
20-
studioId?: string
21-
}
22-
const id = (s: string) => protectString<any>(s)
23-
24-
async function waitFor(predicate: () => boolean, timeoutMs = 5000): Promise<void> {
25-
const start = Date.now()
26-
while (!predicate()) {
27-
if (Date.now() - start > timeoutMs) throw new Error('waitFor timed out')
28-
await new Promise((r) => setTimeout(r, 20))
29-
}
30-
}
11+
import {
12+
CollectionChangeFeed,
13+
createCollectionChangeStream,
14+
ChangeStreamLike,
15+
} from '../../collections/changeStream/collectionChangeFeed'
16+
import { ObserveMultiplexer, ObserveMultiplexerDeps } from '../../collections/changeStream/observeMultiplexer'
17+
import { connectSharedMongoClient, freshCollection } from './_integrationDb'
18+
import { TestDoc, id, waitFor, newSink, makeMultiplexer, subscribeChanges } from './_observeHelpers'
3119

3220
describe('change-stream engine (integration)', () => {
33-
let replSet: MongoMemoryReplSet
3421
let client: MongoClient
35-
let collectionCounter = 0
3622

3723
// Every observe handle created in a test is tracked here and stopped in afterEach, which cascades to
3824
// stopping the underlying change feed (otherwise a closed client would restart-loop forever).
3925
const openHandles: Meteor.LiveQueryHandle[] = []
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
afterEach(() => {
@@ -51,63 +35,13 @@ describe('change-stream engine (integration)', () => {
5135

5236
afterAll(async () => {
5337
await client?.close()
54-
await replSet?.stop()
5538
})
5639

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

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

@@ -125,10 +59,10 @@ describe('change-stream engine (integration)', () => {
12559
})
12660

12761
test('selector enter/leave and projection', async () => {
128-
const collection = await freshCollection()
62+
const collection = await freshCollection<TestDoc>(client)
12963
const sink = newSink()
130-
const m = makeMultiplexer(collection, { studioId: 's1' }, { studioId: 1, val: 1 })
131-
await subscribeChanges(m, sink)
64+
const m = makeMultiplexer(client, collection, { studioId: 's1' }, { studioId: 1, val: 1 })
65+
await subscribeChanges(m, sink, openHandles)
13266

13367
// insert outside selector → nothing; then move it in → added
13468
await collection.insertOne({ _id: id('X'), studioId: 's2', val: 1, name: 'secret' } as any)
@@ -150,7 +84,7 @@ describe('change-stream engine (integration)', () => {
15084
})
15185

15286
test('two observers with identical args share one change stream', async () => {
153-
const collection = await freshCollection([{ _id: id('A'), val: 1 }])
87+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), val: 1 }])
15488
let watchCount = 0
15589

15690
const feed = new CollectionChangeFeed(
@@ -192,11 +126,11 @@ describe('change-stream engine (integration)', () => {
192126
})
193127

194128
test('observe() delivers full documents on add/change/remove', async () => {
195-
const collection = await freshCollection([{ _id: id('A'), val: 1 }])
129+
const collection = await freshCollection<TestDoc>(client, [{ _id: id('A'), val: 1 }])
196130
const added: any[] = []
197131
const changed: Array<[any, any]> = []
198132
const removed: any[] = []
199-
const m = makeMultiplexer(collection, {}, undefined)
133+
const m = makeMultiplexer(client, collection, {}, undefined)
200134
openHandles.push(
201135
await m.addObserveSubscriber(
202136
{
@@ -223,10 +157,10 @@ describe('change-stream engine (integration)', () => {
223157
})
224158

225159
test('a replaceOne is observed as a changed transition', async () => {
226-
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 }])
227161
const sink = newSink()
228-
const m = makeMultiplexer(collection, {}, undefined)
229-
await subscribeChanges(m, sink)
162+
const m = makeMultiplexer(client, collection, {}, undefined)
163+
await subscribeChanges(m, sink, openHandles)
230164
expect(sink.added).toEqual([[id('A'), { name: 'a', val: 1 }]])
231165

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

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

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

0 commit comments

Comments
 (0)