Handle Docker event streams and match both status and Action fields (#1230)

Author: cristianrgreco

packages/testcontainers/src/utils/test-helper.test.ts

@@ -0,0 +1,31 @@
+import { PassThrough } from "stream";
+import { waitForDockerEvent } from "./test-helper";
+
+describe("waitForDockerEvent", () => {
+  it("should resolve when action matches in ndjson stream", async () => {
+    const eventStream = new PassThrough();
+    const waitPromise = waitForDockerEvent(eventStream, "pull");
+
+    eventStream.write('{"Action":"create"}\n{"Action":"pull"}\n');
+
+    await expect(waitPromise).resolves.toBeUndefined();
+  });
+
+  it("should resolve when status matches in ndjson stream", async () => {
+    const eventStream = new PassThrough();
+    const waitPromise = waitForDockerEvent(eventStream, "pull");
+
+    eventStream.write('{"status":"pull"}\n');
+
+    await expect(waitPromise).resolves.toBeUndefined();
+  });
+
+  it("should resolve when action matches in json-seq stream", async () => {
+    const eventStream = new PassThrough();
+    const waitPromise = waitForDockerEvent(eventStream, "pull");
+
+    eventStream.write('\u001e{"Action":"pull"}\n');
+
+    await expect(waitPromise).resolves.toBeUndefined();
+  });
+});

packages/testcontainers/src/utils/test-helper.ts

@@ -125,18 +125,39 @@ export const composeContainerName = async (serviceName: string, index = 1): Prom
 
 export const waitForDockerEvent = async (eventStream: Readable, eventName: string, times = 1) => {
   let currentTimes = 0;
+  let pendingData = "";
+
+  const parseDockerEvent = (eventData: string): { status?: string; Action?: string } | undefined => {
+    try {
+      return JSON.parse(eventData);
+    } catch {
+      return undefined;
+    }
+  };
+
   return new Promise<void>((resolve) => {
-    eventStream.on("data", (data) => {
-      try {
-        if (JSON.parse(data).status === eventName) {
+    const onData = (data: string | Buffer) => {
+      // Docker events can be emitted as ndjson or json-seq; normalize both to line-delimited JSON.
+      pendingData += data.toString().split(String.fromCharCode(30)).join("\n");
+
+      const lines = pendingData.split("\n");
+      pendingData = lines.pop() ?? "";
+
+      for (const line of lines) {
+        const event = parseDockerEvent(line);
+        const action = event?.status ?? event?.Action;
+
+        if (action === eventName) {
           if (++currentTimes === times) {
+            eventStream.off("data", onData);
             resolve();
+            return;
           }
         }
-      } catch (err) {
-        // ignored
       }
-    });
+    };
+
+    eventStream.on("data", onData);
   });
 };