1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import EmrTestFunctionLive, {
12 EmrTestFunction,
13 PROBE_RELEASE_LABEL,
14} from "./handler";
15import EmrSlowTestFunctionLive, { EmrSlowTestFunction } from "./slow-handler";
16
17const testOptions = { providers: AWS.providers() };
18const { test, beforeAll, afterAll } = Test.make(testOptions);
19const sharedStack = Core.scratchStack(testOptions, "EMRBindings");
20
21// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
22// over 60s on a fresh deploy.
23const readinessPolicy = Schedule.max([
24 Schedule.fixed("2 seconds"),
25 Schedule.recurs(75),
26]);
27
28let baseUrl: string;
29let functionArn: string;
30
31class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
32 readonly status: number;
33 readonly body: string;
34}> {}
35
36// The shared Lambda fixture occasionally answers a transient 5xx (cold
37// re-init, IAM propagation on the freshly attached policy that the handler's
38// `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine 4xx surfaces
39// immediately.
40const send = (request: HttpClientRequest.HttpClientRequest) =>
41 HttpClient.execute(request).pipe(
42 Effect.flatMap((response) =>
43 response.status >= 500
44 ? response.text.pipe(
45 Effect.flatMap((body) =>
46 Effect.fail(
47 new TransientUpstream({ status: response.status, body }),
48 ),
49 ),
50 )
51 : Effect.succeed(response),
52 ),
53 Effect.retry({
54 while: (e) => e._tag === "TransientUpstream",
55 schedule: Schedule.max([
56 Schedule.exponential("500 millis"),
57 Schedule.recurs(6),
58 ]),
59 }),
60 );
61
62const getJson = (path: string) =>
63 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
64 Effect.flatMap((r) => r.json),
65 );
66
67describe.sequential(
68 "EMR Bindings",
69 { tags: ["provider:aws", "provider:aws:emr", "provider:aws:lambda", "live"] },
70 () => {
71 beforeAll(
72 Effect.gen(function* () {
73 yield* Effect.logInfo("EMR test setup: destroying previous resources");
74 yield* sharedStack.destroy();
75
76 yield* Effect.logInfo("EMR test setup: deploying fixture");
77 const attrs = yield* sharedStack.deploy(
78 Effect.gen(function* () {
79 return yield* EmrTestFunction;
80 }).pipe(Effect.provide(EmrTestFunctionLive)),
81 );
82
83 expect(attrs.functionUrl).toBeTruthy();
84 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
85 functionArn = attrs.functionArn;
86
87 const readinessUrl = `${baseUrl}/bindings`;
88 yield* HttpClient.get(readinessUrl).pipe(
89 Effect.flatMap((response) =>
90 response.status === 200
91 ? Effect.succeed(response)
92 : Effect.fail(
93 new Error(`Function not ready: ${response.status}`),
94 ),
95 ),
96 Effect.retry({ schedule: readinessPolicy }),
97 );
98 }),
99 { timeout: 240_000 },
100 );
101
102 afterAll(sharedStack.destroy(), { timeout: 120_000 });
103
104 describe("binding registration", () => {
105 test.provider("all four capabilities initialize in the runtime", () =>
106 Effect.gen(function* () {
107 const response = (yield* getJson("/bindings")) as { bound: string[] };
108 expect(response.bound).toHaveLength(4);
109 }),
110 );
111 });
112
113 describe("ListClusters", () => {
114 test.provider("lists the account's active clusters", () =>
115 Effect.gen(function* () {
116 const response = (yield* getJson("/clusters")) as { count: number };
117 expect(response.count).toBeGreaterThanOrEqual(0);
118 }),
119 );
120 });
121
122 describe("ListReleaseLabels", () => {
123 test.provider("lists the region's release catalog, newest first", () =>
124 Effect.gen(function* () {
125 const response = (yield* getJson("/releases")) as {
126 count: number;
127 latest?: string;
128 };
129 expect(response.count).toBeGreaterThan(0);
130 expect(response.latest).toMatch(/^emr-/);
131 }),
132 );
133 });
134
135 describe("DescribeReleaseLabel", () => {
136 test.provider("reads the applications a release ships", () =>
137 Effect.gen(function* () {
138 const response = (yield* getJson("/release")) as {
139 applications: string[];
140 };
141 expect(response.applications).toContain("Spark");
142 }),
143 );
144 });
145
146 describe("ListSupportedInstanceTypes", () => {
147 test.provider(`lists instance types for ${PROBE_RELEASE_LABEL}`, () =>
148 Effect.gen(function* () {
149 const response = (yield* getJson("/instance-types")) as {
150 types: string[];
151 };
152 expect(response.types.length).toBeGreaterThan(0);
153 }),
154 );
155 });
156
157 describe(
158 "consumeClusterEvents",
159 { tags: ["provider:aws:eventbridge"] },
160 () => {
161 test.provider(
162 "the deploy created an EventBridge rule targeting the function",
163 () =>
164 Effect.gen(function* () {
165 // Out-of-band via distilled: the fixture's consumeClusterEvents
166 // must have materialized as a rule on the default bus with the
167 // Lambda as target.
168 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
169 TargetArn: functionArn,
170 });
171 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
172 }),
173 );
174 },
175 );
176 },
177);
178
179// ---------------------------------------------------------------------------
180// Cluster-scoped bindings (steps data plane, inspection reads, managed
181// scaling) need a real EMR cluster (~10-15 min to WAITING, billed per
182// instance-hour) — gated behind AWS_TEST_SLOW=1 like the Cluster lifecycle
183// test.
184// ---------------------------------------------------------------------------
185
186const slowStack = Core.scratchStack(testOptions, "EMRSlowBindings");
187
188test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
189 "steps + inspection + managed-scaling bindings against a live cluster",
190 () =>
191 Effect.gen(function* () {
192 yield* slowStack.destroy();
193
194 yield* Effect.gen(function* () {
195 const { functionUrl } = yield* slowStack.deploy(
196 Effect.gen(function* () {
197 return yield* EmrSlowTestFunction;
198 }).pipe(Effect.provide(EmrSlowTestFunctionLive)),
199 );
200 expect(functionUrl).toBeTruthy();
201 const slowBaseUrl = functionUrl!.replace(/\/+$/, "");
202
203 const get = (path: string) =>
204 HttpClient.get(`${slowBaseUrl}${path}`).pipe(
205 Effect.retry({
206 schedule: Schedule.max([
207 Schedule.exponential("500 millis"),
208 Schedule.recurs(10),
209 ]),
210 }),
211 Effect.flatMap((r) => r.json),
212 );
213
214 // Steps data plane: submit, describe (poll until terminal), list,
215 // and cancel — proves the JobFlowId/ClusterId injections and grants.
216 const added = (yield* get("/steps/add")) as { stepId?: string };
217 expect(added.stepId).toMatch(/^s-/);
218
219 const step = (yield* get(`/step?id=${added.stepId}`).pipe(
220 Effect.repeat({
221 schedule: Schedule.spaced("10 seconds"),
222 until: (s): boolean =>
223 (s as { state?: string }).state !== "PENDING" &&
224 (s as { state?: string }).state !== "RUNNING",
225 times: 30,
226 }),
227 )) as { state?: string; name?: string };
228 expect(step.name).toBe("alchemy-echo");
229 expect(["COMPLETED", "FAILED", "CANCELLED"]).toContain(step.state);
230
231 const steps = (yield* get("/steps")) as { count: number };
232 expect(steps.count).toBeGreaterThanOrEqual(1);
233
234 // Cancelling an already-terminal step reports a per-step failure
235 // status — either status proves the grant.
236 const cancel = (yield* get(`/steps/cancel?id=${added.stepId}`)) as {
237 status?: string;
238 };
239 expect(cancel.status).toBeDefined();
240
241 // Inspection reads.
242 const instances = (yield* get("/instances")) as { count: number };
243 expect(instances.count).toBeGreaterThanOrEqual(1);
244 const groups = (yield* get("/groups")) as { types: string[] };
245 expect(groups.types).toContain("MASTER");
246 const bootstrap = (yield* get("/bootstrap")) as { count: number };
247 expect(bootstrap.count).toBeGreaterThanOrEqual(0);
248
249 // Managed scaling put -> get -> remove (or a typed validation tag on
250 // a master-only cluster).
251 const scaling = (yield* get("/scaling")) as {
252 ok: boolean;
253 max?: number;
254 tag?: string;
255 };
256 if (scaling.ok) {
257 expect(scaling.max).toBe(4);
258 } else {
259 expect(scaling.tag).toBe("ValidationException");
260 }
261 }).pipe(Effect.ensuring(slowStack.destroy().pipe(Effect.orDie)));
262 }),
263 // cluster create (~10-15 min) + step run + probes + termination initiation.
264 {
265 tags: [
266 "provider:aws",
267 "provider:aws:emr",
268 "provider:aws:iam",
269 "provider:aws:lambda",
270 "live",
271 ],
272 timeout: 2_700_000,
273 },
274);