Feature/flow test options

src/flow/flow-test.test.ts

@@ -81,7 +81,7 @@ describe('Flow test', () => {
         await terminate(...terminables);
     });
 
-    it('should start', async () => {
+    it('should test using amqp successfully', async () => {
         await testFlow(conn, {
             type: IOType.AMQP,
             exchange: "test.ex",
@@ -95,6 +95,58 @@ describe('Flow test', () => {
         } as AmqpOutput);
     });
 
+    it('should test using amqp successfully when options are given', async () => {
+        await testFlow(conn, {
+            type: IOType.AMQP,
+            exchange: "test.ex",
+            routingKey: "queue.2.key",
+            payload: inputPayload,
+            options: {
+                replyTo: "test",
+                expiration: 5000
+            }
+        } as AmqpInput, {
+            expectedOutput,
+            type: IOType.AMQP,
+            queues: ["test.queue.1", "test.queue.2"],
+            expectedQueue: "test.queue.2",
+            expectedProperties: {
+                replyTo: "test",
+                expiration: "5000"
+            }
+        } as AmqpOutput);
+    });
+
+    it('should fail when expected options do not match those given', async () => {
+        try {
+            await testFlow(conn, {
+                type: IOType.AMQP,
+                exchange: "test.ex",
+                routingKey: "queue.2.key",
+                payload: inputPayload,
+                expectedProperties: {
+                    replyTo: "test",
+                    expiration: 5000
+                }
+            } as AmqpInput, {
+                expectedOutput,
+                type: IOType.AMQP,
+                queues: ["test.queue.1", "test.queue.2"],
+                expectedQueue: "test.queue.2",
+                expectedProperties: {
+                    replyTo: "not test",
+                    expiration: "5000"
+                }
+            } as AmqpOutput);
+        } catch (e) {
+            expect(e).to.exist;
+
+            return;
+        }
+
+        assert.fail("Did not throw error");
+    });
+
     it('should fail when message is not routed', async () => {
         try {
             await testFlow(conn, {

src/flow/flow-test.ts

@@ -1,4 +1,4 @@
-import { Channel, Connection, ConsumeMessage } from "amqplib";
+import { Channel, Connection, ConsumeMessage, Options, MessageProperties } from "amqplib";
 import axios from 'axios';
 import { assert, expect } from 'chai';
 
@@ -15,6 +15,7 @@ export interface Input {
 export interface AmqpInput extends Input {
     exchange: string;
     routingKey: string;
+    options?: Options.Publish;
 }
 
 export interface HttpInput extends Input {
@@ -30,6 +31,7 @@ export interface AmqpOutput extends Output {
     type: IOType;
     queues: string[];
     expectedQueue: string;
+    expectedProperties: MessageProperties;
 }
 
 interface CreateMessageOut {
@@ -46,7 +48,8 @@ interface CreateMessageOut {
 function createReceiveMessageCallback(
     expectedOutput: any,
     outQueue: string,
-    shouldReceive: boolean
+    shouldReceive: boolean,
+    expectedProperties?: MessageProperties,
 ): CreateMessageOut {
     const res: any = {
         resolve: undefined,
@@ -66,7 +69,7 @@ function createReceiveMessageCallback(
         } else {
             res.resolve();
         }
-    }, 6000);
+    }, 4000);
 
     return {
         consumerPromise,
@@ -85,6 +88,15 @@ function createReceiveMessageCallback(
                 const out = JSON.parse(msg.content.toString());
 
                 expect(out).to.deep.equal(expectedOutput, `Output of queue ${outQueue} does not match expected output`);
+                if (expectedProperties) {
+                    Object.keys(expectedProperties).forEach(key => {
+                        expect((msg.properties as any)[key]).to.deep.equal(
+                            (expectedProperties as any)[key],
+                            `Output property ${key} of queue ${outQueue} does not match expected property`
+                        );
+                    })
+                }
+
                 if (timeoutReached) {
                     throw new Error(`Queue ${outQueue} received a message after the timeout was reached`);
                 }
@@ -123,7 +135,8 @@ export async function testFlow(
                 const res = createReceiveMessageCallback(
                     out.expectedOutput,
                     outQueue,
-                    (out as AmqpOutput).expectedQueue === outQueue
+                    (out as AmqpOutput).expectedQueue === outQueue,
+                    (out as AmqpOutput).expectedProperties
                 );
                 const consumerTag = (await outChan.consume(outQueue, res.messageCallback)).consumerTag;
                 consumerTags.push(consumerTag);
@@ -143,7 +156,8 @@ export async function testFlow(
             await inputChan.publish(
                 (input as AmqpInput).exchange,
                 (input as AmqpInput).routingKey,
-                Buffer.from(JSON.stringify(input.payload))
+                Buffer.from(JSON.stringify(input.payload)),
+                (input as AmqpInput).options,
             );
         } else if (input.type === IOType.HTTP) {
             await axios.get((input as HttpInput).url);

src/flow/process.ts

@@ -18,7 +18,7 @@ export interface Flow {
 export async function start(f: Flow = {}): Promise<Termination> {
     return new Promise<Termination>((resolve, reject) => {
         const {
-            path = 'flows.json',
+            path = 'flow.json',
             port = 1880,
             userDir = '.',
             env = {},

src/index.ts

@@ -10,6 +10,7 @@ export const flow = flw;
 export const probes = prb.builtin;
 export const resources = rsc.builtin;
 export const makeGlobal = rsc.makeGlobal;
+export const testFlow = flw.testFlow;
 
 export interface Options {
     containers?: dkr.Container[];
@@ -65,11 +66,3 @@ export async function teardown(ctx: Context): Promise<void> {
 
     return ctx.destroy();
 }
-
-export async function test(
-    ctx: Context,
-    input: flw.Input,
-    output: flw.Output,
-): Promise<void> {
-    await flw.testFlow(ctx.resources.rabbitmq, input, output);
-}