tests

Signed-off-by: Joe Bowbeer <joe.bowbeer@gmail.com>
dapr · joebowbeer · May 12, 2025 · May 13, 2025 · May 14, 2025 · May 15, 2025
commit 944ad324600d74d0672df5df7a2130cc5c14c2b3
@@ -11,45 +11,23 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// import { setTimeout } from "node:timers/promises";
-// import { GenericContainer, Network, Wait } from "testcontainers";
 import path from "node:path";
-import { Network } from "testcontainers";
+import { Network, TestContainers } from "testcontainers";
+import { DaprClient, DaprServer, LogLevel } from "@dapr/dapr";
 import { DAPR_RUNTIME_IMAGE, DaprContainer } from "./DaprContainer";
 
-// jest.setTimeout(120_000);
-
 describe("DaprContainer", () => {
-  it("should start and stop", async () => {
-    const network = await new Network().start();
-    const container = new DaprContainer("daprio/dapr:latest")
-      .withNetwork(network)
-      .withAppName("dapr-app")
-      .withAppPort(8081)
-      .withDaprLogLevel("debug")
-      .withAppChannelAddress("host.testcontainers.internal");
-    const startedContainer = await container.start();
-    expect(startedContainer.getHost()).toBeDefined();
-    expect(startedContainer.getHttpPort()).toBeDefined();
-    expect(startedContainer.getGrpcPort()).toBeDefined();
-    expect(startedContainer.getHttpEndpoint()).toBeDefined();
-    expect(startedContainer.getGrpcEndpoint()).toBeDefined();
-    await startedContainer.stop();
-    await network.stop();
-  }, 120_000);
   it("should load component from path", async () => {
     const componentPath = path.join(__dirname, "__fixtures__", "dapr-resources", "statestore.yaml");
     const dapr = new DaprContainer(DAPR_RUNTIME_IMAGE)
         .withAppName("dapr-app")
         .withAppPort(8081)
         .withComponentFromPath(componentPath)
         .withAppChannelAddress("host.testcontainers.internal");
-
     const components = dapr.getComponents();
     expect(components.length).toBe(1);
     const kvstore = components[0];
     expect(kvstore.getMetadata().length).toBeTruthy();
-
     const componentYaml = kvstore.toYaml();
     const expectedComponentYaml =
       "apiVersion: dapr.io/v1alpha1\n" +
@@ -66,7 +44,124 @@ describe("DaprContainer", () => {
       "    value: redis:6379\n" +
       "  - name: redisPassword\n" +
       '    value: ""\n';
-
     expect(componentYaml).toBe(expectedComponentYaml);
   });
+
+  it("should start and stop", async () => {
+    const network = await new Network().start();
+    const dapr = new DaprContainer(DAPR_RUNTIME_IMAGE)
+      .withNetwork(network)
+      .withDaprLogLevel("debug")
+      .withAppChannelAddress("host.testcontainers.internal");
+    const startedContainer = await dapr.start();
+    expect(startedContainer.getHost()).toBeDefined();
+    expect(startedContainer.getHttpPort()).toBeDefined();
+    expect(startedContainer.getGrpcPort()).toBeDefined();
+    expect(startedContainer.getHttpEndpoint()).toBeDefined();
+    expect(startedContainer.getGrpcEndpoint()).toBeDefined();
+    await startedContainer.stop();
+    await network.stop();
+  }, 60_000);
+
+  it("should initialize DaprClient", async () => {
+    const network = await new Network().start();
+    const dapr = new DaprContainer(DAPR_RUNTIME_IMAGE)
+      .withNetwork(network)
+      .withDaprLogLevel("debug")
+      .withAppChannelAddress("host.testcontainers.internal");
+    const startedContainer = await dapr.start();
+
+    const client = new DaprClient({
+      daprHost: startedContainer.getHost(),
+      daprPort: startedContainer.getHttpPort().toString(),
+    });
+    await client.start();
+    expect(client.getIsInitialized()).toBe(true);
+    await client.stop();
+
+    await startedContainer.stop();
+    await network.stop();
+  }, 60_000);
+
+  it("should provide kvstore in memory by default", async () => {
+    const network = await new Network().start();
+    const dapr = new DaprContainer(DAPR_RUNTIME_IMAGE)
+      .withNetwork(network)
+      .withDaprLogLevel("debug")
+      .withAppChannelAddress("host.testcontainers.internal")
+    const startedContainer = await dapr.start();
+
+    const client = new DaprClient({
+      daprHost: startedContainer.getHost(),
+      daprPort: startedContainer.getHttpPort().toString(),
+    });
+    await client.start();
+    expect(client.getIsInitialized()).toBe(true);
+    await client.state.save("kvstore", [{ key: "key", value: "value" }]);
+    const result = await client.state.get("kvstore", "key");
+    expect(result).toEqual("value");
+    await client.state.delete("kvstore", "key");
+    const resultAfterDelete = await client.state.get("kvstore", "key");
+    expect(resultAfterDelete).toBe("");
+    await client.stop();
+
+    await startedContainer.stop();
+    await network.stop();
+  }, 60_000);
+
+  it("should provide pubsub in memory by default", async () => {    
+    TestContainers.exposeHostPorts(8081);
+
+    const network = await new Network().start();
+    const dapr = new DaprContainer(DAPR_RUNTIME_IMAGE)
+      .withNetwork(network)
+      .withAppPort(8081)
+      .withDaprLogLevel("debug")
+      .withAppChannelAddress("host.testcontainers.internal")
+    const startedContainer = await dapr.start();
+
+    const server = new DaprServer({
+      serverHost: "127.0.0.1",
+      serverPort: "8081",
+      clientOptions: {
+        daprHost: startedContainer.getHost(),
+        daprPort: startedContainer.getHttpPort().toString()
+      },
+      logger: { level: LogLevel.Debug },
+    });
+
+    const client = new DaprClient({
+      daprHost: startedContainer.getHost(),
+      daprPort: startedContainer.getHttpPort().toString(),
+      logger: { level: LogLevel.Debug },
+    });
+
+    // Promise to resolve when the message is received
+    let processMessage: (data?: unknown) => void;
+    const promise = new Promise((res) => {
+      processMessage = res;
+    });
+
+    await server.pubsub.subscribe("pubsub", "topic", async (message) => {
+      console.log("Message received:", message);
+      processMessage(message);
+    });
+
+    await server.start();
+    // Wait for the server to start
+    await new Promise((resolve) => setTimeout(resolve, 1000));
+
+    console.log("Publishing message...");
+    const response = await client.pubsub.publish("pubsub", "topic", { key: "key", value: "value" });
+    console.log("Publish response:", response);
+
+    // Wait for the message to be processed
+    // await new Promise((resolve) => setTimeout(resolve, 5000));
+    const result = await promise; // FIXME
+    expect(result).toEqual({ key: "key", value: "value" });
+
+    await server.stop();
+    await startedContainer.stop();
+    await network.stop();
+  }, 120_000);
 });
@@ -79,7 +79,7 @@ export class DaprContainer extends GenericContainer {
         .withNetwork(this.startedNetwork)
         .withNetworkAliases(this.placementService)
       if (this.shouldReusePlacement) {
-        container.withReuse();
+        container.withReuse().withAutoRemove(false);
       }
       this.placementContainer = container;
     }
@@ -88,7 +88,7 @@ export class DaprContainer extends GenericContainer {
         .withNetwork(this.startedNetwork)
         .withNetworkAliases(this.schedulerService);
       if (this.shouldReuseScheduler) {
-        container.withReuse();
+        container.withReuse().withAutoRemove(false);
       }
       this.schedulerContainer = container;
     }