1import * as AWS from "@/AWS";
2import { AWSEnvironment } from "@/AWS/Environment";
3import * as Output from "@/Output";
4import * as Test from "@/Test/Alchemy";
5import * as EC2 from "@distilled.cloud/aws/ec2";
6import * as mwaa from "@distilled.cloud/aws/mwaa";
7import { expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import { getDefaultVpc } from "../DefaultVpc.ts";
11
12const { test } = Test.make({ providers: AWS.providers() });
13
14// Ungated typed-error probe: proves the distilled mwaa error union carries the
15// ResourceNotFoundException tag this provider's read/delete paths depend on. A
16// well-formed but nonexistent environment name returns ResourceNotFoundException.
17test.provider(
18 "getEnvironment on a nonexistent environment fails with ResourceNotFoundException",
19 () =>
20 Effect.gen(function* () {
21 const error = yield* Effect.flip(
22 mwaa.getEnvironment({ Name: "alchemy-mwaa-nonexistent-probe" }),
23 );
24 expect(error._tag).toBe("ResourceNotFoundException");
25 }),
26 { tags: ["provider:aws", "provider:aws:mwaa", "live"] },
27);
28
29// Resolve two subnets from the account's default VPC in distinct Availability
30// Zones plus the VPC's default security group. NOTE: a REAL MWAA environment
31// requires PRIVATE subnets with outbound internet egress (NAT gateway or the
32// full set of MWAA VPC interface endpoints). The factory never creates NAT
33// gateways, so this networking is expected to be supplied out of band by an
34// entitled/patient runner — which is exactly why the lifecycle below is gated.
35const resolveNetwork = Effect.gen(function* () {
36 const vpc = yield* getDefaultVpc;
37 const subnets = yield* EC2.describeSubnets({
38 Filters: [
39 { Name: "vpc-id", Values: [vpc.vpcId] },
40 { Name: "default-for-az", Values: ["true"] },
41 ],
42 });
43 const byAz = new Map<string, string>();
44 for (const s of subnets.Subnets ?? []) {
45 if (s.AvailabilityZone && s.SubnetId && !byAz.has(s.AvailabilityZone)) {
46 byAz.set(s.AvailabilityZone, s.SubnetId);
47 }
48 }
49 const subnetIds = [...byAz.values()].slice(0, 2);
50
51 const sgs = yield* EC2.describeSecurityGroups({
52 Filters: [
53 { Name: "vpc-id", Values: [vpc.vpcId] },
54 { Name: "group-name", Values: ["default"] },
55 ],
56 });
57 const securityGroupId = sgs.SecurityGroups?.[0]?.GroupId;
58 return { subnetIds, securityGroupId };
59});
60
61// MWAA execution role: trusts the Airflow environment + task principals and
62// grants the standard access to the DAGs bucket, CloudWatch Logs, SQS, and KMS.
63const executionRolePolicy = (
64 bucketArn: Output.Output<string>,
65 region: string,
66 account: string,
67) => ({
68 Version: "2012-10-17" as const,
69 Statement: [
70 {
71 Effect: "Allow" as const,
72 Action: ["airflow:PublishMetrics"],
73 Resource: [`arn:aws:airflow:${region}:${account}:environment/*`],
74 },
75 {
76 Effect: "Allow" as const,
77 Action: ["s3:GetObject*", "s3:GetBucket*", "s3:List*"],
78 Resource: [bucketArn, Output.interpolate`${bucketArn}/*`],
79 },
80 {
81 Effect: "Allow" as const,
82 Action: [
83 "logs:CreateLogStream",
84 "logs:CreateLogGroup",
85 "logs:PutLogEvents",
86 "logs:GetLogEvents",
87 "logs:GetLogRecord",
88 "logs:GetLogGroupFields",
89 "logs:GetQueryResults",
90 "logs:DescribeLogGroups",
91 ],
92 Resource: [`arn:aws:logs:${region}:${account}:log-group:airflow-*`],
93 },
94 {
95 Effect: "Allow" as const,
96 Action: ["cloudwatch:PutMetricData"],
97 Resource: ["*"],
98 },
99 {
100 Effect: "Allow" as const,
101 Action: [
102 "sqs:ChangeMessageVisibility",
103 "sqs:DeleteMessage",
104 "sqs:GetQueueAttributes",
105 "sqs:GetQueueUrl",
106 "sqs:ReceiveMessage",
107 "sqs:SendMessage",
108 ],
109 Resource: [`arn:aws:sqs:${region}:*:airflow-celery-*`],
110 },
111 ],
112});
113
114// MWAA environments take ~20-30 minutes to create, ~20-30 minutes to delete,
115// and are billed hourly while they exist. The full lifecycle is gated behind
116// AWS_TEST_SLOW=1 and always destroys what it created.
117test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
118 "create environment, verify available, destroy, verify gone",
119 (stack) =>
120 Effect.gen(function* () {
121 yield* stack.destroy();
122
123 const { accountId, region } = yield* AWSEnvironment.current;
124 const { subnetIds, securityGroupId } = yield* resolveNetwork;
125 expect(subnetIds.length).toBe(2);
126 expect(securityGroupId).toBeDefined();
127
128 const { environment } = yield* stack.deploy(
129 Effect.gen(function* () {
130 const bucket = yield* AWS.S3.Bucket("Dags", {
131 versioning: "Enabled",
132 forceDestroy: true,
133 });
134 const role = yield* AWS.IAM.Role("ExecutionRole", {
135 assumeRolePolicyDocument: {
136 Version: "2012-10-17",
137 Statement: [
138 {
139 Effect: "Allow",
140 Principal: {
141 Service: [
142 "airflow.amazonaws.com",
143 "airflow-env.amazonaws.com",
144 ],
145 },
146 Action: ["sts:AssumeRole"],
147 },
148 ],
149 },
150 inlinePolicies: {
151 "mwaa-execution": executionRolePolicy(
152 bucket.bucketArn,
153 region,
154 accountId,
155 ),
156 },
157 });
158 const environment = yield* AWS.MWAA.Environment("Airflow", {
159 executionRoleArn: role.roleArn,
160 sourceBucketArn: bucket.bucketArn,
161 dagS3Path: "dags",
162 subnetIds,
163 securityGroupIds: securityGroupId ? [securityGroupId] : undefined,
164 environmentClass: "mw1.small",
165 maxWorkers: 2,
166 minWorkers: 1,
167 webserverAccessMode: "PUBLIC_ONLY",
168 loggingConfiguration: {
169 schedulerLogs: { enabled: true, logLevel: "INFO" },
170 taskLogs: { enabled: true, logLevel: "INFO" },
171 },
172 tags: { fixture: "mwaa-environment" },
173 });
174 return { environment };
175 }),
176 );
177
178 expect(environment.environmentName).toBeDefined();
179 expect(environment.arn).toContain(":environment/");
180 expect(environment.status).toBe("AVAILABLE");
181 expect(environment.webserverUrl).toBeDefined();
182
183 // Out-of-band verification via distilled.
184 const described = yield* mwaa.getEnvironment({
185 Name: environment.environmentName,
186 });
187 expect(described.Environment?.Status).toBe("AVAILABLE");
188 expect(described.Environment?.EnvironmentClass).toBe("mw1.small");
189 expect(described.Environment?.Tags?.fixture).toBe("mwaa-environment");
190
191 yield* stack.destroy();
192 yield* assertEnvironmentDeleted(environment.environmentName);
193 }),
194 // create (~20-30 min) + destroy initiation, one test.
195 {
196 tags: [
197 "provider:aws",
198 "provider:aws:ec2",
199 "provider:aws:iam",
200 "provider:aws:mwaa",
201 "provider:aws:s3",
202 "live",
203 ],
204 timeout: 4_200_000,
205 },
206);
207
208// Deletion is verified as INITIATED (status DELETING, irreversible) or fully
209// gone. Full disappearance takes ~20-30 more minutes server-side.
210const assertEnvironmentDeleted = (name: string) =>
211 Effect.gen(function* () {
212 const status = yield* mwaa.getEnvironment({ Name: name }).pipe(
213 Effect.map((r) => r.Environment?.Status ?? "gone"),
214 Effect.catchTag("ResourceNotFoundException", () =>
215 Effect.succeed("gone" as const),
216 ),
217 );
218 if (status !== "gone" && status !== "DELETING") {
219 return yield* Effect.fail(
220 new Error(
221 `MWAA environment '${name}' still exists (status: ${status})`,
222 ),
223 );
224 }
225 }).pipe(
226 Effect.retry({
227 schedule: Schedule.max([
228 Schedule.fixed("15 seconds"),
229 Schedule.recurs(20),
230 ]),
231 }),
232 );