1import * as AWS from "@/AWS";
2import * as Provider from "@/Provider";
3import * as Test from "@/Test/Alchemy";
4import * as Kinesis from "@distilled.cloud/aws/kinesis";
5import { describe, expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7
8const { test } = Test.make({ providers: AWS.providers() });
9
10describe(
11 "AWS.Kinesis.StreamConsumer",
12 { tags: ["provider:aws", "provider:aws:kinesis", "live"] },
13 () => {
14 test.provider(
15 "create, update tags, and delete a stream consumer",
16 (stack) =>
17 Effect.gen(function* () {
18 const deployed = yield* stack.deploy(
19 Effect.gen(function* () {
20 const stream = yield* AWS.Kinesis.Stream("ConsumerSourceStream", {
21 streamMode: "PROVISIONED",
22 shardCount: 1,
23 });
24
25 const consumer = yield* AWS.Kinesis.StreamConsumer(
26 "AnalyticsConsumer",
27 {
28 streamArn: stream.streamArn,
29 tags: {
30 fixture: "consumer-test",
31 },
32 },
33 );
34
35 return { stream, consumer };
36 }),
37 );
38
39 const description = yield* Kinesis.describeStreamConsumer({
40 ConsumerARN: deployed.consumer.consumerArn,
41 });
42 expect(description.ConsumerDescription.ConsumerStatus).toEqual(
43 "ACTIVE",
44 );
45
46 const initialTags = yield* Kinesis.listTagsForResource({
47 ResourceARN: deployed.consumer.consumerArn,
48 });
49 expect(initialTags.Tags).toContainEqual({
50 Key: "fixture",
51 Value: "consumer-test",
52 });
53
54 const updated = yield* stack.deploy(
55 Effect.gen(function* () {
56 const stream = yield* AWS.Kinesis.Stream("ConsumerSourceStream", {
57 streamMode: "PROVISIONED",
58 shardCount: 1,
59 });
60
61 const consumer = yield* AWS.Kinesis.StreamConsumer(
62 "AnalyticsConsumer",
63 {
64 streamArn: stream.streamArn,
65 tags: {
66 fixture: "consumer-test-updated",
67 team: "platform",
68 },
69 },
70 );
71
72 return { stream, consumer };
73 }),
74 );
75
76 const updatedTags = yield* Kinesis.listTagsForResource({
77 ResourceARN: updated.consumer.consumerArn,
78 });
79 expect(updatedTags.Tags).toContainEqual({
80 Key: "fixture",
81 Value: "consumer-test-updated",
82 });
83 expect(updatedTags.Tags).toContainEqual({
84 Key: "team",
85 Value: "platform",
86 });
87
88 yield* stack.destroy();
89
90 const deleted = yield* Kinesis.describeStreamConsumer({
91 ConsumerARN: updated.consumer.consumerArn,
92 }).pipe(
93 Effect.map(() => false),
94 Effect.catchTag("ResourceNotFoundException", () =>
95 Effect.succeed(true),
96 ),
97 );
98 expect(deleted).toBe(true);
99 }),
100 { timeout: 180_000 },
101 );
102
103 test.provider(
104 "list enumerates the deployed stream consumer",
105 (stack) =>
106 Effect.gen(function* () {
107 yield* stack.destroy();
108
109 const deployed = yield* stack.deploy(
110 Effect.gen(function* () {
111 const stream = yield* AWS.Kinesis.Stream("ListConsumerStream", {
112 streamMode: "PROVISIONED",
113 shardCount: 1,
114 });
115
116 const consumer = yield* AWS.Kinesis.StreamConsumer(
117 "ListConsumer",
118 {
119 streamArn: stream.streamArn,
120 },
121 );
122
123 return { stream, consumer };
124 }),
125 );
126
127 const provider = yield* Provider.findProvider(
128 AWS.Kinesis.StreamConsumer,
129 );
130 const all = yield* provider.list();
131
132 expect(
133 all.some((c) => c.consumerArn === deployed.consumer.consumerArn),
134 ).toBe(true);
135
136 yield* stack.destroy();
137 }),
138 { timeout: 180_000 },
139 );
140 },
141);