fix!: improve test scenario reliability

Replace the internal event mechanism with server-side pause points to ensure test scenarios reproduce the same result on every run, regardless of network latency.

Author: urish

src/APIClient.ts

@@ -10,6 +10,7 @@ import type {
   APISimStartParams,
   PinReadResponse,
 } from './APITypes.js';
+import { PausePoint, type PausePointParams } from './PausePoint.js';
 import { readVersion } from './readVersion.js';
 
 const DEFAULT_SERVER = process.env.WOKWI_CLI_SERVER ?? 'wss://wokwi.com/api/ws/beta';
@@ -19,10 +20,12 @@ export class APIClient {
   private socket: WebSocket;
   private connectionAttempts = 0;
   private lastId = 0;
+  private lastPausePointId = 0;
   private closed = false;
   private _running = false;
   private _lastNanos = 0;
   private readonly apiEvents = new EventTarget();
+  private readonly pausePoints = new Map<string, PausePoint>();
   private readonly pendingCommands = new Map<
     string,
     [(result: any) => void, (error: Error) => void]
@@ -139,9 +142,9 @@ export class APIClient {
     return await this.sendCommand('sim:pause');
   }
 
-  async simResume(pauseAfter?: number, { waitForBytes }: { waitForBytes?: number[] } = {}) {
+  async simResume(pauseAfter?: number) {
     this._running = true;
-    return await this.sendCommand('sim:resume', { pauseAfter, waitForBytes });
+    return await this.sendCommand('sim:resume', { pauseAfter });
   }
 
   async simRestart({ pause }: { pause?: boolean } = {}) {
@@ -162,6 +165,15 @@ export class APIClient {
     });
   }
 
+  get pausedPromise() {
+    if (!this._running) {
+      return Promise.resolve();
+    }
+    return new Promise<APIEvent<any>>((resolve) => {
+      this.listen('sim:pause', resolve, { once: true });
+    });
+  }
+
   serialMonitorWritable() {
     return new Writable({
       write: (chunk, encoding, callback) => {
@@ -196,6 +208,46 @@ export class APIClient {
     });
   }
 
+  async addPausePoint(params: PausePointParams, resume = false) {
+    const id = `pp${this.lastPausePointId++}_${params.type}`;
+    const commands = [this.sendCommand('pause-point:add', { id, ...params })];
+    if (resume && !this._running) {
+      commands.push(this.simResume());
+      this._running = true;
+    }
+    await Promise.all(commands);
+    const pausePoint = new PausePoint(id, params);
+    this.pausePoints.set(id, pausePoint);
+    return pausePoint;
+  }
+
+  async removePausePoint(pausePoint: PausePoint) {
+    if (this.pausePoints.has(pausePoint.id)) {
+      this.pausePoints.delete(pausePoint.id);
+      await this.sendCommand('pause-point:remove', { id: pausePoint.id });
+      return true;
+    }
+    return false;
+  }
+
+  async atNanos(nanos: number) {
+    const pausePoint = await this.addPausePoint({ type: 'time-absolute', nanos });
+    await pausePoint.promise;
+  }
+
+  async delay(nanos: number) {
+    const pausePoint = await this.addPausePoint({ type: 'time-relative', nanos }, true);
+    await pausePoint.promise;
+  }
+
+  async waitForSerialBytes(bytes: number[] | Uint8Array) {
+    if (bytes instanceof Uint8Array) {
+      bytes = Array.from(bytes);
+    }
+    const pausePoint = await this.addPausePoint({ type: 'serial-bytes', bytes }, true);
+    await pausePoint.promise;
+  }
+
   async sendCommand<T = unknown>(command: string, params?: any) {
     return await new Promise<T>((resolve, reject) => {
       const id = this.lastId++;
@@ -251,6 +303,12 @@ export class APIClient {
   processEvent(message: APIEvent) {
     if (message.event === 'sim:pause') {
       this._running = false;
+      const pausePointId: string = message.payload.pausePoint;
+      const pausePoint = this.pausePoints.get(pausePointId);
+      if (pausePoint) {
+        pausePoint.resolve(message.payload.pausePointInfo);
+        this.pausePoints.delete(pausePointId);
+      }
     }
     this._lastNanos = message.nanos;
     this.apiEvents.dispatchEvent(new CustomEvent<APIEvent>(message.event, { detail: message }));

src/EventManager.spec.ts

@@ -0,0 +1,31 @@
+export type PausePointType = 'time-relative' | 'time-absolute' | 'serial-bytes';
+
+export interface ITimePausePoint {
+  type: 'time-relative' | 'time-absolute';
+  nanos: number;
+}
+
+export interface ISerialBytesPausePoint {
+  type: 'serial-bytes';
+  bytes: number[];
+}
+
+export type PausePointParams = ISerialBytesPausePoint | ITimePausePoint;
+
+export class PausePoint<T = any> {
+  private _resolve!: (info: T) => void;
+  readonly promise: Promise<T>;
+
+  constructor(
+    readonly id: string,
+    readonly params: PausePointParams,
+  ) {
+    this.promise = new Promise<T>((resolve) => {
+      this._resolve = resolve;
+    });
+  }
+
+  resolve(info: T) {
+    this._resolve(info);
+  }
+}

src/EventManager.ts

@@ -0,0 +1,9 @@
+export class SimulationTimeoutError extends Error {
+  constructor(
+    public readonly exitCode: number,
+    message: string,
+  ) {
+    super(message);
+    this.name = 'SimulationTimeoutError';
+  }
+}

src/PausePoint.ts

@@ -1,6 +1,5 @@
 import chalkTemplate from 'chalk-template';
 import type { APIClient } from './APIClient.js';
-import type { EventManager } from './EventManager.js';
 
 export interface IScenarioCommand {
   /** Validates the input to the command. Throws an exception of the input is not valid */
@@ -28,12 +27,12 @@ export class TestScenario {
   private stepIndex = 0;
   private client?: APIClient;
 
+  private serialNewLine = false;
+  private skipSerialNewline = false;
+
   readonly handlers: Record<string, IScenarioCommand> = {};
 
-  constructor(
-    readonly scenario: IScenarioDefinition,
-    readonly eventManager: EventManager,
-  ) {}
+  constructor(readonly scenario: IScenarioDefinition) {}
 
   registerCommands(commands: Record<string, IScenarioCommand>) {
     Object.assign(this.handlers, commands);
@@ -77,9 +76,6 @@ export class TestScenario {
     this.stepIndex = 0;
     this.client = client;
     for (const step of this.scenario.steps) {
-      if (client.running) {
-        void client.simPause();
-      }
       if (step.name) {
         this.log(chalkTemplate`{gray Executing step:} {yellow ${step.name}}`);
       }
@@ -102,17 +98,34 @@ export class TestScenario {
     process.exit(1);
   }
 
-  log(message: string) {
-    console.log(chalkTemplate`{cyan [${this.scenario.name}]}`, message);
+  /**
+   * The purpose of the method is to ensure that scenario logs don't appear in the middle of a line
+   * printed by the simulated project.
+   * @param bytes - incoming serial bytes
+   * @returns processed bytes (with leading newline removed if needed)
+   */
+  processSerialBytes(bytes: number[]) {
+    if (((bytes[0] === 13 && bytes[1] === 10) || bytes[0] === 10) && this.skipSerialNewline) {
+      bytes = bytes.slice(bytes[0] === 13 ? 2 : 1);
+    }
+    this.skipSerialNewline = false;
+    if (bytes.length > 0) {
+      this.serialNewLine = bytes[bytes.length - 1] === 10;
+    }
+    return bytes;
   }
 
-  fail(message: string) {
-    throw new Error(`[${this.client?.lastNanos}ns] ${message}`);
+  log(message: string) {
+    let prefix = '';
+    if (!this.serialNewLine) {
+      prefix = '\n';
+      this.skipSerialNewline = true;
+      this.serialNewLine = true;
+    }
+    console.log(prefix + chalkTemplate`{cyan [${this.scenario.name}]}`, message);
   }
 
-  async resume() {
-    await this.client?.simResume(
-      this.eventManager.timeToNextEvent >= 0 ? this.eventManager.timeToNextEvent : undefined,
-    );
+  fail(message: string) {
+    throw new Error(`[${this.client?.lastNanos.toFixed(0)}ns] ${message}`);
   }
 }