Fix/test stability

config/test.json

@@ -24,5 +24,12 @@
         }
     },
     "infrastructure": { "httpServer": { "enabled": true } },
-    "modules": { "coreHttpApi": { "docs": { "enabled": true } } }
+    "modules": {
+        "coreHttpApi": { "docs": { "enabled": true } },
+        "webhooks": {
+            "enabled": false,
+            "webhooks": [{ "triggers": ["**"], "target": "test" }],
+            "targets": { "test": {} }
+        }
+    }
 }

src/index.ts

@@ -60,22 +60,12 @@ function applyAlias(variable: { key: string; value: any }) {
     }
 }
 
-function isNumeric(value: string) {
-    if (typeof value !== "string") return false;
-
-    return !isNaN(value as any) && !isNaN(parseFloat(value));
-}
-
 function parseString(value: string) {
-    if (value === "true") {
-        return true;
-    } else if (value === "false") {
-        return false;
-    } else if (isNumeric(value)) {
-        return parseFloat(value);
+    try {
+        return JSON.parse(value);
+    } catch (_) {
+        return value;
     }
-
-    return value;
 }
 
 async function run() {

test/attributes.test.ts

@@ -1,5 +1,5 @@
-import { ConnectorClient, ConnectorRelationshipAttribute } from "@nmshd/connector-sdk";
-import { Launcher } from "./lib/Launcher";
+import { ConnectorRelationshipAttribute } from "@nmshd/connector-sdk";
+import { ConnectorClientWithMetadata, Launcher } from "./lib/Launcher";
 import { QueryParamConditions } from "./lib/QueryParamConditions";
 import {
     createRepositoryAttribute,
@@ -12,8 +12,8 @@ import {
 import { ValidationSchema } from "./lib/validation";
 
 const launcher = new Launcher();
-let client1: ConnectorClient;
-let client2: ConnectorClient;
+let client1: ConnectorClientWithMetadata;
+let client2: ConnectorClientWithMetadata;
 let client1Address: string;
 let client2Address: string;
 
@@ -25,6 +25,11 @@ beforeAll(async () => {
 }, 30000);
 afterAll(() => launcher.stop());
 
+beforeEach(() => {
+    client1._eventBus?.reset();
+    client2._eventBus?.reset();
+});
+
 describe("Attributes", () => {
     test("should create a repository attribute", async () => {
         const createAttributeResponse = await client1.attributes.createRepositoryAttribute({

test/lib/Launcher.ts

@@ -1,70 +1,111 @@
+import { DataEvent } from "@js-soft/ts-utils";
 import { ConnectorClient } from "@nmshd/connector-sdk";
+import { Random, RandomCharacterRange } from "@nmshd/transport";
 import { ChildProcess, spawn } from "child_process";
+import express from "express";
+import { Server } from "http";
 import path from "path";
+import { MockEventBus } from "./MockEventBus";
 import getPort from "./getPort";
 import waitForConnector from "./waitForConnector";
 
+export type ConnectorClientWithMetadata = ConnectorClient & {
+    /* eslint-disable @typescript-eslint/naming-convention */
+    _metadata?: Record<string, string>;
+    _eventBus?: MockEventBus;
+    /* eslint-enable @typescript-eslint/naming-convention */
+};
+
 export class Launcher {
-    private readonly _processes: ChildProcess[] = [];
+    private readonly _processes: { connector: ChildProcess; webhookServer: Server | undefined }[] = [];
     private readonly apiKey = "xxx";
 
-    private spawnConnector(port: number, accountName: string) {
-        const env = process.env;
-        env["infrastructure:httpServer:port"] = port.toString();
-        env["infrastructure:httpServer:apiKey"] = this.apiKey;
-
-        const notDefinedEnvironmentVariables = ["NMSHD_TEST_BASEURL", "NMSHD_TEST_CLIENTID", "NMSHD_TEST_CLIENTSECRET"].filter((env) => !process.env[env]);
-        if (notDefinedEnvironmentVariables.length > 0) {
-            throw new Error(`Missing environment variable(s): ${notDefinedEnvironmentVariables.join(", ")}}`);
-        }
-
-        env["transportLibrary:baseUrl"] = process.env["NMSHD_TEST_BASEURL"];
-        env["transportLibrary:platformClientId"] = process.env["NMSHD_TEST_CLIENTID"];
-        env["transportLibrary:platformClientSecret"] = process.env["NMSHD_TEST_CLIENTSECRET"];
-
-        env.NODE_CONFIG_ENV = "test";
-        env.DATABASE_NAME = accountName;
-
-        return spawn("node", ["dist/index.js"], {
-            env: { ...process.env, ...env },
-            cwd: path.resolve(`${__dirname}/../..`),
-            stdio: "inherit"
-        });
-    }
-
     public async launchSimple(): Promise<string> {
         const port = await getPort();
-        const accountName = this.randomString();
+        const accountName = await this.randomString();
 
-        this._processes.push(this.spawnConnector(port, accountName));
+        this._processes.push(await this.spawnConnector(port, accountName));
 
         await waitForConnector(port);
 
         return `http://localhost:${port}`;
     }
 
-    public async launch(count: number): Promise<ConnectorClient[]> {
-        const clients: ConnectorClient[] = [];
+    public async launch(count: number): Promise<ConnectorClientWithMetadata[]> {
+        const clients: ConnectorClientWithMetadata[] = [];
         const ports: number[] = [];
 
         for (let i = 0; i < count; i++) {
             const port = await getPort();
-            clients.push(ConnectorClient.create({ baseUrl: `http://localhost:${port}`, apiKey: this.apiKey }));
+            const connectorClient = ConnectorClient.create({ baseUrl: `http://localhost:${port}`, apiKey: this.apiKey }) as ConnectorClientWithMetadata;
+
+            connectorClient._eventBus = new MockEventBus();
+
+            clients.push(connectorClient);
             ports.push(port);
 
-            const accountName = this.randomString();
-            this._processes.push(this.spawnConnector(port, accountName));
+            const accountName = await this.randomString();
+            this._processes.push(await this.spawnConnector(port, accountName, connectorClient._eventBus));
         }
 
         await Promise.all(ports.map(waitForConnector));
         return clients;
     }
 
-    private randomString(): string {
-        return Math.random().toString(36).substring(7);
+    private async randomString(): Promise<string> {
+        return await Random.string(7, RandomCharacterRange.Alphabet);
+    }
+
+    private async spawnConnector(port: number, accountName: string, eventBus?: MockEventBus) {
+        const env = process.env;
+        env["infrastructure:httpServer:port"] = port.toString();
+        env["infrastructure:httpServer:apiKey"] = this.apiKey;
+
+        const notDefinedEnvironmentVariables = ["NMSHD_TEST_BASEURL", "NMSHD_TEST_CLIENTID", "NMSHD_TEST_CLIENTSECRET"].filter((env) => !process.env[env]);
+        if (notDefinedEnvironmentVariables.length > 0) {
+            throw new Error(`Missing environment variable(s): ${notDefinedEnvironmentVariables.join(", ")}}`);
+        }
+
+        env["transportLibrary:baseUrl"] = process.env["NMSHD_TEST_BASEURL"];
+        env["transportLibrary:platformClientId"] = process.env["NMSHD_TEST_CLIENTID"];
+        env["transportLibrary:platformClientSecret"] = process.env["NMSHD_TEST_CLIENTSECRET"];
+
+        env.NODE_CONFIG_ENV = "test";
+        env.DATABASE_NAME = accountName;
+
+        let webhookServer: Server | undefined;
+        if (eventBus) {
+            const webhookServerPort = await getPort();
+            env["modules:webhooks:enabled"] = "true";
+            env["modules:webhooks:targets:test:url"] = `http://localhost:${webhookServerPort}`;
+            webhookServer = this.startWebHookServer(webhookServerPort, eventBus);
+        }
+
+        return {
+            connector: spawn("node", ["dist/index.js"], {
+                env: { ...process.env, ...env },
+                cwd: path.resolve(`${__dirname}/../..`),
+                stdio: "inherit"
+            }),
+            webhookServer
+        };
+    }
+
+    private startWebHookServer(port: number, eventBus: MockEventBus): Server {
+        return express()
+            .use(express.json())
+            .use((req, res) => {
+                res.status(200).send("OK");
+
+                eventBus.publish(new DataEvent(req.body.trigger, req.body.data));
+            })
+            .listen(port);
     }
 
     public stop(): void {
-        this._processes.forEach((p) => p.kill());
+        this._processes.forEach((p) => {
+            p.connector.kill();
+            p.webhookServer?.close();
+        });
     }
 }

test/lib/MockEventBus.ts

@@ -0,0 +1,42 @@
+import { Event, EventEmitter2EventBus, getEventNamespaceFromObject } from "@js-soft/ts-utils";
+import { waitForEvent } from "./testUtils";
+
+export class MockEventBus extends EventEmitter2EventBus {
+    public publishedEvents: Event[] = [];
+
+    private publishPromises: Promise<any>[] = [];
+
+    public constructor() {
+        super((e) => console.log(e)); // eslint-disable-line no-console
+    }
+
+    public override publish(event: Event): void {
+        this.publishedEvents.push(event);
+
+        const namespace = getEventNamespaceFromObject(event);
+
+        if (!namespace) {
+            throw Error("The event needs a namespace. Use the EventNamespace-decorator in order to define a namespace for a event.");
+        }
+        this.publishPromises.push(this.emitter.emitAsync(namespace, event));
+    }
+
+    public async waitForEvent<TEvent extends Event>(subscriptionTarget: string, predicate?: (event: TEvent) => boolean): Promise<TEvent> {
+        const alreadyTriggeredEvents = this.publishedEvents.find((e) => e.namespace === subscriptionTarget && (!predicate || predicate(e as TEvent))) as TEvent | undefined;
+        if (alreadyTriggeredEvents) {
+            return alreadyTriggeredEvents;
+        }
+
+        const event = await waitForEvent(this, subscriptionTarget, predicate);
+        return event;
+    }
+
+    public async waitForRunningEventHandlers(): Promise<void> {
+        await Promise.all(this.publishPromises);
+    }
+
+    public reset(): void {
+        this.publishedEvents = [];
+        this.publishPromises = [];
+    }
+}

test/lib/testUtils.ts

@@ -1,4 +1,4 @@
-import { sleep } from "@js-soft/ts-utils";
+import { DataEvent, EventBus, SubscriptionTarget, sleep } from "@js-soft/ts-utils";
 import {
     ConnectorAttribute,
     ConnectorClient,
@@ -17,6 +17,7 @@ import {
 } from "@nmshd/connector-sdk";
 import fs from "fs";
 import { DateTime } from "luxon";
+import { ConnectorClientWithMetadata } from "./Launcher";
 import { ValidationSchema } from "./validation";
 
 export async function syncUntil(client: ConnectorClient, until: (syncResult: ConnectorSyncResult) => boolean): Promise<ConnectorSyncResult> {
@@ -57,32 +58,44 @@ export async function syncUntilHasMessages(client: ConnectorClient, expectedNumb
     return syncResult.messages;
 }
 
-export async function syncUntilHasMessageWithRequest(client: ConnectorClient, requestId: string): Promise<ConnectorMessage> {
+export async function syncUntilHasMessageWithRequest(client: ConnectorClientWithMetadata, requestId: string): Promise<ConnectorMessage> {
+    const isRequest = (content: any) => content["@type"] === "Request" && content.id === requestId;
     const filterRequestMessagesByRequestId = (syncResult: ConnectorSyncResult) => {
-        return syncResult.messages.filter((m: ConnectorMessage) => (m.content as any)["@type"] === "Request" && (m.content as any).id === requestId);
+        return syncResult.messages.filter((m: ConnectorMessage) => isRequest(m.content));
     };
+
     const syncResult = await syncUntil(client, (syncResult) => filterRequestMessagesByRequestId(syncResult).length !== 0);
+    await client._eventBus!.waitForEvent<DataEvent<any>>("consumption.messageProcessed", (e) => isRequest(e.data.message?.content));
+
     return filterRequestMessagesByRequestId(syncResult)[0];
 }
-export async function syncUntilHasMessageWithNotification(client: ConnectorClient, notificationId: string): Promise<ConnectorMessage> {
-    const filterRequestMessagesByRequestId = (syncResult: ConnectorSyncResult) => {
-        return syncResult.messages.filter((m: ConnectorMessage) => (m.content as any)["@type"] === "Notification" && (m.content as any).id === notificationId);
+export async function syncUntilHasMessageWithNotification(client: ConnectorClientWithMetadata, notificationId: string): Promise<ConnectorMessage> {
+    const isNotification = (content: any) => {
+        if (!content) {
+            return false;
+        }
+        return content["@type"] === "Notification" && content.id === notificationId;
     };
+
+    const filterRequestMessagesByRequestId = (syncResult: ConnectorSyncResult) => syncResult.messages.filter((m: ConnectorMessage) => isNotification(m.content));
+
     const syncResult = await syncUntil(client, (syncResult) => filterRequestMessagesByRequestId(syncResult).length !== 0);
+
+    await client._eventBus!.waitForEvent<DataEvent<any>>("consumption.messageProcessed", (e) => isNotification(e.data.message?.content));
+
     return filterRequestMessagesByRequestId(syncResult)[0];
 }
 
-export async function syncUntilHasMessageWithResponse(client: ConnectorClient, requestId: string): Promise<ConnectorMessage> {
+export async function syncUntilHasMessageWithResponse(client: ConnectorClientWithMetadata, requestId: string): Promise<ConnectorMessage> {
+    const isResponse = (content: any) => content["@type"] === "ResponseWrapper" && content.requestId === requestId;
     const filterRequestMessagesByRequestId = (syncResult: ConnectorSyncResult) => {
-        return syncResult.messages.filter((m: ConnectorMessage) => {
-            const matcher = (m.content as any)["@type"] === "ResponseWrapper" && (m.content as any).requestId === requestId;
-            return matcher;
-        });
+        return syncResult.messages.filter((m: ConnectorMessage) => isResponse(m.content));
     };
-    const syncResult = await syncUntil(client, (syncResult) => {
-        const length = filterRequestMessagesByRequestId(syncResult).length;
-        return length !== 0;
-    });
+
+    const syncResult = await syncUntil(client, (syncResult) => filterRequestMessagesByRequestId(syncResult).length !== 0);
+
+    await client._eventBus!.waitForEvent<DataEvent<any>>("consumption.messageProcessed", (e) => isResponse(e.data.message?.content));
+
     return filterRequestMessagesByRequestId(syncResult)[0];
 }
 
@@ -257,8 +270,8 @@ export async function createRepositoryAttribute(client: ConnectorClient, request
  * Returns the sender's own shared relationship attribute.
  */
 export async function executeFullCreateAndShareRelationshipAttributeFlow(
-    sender: ConnectorClient,
-    recipient: ConnectorClient,
+    sender: ConnectorClientWithMetadata,
+    recipient: ConnectorClientWithMetadata,
     attributeContent: Omit<ConnectorRelationshipAttribute, "owner" | "@type">
 ): Promise<ConnectorAttribute> {
     const senderIdentityInfoResult = await sender.account.getIdentityInfo();
@@ -315,8 +328,8 @@ export async function executeFullCreateAndShareRelationshipAttributeFlow(
  * Returns the sender's own shared identity attribute.
  */
 export async function executeFullCreateAndShareRepositoryAttributeFlow(
-    sender: ConnectorClient,
-    recipient: ConnectorClient,
+    sender: ConnectorClientWithMetadata,
+    recipient: ConnectorClientWithMetadata,
     attributeContent: ConnectorIdentityAttribute
 ): Promise<ConnectorAttribute> {
     const recipientIdentityInfoResult = await recipient.account.getIdentityInfo();
@@ -402,3 +415,34 @@ export function combinations<T>(...arrays: T[][]): T[][] {
     }
     return result;
 }
+
+export async function waitForEvent<TEvent>(
+    eventBus: EventBus,
+    subscriptionTarget: SubscriptionTarget<TEvent>,
+    assertionFunction?: (t: TEvent) => boolean,
+    timeout = 5000
+): Promise<TEvent> {
+    let subscriptionId: number;
+
+    const eventPromise = new Promise<TEvent>((resolve) => {
+        subscriptionId = eventBus.subscribe(subscriptionTarget, (event: TEvent) => {
+            if (assertionFunction && !assertionFunction(event)) return;
+
+            resolve(event);
+        });
+    });
+    if (!timeout) return await eventPromise.finally(() => eventBus.unsubscribe(subscriptionId));
+
+    let timeoutId: NodeJS.Timeout;
+    const timeoutPromise = new Promise<TEvent>((_resolve, reject) => {
+        timeoutId = setTimeout(
+            () => reject(new Error(`timeout exceeded for waiting for event ${typeof subscriptionTarget === "string" ? subscriptionTarget : subscriptionTarget.name}`)),
+            timeout
+        );
+    });
+
+    return await Promise.race([eventPromise, timeoutPromise]).finally(() => {
+        eventBus.unsubscribe(subscriptionId);
+        clearTimeout(timeoutId);
+    });
+}