chore(e2e): extend e2e tests with enrichment

.projenrc.js

@@ -25,6 +25,7 @@ const project = new awscdk.AwsCdkConstructLibrary({
     'cdk:deploy:target:eventbus:dynamo-stream': "cdk deploy --app 'npx ts-node --prefer-ts-exts e2e/targets/EventBus/DynamoStreamEventBus.ts'",
     'cdk:deploy:target:eventbus:sqs': "cdk deploy --app 'npx ts-node --prefer-ts-exts e2e/targets/EventBus/SqsEventBus.ts'",
     'cdk:deploy:target:eventbus:kinesis-stream': "cdk deploy --app 'npx ts-node --prefer-ts-exts e2e/targets/EventBus/KinesisStreamEventBus.ts'",
+    'cdk:deploy:all:sqs:sfn:eventbus': "cdk deploy --app 'npx ts-node --prefer-ts-exts e2e/SqsSfnEventBus.ts'",
   },
   deps: [
   ], /* Runtime dependencies of this module. */

.talismanrc

@@ -1,8 +1,8 @@
 fileignoreconfig:
-- filename: API.md
-  checksum: 344f1e758aadd2fc66a98bf3b0072779b1eb5d7c5080985269e0c8e20ea6433f
-- filename: README.md
-  checksum: 3ad06860bbaaa3b41e9e3dfe665ede1564446fa7052a4ed7fdf609af7ce5bcc1
+- filename: e2e/SqsSfnEventBus.ts
+  checksum: 5f1ecebc35cd23f2b7b05239a2f17795e18936ce5bd08c5b5e761fa746d94c0c
+version: ""
+cksum: 3ad06860bbaaa3b41e9e3dfe665ede1564446fa7052a4ed7fdf609af7ce5bcc1
 - filename: e2e/sources/kafkaProducer.ts
   checksum: d8518b18d16551a7ffc49b7c0683b5fdf4c82a9d1709ca3330eada84c0993183
 - filename: src/PipeInputTransformation.test.ts

e2e/SqsSfnEventBus.ts

@@ -0,0 +1,71 @@
+import { App, RemovalPolicy, Stack } from 'aws-cdk-lib';
+import { EventBus, Rule } from 'aws-cdk-lib/aws-events';
+import { CloudWatchLogGroup } from 'aws-cdk-lib/aws-events-targets';
+import { LogGroup } from 'aws-cdk-lib/aws-logs';
+import { Queue } from 'aws-cdk-lib/aws-sqs';
+import { LogLevel, Pass, StateMachine, StateMachineType } from 'aws-cdk-lib/aws-stepfunctions';
+import { DesiredState, EventBridgeEventBusTarget, Pipe, PipeInputTransformation, PipeTargetParameter, PipeVariables, SqsSource, StepFunctionEnrichment } from '../src';
+
+const app = new App();
+const stack = new Stack(app, 'E2EPipesSqsSfnEventBus', {});
+
+const sourceQueue = new Queue(stack, 'SourceQueue');
+const source = new SqsSource(sourceQueue);
+
+const stateMachineLogGroup = new LogGroup(stack, 'StateMachineLogGroup', {
+  removalPolicy: RemovalPolicy.DESTROY,
+});
+const stateMachine = new StateMachine(stack, 'StateMachine', {
+  definition: new Pass( stack, 'Pass', {
+    // resultPath: '$.output[]',
+
+  } ),
+  stateMachineType: StateMachineType.EXPRESS,
+  logs: {
+    destination: stateMachineLogGroup,
+    level: LogLevel.ALL,
+    includeExecutionData: true,
+  },
+});
+const enrichment = new StepFunctionEnrichment(stateMachine, {
+  inputTransformation: PipeInputTransformation.fromJson({
+    sqsEvent: PipeTargetParameter.fromPipeVariable(PipeVariables.PIPE_EVENT_JSON),
+    sqsEventBody: PipeTargetParameter.fromJsonPath('$.body'),
+  }),
+
+});
+
+const targetEventBus = new EventBus(stack, 'TargetEventBus', {
+
+
+});
+const target = new EventBridgeEventBusTarget(targetEventBus, {
+  detailType: 'test',
+  inputTemplate: PipeInputTransformation.fromJson({
+    ppeSourceEvent: PipeTargetParameter.fromPipeVariable(PipeVariables.PIPE_EVENT_JSON),
+    staticString: 'staticString',
+    body: PipeTargetParameter.fromJsonPath('$.body'),
+    enrichment: PipeTargetParameter.fromPipeVariable(PipeVariables.PIPE_ENRICHMENT_ARN),
+  }),
+
+});
+new Pipe(stack, 'Pipe', {
+  source,
+  enrichment,
+  target,
+  desiredState: DesiredState.RUNNING,
+});
+
+// DEBUG: add a rule to the event bus to see what messages are being sent
+const logGroup = new LogGroup(stack, 'LogGroup', {
+  removalPolicy: RemovalPolicy.DESTROY,
+});
+new Rule(stack, 'Rule', {
+  eventBus: targetEventBus,
+  eventPattern: {
+    account: [Stack.of(stack).account],
+  },
+  targets: [new CloudWatchLogGroup(logGroup, {
+
+  })],
+});

e2e/targets/EventBus/KinesisStreamEventBus.ts

@@ -3,7 +3,7 @@ import { EventBus, Rule } from 'aws-cdk-lib/aws-events';
 import { CloudWatchLogGroup } from 'aws-cdk-lib/aws-events-targets';
 import { Stream } from 'aws-cdk-lib/aws-kinesis';
 import { LogGroup } from 'aws-cdk-lib/aws-logs';
-import { EventBusTarget, KinesisStreamSource, Pipe, PipeSourceStartingPosition } from '../../../src';
+import { EventBridgeEventBusTarget, KinesisStreamSource, Pipe, PipeSourceStartingPosition } from '../../../src';
 
 const app = new App();
 const stack = new Stack(app, 'E2EPipesKinesisEventBus', {});
@@ -17,7 +17,7 @@ const source = new KinesisStreamSource(kinesisStream, {
 const targetEventBus = new EventBus(stack, 'TargetEventBus', {
 
 });
-const target = new EventBusTarget(targetEventBus, {
+const target = new EventBridgeEventBusTarget(targetEventBus, {
 
 });
 new Pipe(stack, 'Pipe', {

e2e/targets/EventBus/SqsEventBus.ts

@@ -3,7 +3,7 @@ import { EventBus, Rule } from 'aws-cdk-lib/aws-events';
 import { CloudWatchLogGroup } from 'aws-cdk-lib/aws-events-targets';
 import { LogGroup } from 'aws-cdk-lib/aws-logs';
 import { Queue } from 'aws-cdk-lib/aws-sqs';
-import { EventBusTarget, Pipe, SqsSource } from '../../../src';
+import { EventBridgeEventBusTarget, Pipe, SqsSource } from '../../../src';
 
 const app = new App();
 const stack = new Stack(app, 'E2EPipesSqsEventBus', {});
@@ -13,7 +13,7 @@ const source = new SqsSource(sourceQueue);
 const targetEventBus = new EventBus(stack, 'TargetEventBus', {
 
 });
-const target = new EventBusTarget(targetEventBus, {
+const target = new EventBridgeEventBusTarget(targetEventBus, {
 
 });
 new Pipe(stack, 'Pipe', {