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 MediaConnectTestFunctionLive, {
12 MediaConnectTestFunction,
13} from "./fixtures/handler";
14
15const testOptions = { providers: AWS.providers() };
16const { test, beforeAll, afterAll } = Test.make(testOptions);
17const sharedStack = Core.scratchStack(testOptions, "MediaConnectBindings");
18
19// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
20// over 60s on a fresh deploy.
21const readinessPolicy = Schedule.max([
22 Schedule.fixed("2 seconds"),
23 Schedule.recurs(75),
24]);
25
26let baseUrl: string;
27let functionArn: string;
28
29class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
30 readonly status: number;
31 readonly body: string;
32}> {}
33
34// The shared Lambda fixture occasionally answers a transient 5xx under load
35// (cold re-init, IAM propagation on the freshly attached policy that the
36// handler's `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine
37// 4xx/assertion failure surfaces immediately.
38const send = (request: HttpClientRequest.HttpClientRequest) =>
39 HttpClient.execute(request).pipe(
40 Effect.flatMap((response) =>
41 response.status >= 500
42 ? response.text.pipe(
43 Effect.flatMap((body) =>
44 Effect.fail(
45 new TransientUpstream({ status: response.status, body }),
46 ),
47 ),
48 )
49 : Effect.succeed(response),
50 ),
51 Effect.retry({
52 while: (e) => e._tag === "TransientUpstream",
53 schedule: Schedule.max([
54 Schedule.exponential("500 millis"),
55 Schedule.recurs(6),
56 ]),
57 }),
58 );
59
60const getJson = (path: string) =>
61 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
62 Effect.flatMap((r) => r.json),
63 );
64
65const postJson = (path: string) =>
66 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
67 Effect.flatMap((r) => r.json),
68 );
69
70describe.sequential(
71 "MediaConnect Bindings",
72 {
73 tags: [
74 "provider:aws",
75 "provider:aws:lambda",
76 "provider:aws:mediaconnect",
77 "live",
78 ],
79 },
80 () => {
81 beforeAll(
82 Effect.gen(function* () {
83 yield* Effect.logInfo(
84 "MediaConnect test setup: destroying previous resources",
85 );
86 yield* sharedStack.destroy();
87
88 yield* Effect.logInfo("MediaConnect test setup: deploying fixture");
89 const attrs = yield* sharedStack.deploy(
90 Effect.gen(function* () {
91 return yield* MediaConnectTestFunction;
92 }).pipe(Effect.provide(MediaConnectTestFunctionLive)),
93 );
94
95 expect(attrs.functionUrl).toBeTruthy();
96 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
97 functionArn = attrs.functionArn;
98
99 const readinessUrl = `${baseUrl}/bindings`;
100 yield* Effect.logInfo(
101 `MediaConnect test setup: probing readiness at ${readinessUrl}`,
102 );
103 yield* HttpClient.get(readinessUrl).pipe(
104 Effect.flatMap((response) =>
105 response.status === 200
106 ? Effect.succeed(response)
107 : Effect.fail(
108 new Error(`Function not ready: ${response.status}`),
109 ),
110 ),
111 Effect.tapError((error) =>
112 Effect.logWarning(
113 `MediaConnect test setup: fixture not ready yet (${String(error)})`,
114 ),
115 ),
116 Effect.retry({ schedule: readinessPolicy }),
117 );
118 }),
119 { timeout: 300_000 },
120 );
121
122 afterAll(sharedStack.destroy(), { timeout: 240_000 });
123
124 describe("binding registration", () => {
125 test.provider("all nine capabilities initialize in the runtime", () =>
126 Effect.gen(function* () {
127 const response = (yield* getJson("/bindings")) as { bound: string[] };
128 expect(response.bound).toHaveLength(9);
129 }),
130 );
131 });
132
133 describe("DescribeFlow", () => {
134 test.provider("the fixture flow idles in STANDBY", () =>
135 Effect.gen(function* () {
136 const response = (yield* getJson("/flow")) as {
137 status?: string;
138 sourceName?: string;
139 };
140 expect(response.status).toBe("STANDBY");
141 expect(response.sourceName).toBe("primary");
142 }),
143 );
144 });
145
146 describe("DescribeFlowSourceMetadata", () => {
147 test.provider(
148 "an idle flow has no programs — only messages or the typed BadRequestException",
149 () =>
150 Effect.gen(function* () {
151 const response = (yield* getJson("/source-metadata")) as {
152 programs: number;
153 messages: number;
154 tag?: string;
155 };
156 expect(response.programs).toBe(0);
157 if (response.tag !== undefined) {
158 expect(response.tag).toBe("BadRequestException");
159 }
160 }),
161 );
162 });
163
164 describe("DescribeFlowSourceThumbnail", () => {
165 test.provider(
166 "an idle flow has no thumbnail image — only messages or the typed BadRequestException",
167 () =>
168 Effect.gen(function* () {
169 const response = (yield* getJson("/thumbnail")) as {
170 hasImage: boolean;
171 messages: number;
172 tag?: string;
173 };
174 expect(response.hasImage).toBe(false);
175 if (response.tag !== undefined) {
176 expect(response.tag).toBe("BadRequestException");
177 }
178 }),
179 );
180 });
181
182 describe("ListFlows", () => {
183 test.provider(
184 "account-level flow enumeration sees the fixture flow",
185 () =>
186 Effect.gen(function* () {
187 const response = (yield* getJson("/flows")) as { count: number };
188 expect(response.count).toBeGreaterThanOrEqual(1);
189 }),
190 );
191 });
192
193 describe("ListEntitlements", () => {
194 test.provider("account-level entitlement enumeration succeeds", () =>
195 Effect.gen(function* () {
196 const response = (yield* getJson("/entitlements")) as {
197 count: number;
198 };
199 expect(response.count).toBeGreaterThanOrEqual(0);
200 }),
201 );
202 });
203
204 describe("StopFlow", () => {
205 test.provider(
206 "stopping a STANDBY flow is the typed BadRequestException",
207 () =>
208 Effect.gen(function* () {
209 const response = (yield* postJson("/stop")) as {
210 status?: string;
211 tag?: string;
212 };
213 expect(response.tag).toBe("BadRequestException");
214 expect(response.status).toBeUndefined();
215 }),
216 );
217 });
218
219 describe("GrantFlowEntitlements + RevokeFlowEntitlement", () => {
220 test.provider("granting and revoking an entitlement roundtrips", () =>
221 Effect.gen(function* () {
222 const response = (yield* postJson("/entitlement-cycle")) as {
223 granted: boolean;
224 revoked: boolean;
225 tag?: string;
226 };
227 // A failure surfaces its typed tag in the response for diagnosis.
228 expect(response.tag).toBeUndefined();
229 expect(response.granted).toBe(true);
230 expect(response.revoked).toBe(true);
231 }),
232 );
233 });
234
235 describe("StartFlow", () => {
236 // An ACTIVE flow bills for transport by the hour, so the start/stop
237 // roundtrip only runs when explicitly enabled.
238 test.provider.skipIf(!process.env.AWS_TEST_MEDIACONNECT)(
239 "start/stop roundtrip returns the flow to rest",
240 () =>
241 Effect.gen(function* () {
242 const response = (yield* postJson("/start-stop")) as {
243 startStatus?: string;
244 stopStatus?: string;
245 };
246 expect(["STARTING", "ACTIVE"]).toContain(response.startStatus);
247 expect(["STOPPING", "STANDBY"]).toContain(response.stopStatus);
248 }),
249 { timeout: 150_000 },
250 );
251 });
252
253 describe(
254 "consumeFlowEvents",
255 { tags: ["provider:aws:eventbridge"] },
256 () => {
257 test.provider(
258 "the deploy created an EventBridge rule targeting the function",
259 () =>
260 Effect.gen(function* () {
261 // Out-of-band via distilled: the fixture's consumeFlowEvents
262 // must have materialized as a rule on the default bus with the
263 // Lambda as target.
264 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
265 TargetArn: functionArn,
266 });
267 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
268 }),
269 );
270 },
271 );
272 },
273);