mirror of
https://github.com/gravitational/teleport.git
synced 2026-10-12 02:15:01 +00:00
Fix flaky test in profileWatcher.test.ts (#64569)
* Fix flaky test in `profileWatcher.test.ts` * Use virtual watcher in debounce test
This commit is contained in:
@@ -16,6 +16,7 @@
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'node:events';
|
||||
import fs from 'node:fs/promises';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
@@ -26,7 +27,7 @@ import { wait } from 'shared/utils/wait';
|
||||
import { makeRootCluster } from 'teleterm/services/tshd/testHelpers';
|
||||
import { RootClusterUri, routing } from 'teleterm/ui/uri';
|
||||
|
||||
import { watchProfiles } from './profileWatcher';
|
||||
import { CreateFsWatcher, FsWatcher, watchProfiles } from './profileWatcher';
|
||||
|
||||
let tempDir: string;
|
||||
|
||||
@@ -136,6 +137,46 @@ function mockClusterStore(initial: { clusters: Cluster[] }) {
|
||||
};
|
||||
}
|
||||
|
||||
class VirtualWatcher extends EventEmitter implements FsWatcher {
|
||||
private closed = false;
|
||||
constructor(
|
||||
private readonly onEvent: () => void,
|
||||
signal?: AbortSignal
|
||||
) {
|
||||
super();
|
||||
signal?.addEventListener('abort', () => this.close(), { once: true });
|
||||
}
|
||||
|
||||
emitFileSystemEvent(): void {
|
||||
if (!this.closed) {
|
||||
this.onEvent();
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
this.closed = true;
|
||||
this.emit('close');
|
||||
}
|
||||
}
|
||||
|
||||
function makeVirtualWatcher(): {
|
||||
createFsWatcher: CreateFsWatcher;
|
||||
emitEvent: () => void;
|
||||
} {
|
||||
let watcher: VirtualWatcher | undefined;
|
||||
|
||||
return {
|
||||
createFsWatcher: ({ onEvent, signal }) => {
|
||||
watcher = new VirtualWatcher(onEvent, signal);
|
||||
return watcher;
|
||||
},
|
||||
emitEvent: () => watcher?.emitFileSystemEvent(),
|
||||
};
|
||||
}
|
||||
|
||||
test('yields an "added" change when new cluster appears', async () => {
|
||||
const tshDir = await makePerTestDir();
|
||||
const tshClientMock = await mockTshClient(tshDir, { clusters: [] });
|
||||
@@ -239,7 +280,7 @@ test('file system events are debounced', async () => {
|
||||
const tshDir = await makePerTestDir();
|
||||
const tshClientMock = await mockTshClient(tshDir, { clusters: [] });
|
||||
const clusterStoreMock = mockClusterStore({ clusters: [] });
|
||||
const handler = jest.fn().mockImplementation(() => Promise.resolve());
|
||||
const fsWatcher = makeVirtualWatcher();
|
||||
const testDebounceMs = 100;
|
||||
const watcher = watchProfiles({
|
||||
tshDirectory: tshDir,
|
||||
@@ -247,22 +288,43 @@ test('file system events are debounced', async () => {
|
||||
clusterStore: clusterStoreMock,
|
||||
signal: abortController.signal,
|
||||
debounceMs: testDebounceMs,
|
||||
createFsWatcher: fsWatcher.createFsWatcher,
|
||||
});
|
||||
|
||||
void (async () => {
|
||||
for await (let e of watcher) {
|
||||
await handler(e);
|
||||
}
|
||||
})();
|
||||
const cluster1 = makeRootCluster({ uri: '/clusters/new-cluster' });
|
||||
const cluster2 = makeRootCluster();
|
||||
jest.useFakeTimers();
|
||||
try {
|
||||
// Start watching before writing to the filesystem.
|
||||
const firstEventPromise = watcher.next();
|
||||
|
||||
const cluster = makeRootCluster();
|
||||
await tshClientMock.insertOrUpdateCluster(cluster1);
|
||||
fsWatcher.emitEvent();
|
||||
await tshClientMock.insertOrUpdateCluster(cluster2);
|
||||
fsWatcher.emitEvent();
|
||||
|
||||
// Insert two rapid events within debounce interval.
|
||||
await tshClientMock.insertOrUpdateCluster(cluster);
|
||||
await tshClientMock.insertOrUpdateCluster(cluster);
|
||||
// Wait slightly longer than debounce interval to ensure a single handler is called.
|
||||
await wait(2 * testDebounceMs);
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
await jest.advanceTimersByTimeAsync(testDebounceMs);
|
||||
expect(await firstEventPromise).toEqual({
|
||||
done: false,
|
||||
value: [
|
||||
{ op: 'added', cluster: cluster1 },
|
||||
{ op: 'added', cluster: cluster2 },
|
||||
],
|
||||
});
|
||||
|
||||
const secondEvent = Promise.race([
|
||||
watcher.next(),
|
||||
wait(testDebounceMs).then(() => 'timeout'),
|
||||
]);
|
||||
await jest.advanceTimersByTimeAsync(testDebounceMs);
|
||||
// Only one event is expected.
|
||||
expect(await secondEvent).toBe('timeout');
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
|
||||
// Cancel the watcher with the abort signal, it's blocked on `watcher.next()`.
|
||||
abortController.abort();
|
||||
});
|
||||
|
||||
test('no events are lost when handler is slow', async () => {
|
||||
|
||||
@@ -35,6 +35,23 @@ interface ClusterStore {
|
||||
getRootClusters(): Cluster[];
|
||||
}
|
||||
|
||||
export interface FsWatcher {
|
||||
on(event: 'close', listener: () => void): this;
|
||||
on(event: 'error', listener: (error: Error) => void): this;
|
||||
off(event: 'close', listener: () => void): this;
|
||||
off(event: 'error', listener: (error: Error) => void): this;
|
||||
close(): void;
|
||||
}
|
||||
|
||||
export type CreateFsWatcher = (options: {
|
||||
path: string;
|
||||
signal?: AbortSignal;
|
||||
onEvent(): void;
|
||||
}) => FsWatcher;
|
||||
|
||||
const createNodeFsWatcher: CreateFsWatcher = ({ path, signal, onEvent }) =>
|
||||
watch(path, { signal, recursive: true }, onEvent);
|
||||
|
||||
/**
|
||||
* Watches the specified `tshDirectory` for profile changes.
|
||||
* File system events are debounced with a default 200 ms delay.
|
||||
@@ -51,6 +68,7 @@ export async function* watchProfiles({
|
||||
debounceMs = 200,
|
||||
maxFileSystemEvents = 4096,
|
||||
signal,
|
||||
createFsWatcher = createNodeFsWatcher,
|
||||
}: {
|
||||
tshDirectory: string;
|
||||
tshClient: TshClient;
|
||||
@@ -66,6 +84,7 @@ export async function* watchProfiles({
|
||||
* */
|
||||
maxFileSystemEvents?: number;
|
||||
signal?: AbortSignal;
|
||||
createFsWatcher?: CreateFsWatcher;
|
||||
}): AsyncGenerator<ProfileChangeSet, void, void> {
|
||||
while (!signal?.aborted) {
|
||||
try {
|
||||
@@ -73,7 +92,8 @@ export async function* watchProfiles({
|
||||
tshDirectory,
|
||||
debounceMs,
|
||||
maxFileSystemEvents,
|
||||
signal
|
||||
signal,
|
||||
createFsWatcher
|
||||
)) {
|
||||
const clusters = await tshClient.listRootClusters();
|
||||
const newClusters = new Map(clusters.map(c => [c.uri, c]));
|
||||
@@ -180,7 +200,8 @@ async function* debounceWatch(
|
||||
path: string,
|
||||
debounceMs: number,
|
||||
maxFileSystemEvents: number,
|
||||
abortSignal: AbortSignal | undefined
|
||||
abortSignal: AbortSignal | undefined,
|
||||
createFsWatcher: CreateFsWatcher
|
||||
): AsyncGenerator<void> {
|
||||
let signal = Promise.withResolvers<void>();
|
||||
let closed = false;
|
||||
@@ -200,11 +221,11 @@ async function* debounceWatch(
|
||||
scheduleYield();
|
||||
};
|
||||
|
||||
const watcher = watch(
|
||||
const watcher = createFsWatcher({
|
||||
path,
|
||||
{ signal: abortSignal, recursive: true },
|
||||
onEvent
|
||||
);
|
||||
signal: abortSignal,
|
||||
onEvent,
|
||||
});
|
||||
|
||||
const closeHandler = () => {
|
||||
closed = true;
|
||||
|
||||
Reference in New Issue
Block a user