1import * as GCP from "@/GCP";
2import { GcpEnvironment } from "@/GCP/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
5import { expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import { MinimumLogLevel } from "effect/References";
8import * as Schedule from "effect/Schedule";
9import { withDataplexSlot } from "./quota.ts";
10
11const { test } = Test.make({ providers: GCP.providers() });
12
13const logLevel = Effect.provideService(
14 MinimumLogLevel,
15 process.env.DEBUG ? "Debug" : "Info",
16);
17
18const waitUntilGone = (name: string) =>
19 dataplex.getProjectsLocationsLakesTasks({ name }).pipe(
20 Effect.as("found" as const),
21 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
22 Effect.repeat({
23 schedule: Schedule.spaced("2 seconds"),
24 until: (status) => status === "gone",
25 times: 10,
26 }),
27 );
28
29const serviceAccountOf = GcpEnvironment.current.pipe(
30 Effect.map(
31 ({ project }) => `alchemy-testing@${project}.iam.gserviceaccount.com`,
32 ),
33);
34
35test.provider.skipIf(!!process.env.FAST)(
36 "create, update, and delete a lake task",
37 (stack) =>
38 Effect.gen(function* () {
39 yield* stack.destroy();
40 const serviceAccount = yield* serviceAccountOf;
41
42 const created = yield* stack.deploy(
43 Effect.gen(function* () {
44 const lake = yield* GCP.Dataplex.Lake("Warehouse", {
45 location: "us-central1",
46 labels: { env: "test" },
47 });
48 const task = yield* GCP.Dataplex.LakesTask("SelectOne", {
49 lake: lake.name,
50 displayName: "select one",
51 description: "on demand sql",
52 labels: { env: "test" },
53 triggerSpec: { type: "ON_DEMAND" },
54 executionSpec: {
55 serviceAccount,
56 args: { output_location: "gs://dataplex-task-output/out" },
57 },
58 spark: { sqlScript: "SELECT 1" },
59 });
60 return { lake, task };
61 }),
62 );
63
64 expect(created.task.name).toContain("/tasks/");
65 expect(created.task.taskId).toEqual(expect.any(String));
66 expect(created.task.lake).toEqual(created.lake.name);
67 expect(created.task.triggerType).toEqual("ON_DEMAND");
68 expect(created.task.labels).toMatchObject({ env: "test" });
69
70 const fetched = yield* dataplex.getProjectsLocationsLakesTasks({
71 name: created.task.name,
72 });
73 expect(fetched.name).toEqual(created.task.name);
74 expect(fetched.spark?.sqlScript).toEqual("SELECT 1");
75
76 const updated = yield* stack.deploy(
77 Effect.gen(function* () {
78 const lake = yield* GCP.Dataplex.Lake("Warehouse", {
79 lakeId: created.lake.lakeId,
80 location: "us-central1",
81 labels: { env: "test" },
82 });
83 const task = yield* GCP.Dataplex.LakesTask("SelectOne", {
84 lake: lake.name,
85 taskId: created.task.taskId,
86 displayName: "select two",
87 description: "on demand sql b",
88 labels: { env: "prod", team: "data" },
89 triggerSpec: { type: "ON_DEMAND" },
90 executionSpec: {
91 serviceAccount,
92 args: { output_location: "gs://dataplex-task-output/out" },
93 },
94 spark: { sqlScript: "SELECT 2" },
95 });
96 return { lake, task };
97 }),
98 );
99
100 expect(updated.task.name).toEqual(created.task.name);
101 expect(updated.task.displayName).toEqual("select two");
102 expect(updated.task.description).toEqual("on demand sql b");
103 expect(updated.task.labels).toMatchObject({ env: "prod", team: "data" });
104
105 yield* stack.destroy();
106 const gone = yield* waitUntilGone(created.task.name);
107 expect(gone).toEqual("gone");
108 }).pipe(logLevel, withDataplexSlot),
109 { tags: ["provider:gcp", "provider:gcp:dataplex", "live"], timeout: 900_000 },
110);