1import * as AWS from "@/AWS";
2import { Crawler, Database } from "@/AWS/Glue";
3import { Role } from "@/AWS/IAM";
4import { Bucket } from "@/AWS/S3";
5import * as Test from "@/Test/Alchemy";
6import * as glue from "@distilled.cloud/aws/glue";
7import * as s3 from "@distilled.cloud/aws/s3";
8import { expect } from "alchemy-test";
9import * as Output from "@/Output";
10import * as Effect from "effect/Effect";
11import * as Schedule from "effect/Schedule";
12
13const { test } = Test.make({ providers: AWS.providers() });
14
15const getCrawler = (name: string) =>
16 glue.getCrawler({ Name: name }).pipe(
17 Effect.map((r) => r.Crawler),
18 Effect.catchTag("EntityNotFoundException", () => Effect.succeed(undefined)),
19 );
20
21const crawlerRole = () =>
22 Role("GlueCrawlerRole", {
23 assumeRolePolicyDocument: {
24 Version: "2012-10-17",
25 Statement: [
26 {
27 Effect: "Allow",
28 Principal: { Service: "glue.amazonaws.com" },
29 Action: ["sts:AssumeRole"],
30 },
31 ],
32 },
33 managedPolicyArns: [
34 "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole",
35 "arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess",
36 ],
37 });
38
39test.provider(
40 "create, update, delete Glue crawler definition",
41 (stack) =>
42 Effect.gen(function* () {
43 yield* stack.destroy();
44
45 const created = yield* stack.deploy(
46 Effect.gen(function* () {
47 const database = yield* Database("CrawlerDb", {});
48 const bucket = yield* Bucket("CrawlerBucket", { forceDestroy: true });
49 const role = yield* crawlerRole();
50 const crawler = yield* Crawler("EventsCrawler", {
51 role: role.roleArn,
52 databaseName: database.databaseName,
53 targets: {
54 s3Targets: [
55 { path: Output.interpolate`s3://${bucket.bucketName}/data/` },
56 ],
57 },
58 tablePrefix: "raw_",
59 tags: { Environment: "test" },
60 });
61 return { database, bucket, crawler };
62 }),
63 );
64
65 expect(created.crawler.crawlerName).toBeDefined();
66 expect(created.crawler.crawlerArn).toContain(
67 `:crawler/${created.crawler.crawlerName}`,
68 );
69
70 // out-of-band verification
71 const observed = yield* getCrawler(created.crawler.crawlerName);
72 expect(observed?.Name).toEqual(created.crawler.crawlerName);
73 expect(observed?.DatabaseName).toEqual(created.database.databaseName);
74 expect(observed?.TablePrefix).toEqual("raw_");
75 expect(observed?.Targets?.S3Targets?.[0]?.Path).toEqual(
76 `s3://${created.bucket.bucketName}/data/`,
77 );
78
79 // tags (crawlers ARE ARN-taggable)
80 const tags = yield* glue.getTags({
81 ResourceArn: created.crawler.crawlerArn,
82 });
83 expect(tags.Tags?.["alchemy::id"]).toBeDefined();
84 expect(tags.Tags?.Environment).toEqual("test");
85
86 // update: description + schedule
87 yield* stack.deploy(
88 Effect.gen(function* () {
89 const database = yield* Database("CrawlerDb", {});
90 const bucket = yield* Bucket("CrawlerBucket", { forceDestroy: true });
91 const role = yield* crawlerRole();
92 const crawler = yield* Crawler("EventsCrawler", {
93 role: role.roleArn,
94 databaseName: database.databaseName,
95 description: "crawls the events prefix",
96 targets: {
97 s3Targets: [
98 { path: Output.interpolate`s3://${bucket.bucketName}/data/` },
99 ],
100 },
101 tablePrefix: "raw_",
102 schedule: "cron(0 12 * * ? *)",
103 schemaChangePolicy: {
104 updateBehavior: "UPDATE_IN_DATABASE",
105 deleteBehavior: "DEPRECATE_IN_DATABASE",
106 },
107 tags: { Environment: "test" },
108 });
109 return { crawler };
110 }),
111 );
112
113 const reobserved = yield* getCrawler(created.crawler.crawlerName);
114 expect(reobserved?.Description).toEqual("crawls the events prefix");
115 expect(reobserved?.Schedule?.ScheduleExpression).toEqual(
116 "cron(0 12 * * ? *)",
117 );
118 expect(reobserved?.SchemaChangePolicy?.UpdateBehavior).toEqual(
119 "UPDATE_IN_DATABASE",
120 );
121
122 yield* stack.destroy();
123 const gone = yield* getCrawler(created.crawler.crawlerName);
124 expect(gone).toBeUndefined();
125 }),
126 {
127 tags: [
128 "provider:aws",
129 "provider:aws:glue",
130 "provider:aws:iam",
131 "provider:aws:s3",
132 "live",
133 ],
134 },
135);
136
137// A live crawl takes ~2-4 minutes end to end — gated behind AWS_TEST_SLOW=1.
138test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
139 "run a live crawl that populates the catalog",
140 (stack) =>
141 Effect.gen(function* () {
142 yield* stack.destroy();
143
144 const deployed = yield* stack.deploy(
145 Effect.gen(function* () {
146 const database = yield* Database("LiveCrawlDb", {});
147 const bucket = yield* Bucket("LiveCrawlBucket", {
148 forceDestroy: true,
149 });
150 const role = yield* crawlerRole();
151 const crawler = yield* Crawler("LiveCrawler", {
152 role: role.roleArn,
153 databaseName: database.databaseName,
154 targets: {
155 s3Targets: [
156 { path: Output.interpolate`s3://${bucket.bucketName}/data/` },
157 ],
158 },
159 });
160 return { database, bucket, crawler };
161 }),
162 );
163
164 // seed a CSV object so the crawler infers a table
165 yield* s3.putObject({
166 Bucket: deployed.bucket.bucketName,
167 Key: "data/events/part-0.csv",
168 Body: new TextEncoder().encode("id,amount\n1,10.5\n2,20.0\n"),
169 ContentType: "text/csv",
170 });
171
172 yield* glue.startCrawler({ Name: deployed.crawler.crawlerName });
173
174 // poll until the crawler returns to READY with a terminal last-crawl
175 const finalState = yield* getCrawler(deployed.crawler.crawlerName).pipe(
176 Effect.repeat({
177 schedule: Schedule.spaced("15 seconds"),
178 until: (c) =>
179 c?.State === "READY" && c?.LastCrawl?.Status !== undefined,
180 times: 20,
181 }),
182 );
183 expect(finalState?.LastCrawl?.Status).toEqual("SUCCEEDED");
184
185 // a table should now exist in the catalog
186 const tables = yield* glue.getTables({
187 DatabaseName: deployed.database.databaseName,
188 });
189 expect(tables.TableList?.length ?? 0).toBeGreaterThan(0);
190
191 yield* stack.destroy();
192 }),
193 {
194 tags: [
195 "provider:aws",
196 "provider:aws:glue",
197 "provider:aws:iam",
198 "provider:aws:s3",
199 "live",
200 ],
201 timeout: 360_000,
202 },
203);