1import { GcpEnvironment } from "@/GCP/Environment";
2import * as GCP from "@/GCP";
3import * as Output from "@/Output";
4import * as Test from "@/Test/Alchemy";
5import * as datapipelines from "@distilled.cloud/gcp/datapipelines_v1";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
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 datapipelines.getProjectsLocationsPipelines({ name }).pipe(
20 Effect.as("found" as const),
21 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
22 Effect.repeat({
23 schedule: Schedule.spaced("1 second"),
24 until: (status) => status === "gone",
25 times: 10,
26 }),
27 );
28
29test.provider(
30 "getProjectsLocationsPipelines on a missing pipeline fails with a typed tag",
31 (stack) =>
32 Effect.gen(function* () {
33 const { project } = yield* GcpEnvironment.current;
34 const parent = `projects/${project}/locations/us-central1`;
35
36 yield* stack.destroy();
37
38 const error = yield* Effect.flip(
39 datapipelines.getProjectsLocationsPipelines({
40 name: `${parent}/pipelines/alchemy-missing-pipeline`,
41 }),
42 );
43 expect(error._tag).toEqual("NotFound");
44
45 const page = yield* datapipelines.listProjectsLocationsPipelines({
46 parent,
47 pageSize: 10,
48 });
49 expect(
50 (page.pipelines ?? []).map((pipeline) =>
51 pipeline.name?.split("/").pop(),
52 ),
53 ).not.toContain("alchemy-missing-pipeline");
54
55 yield* stack.destroy();
56 }).pipe(logLevel),
57 {
58 tags: ["provider:gcp", "provider:gcp:datapipelines", "live"],
59 timeout: 90_000,
60 },
61);
62
63test.provider(
64 "create, update, and delete a data pipeline",
65 (stack) =>
66 Effect.gen(function* () {
67 const { project } = yield* GcpEnvironment.current;
68
69 yield* stack.destroy();
70
71 const created = yield* stack.deploy(
72 Effect.gen(function* () {
73 const bucket = yield* GCP.Storage.Bucket("PipelineTmp", {
74 location: "US-CENTRAL1",
75 forceDestroy: true,
76 });
77 return yield* GCP.DataPipelines.Pipeline("WordCount", {
78 type: "PIPELINE_TYPE_BATCH",
79 displayName: "word-count",
80 scheduleInfo: { schedule: "0 0 1 1 *", timeZone: "UTC" },
81 workload: {
82 dataflowLaunchTemplateRequest: {
83 projectId: project,
84 location: "us-central1",
85 gcsPath: "gs://dataflow-templates/latest/Word_Count",
86 launchParameters: {
87 jobName: "alchemy-word-count",
88 parameters: {
89 inputFile: "gs://dataflow-samples/shakespeare/kinglear.txt",
90 output: Output.interpolate`gs://${bucket.bucketName}/out`,
91 },
92 environment: {
93 tempLocation: Output.interpolate`gs://${bucket.bucketName}/tmp`,
94 },
95 },
96 },
97 },
98 });
99 }),
100 );
101
102 expect(created.name).toContain("/pipelines/");
103 expect(created.location).toEqual("us-central1");
104 expect(created.displayName).toEqual("word-count");
105 expect(created.type).toEqual("PIPELINE_TYPE_BATCH");
106 expect(created.scheduleInfo?.schedule).toEqual("0 0 1 1 *");
107 expect(created.pipelineId.length).toBeGreaterThan(0);
108
109 const fetched = yield* datapipelines.getProjectsLocationsPipelines({
110 name: created.name,
111 });
112 expect(fetched.name).toEqual(created.name);
113 expect(fetched.displayName).toEqual("word-count");
114 expect(fetched.pipelineSources?.["alchemy-id"]).toEqual(
115 expect.any(String),
116 );
117 expect(fetched.type).toEqual("PIPELINE_TYPE_BATCH");
118
119 const updated = yield* stack.deploy(
120 Effect.gen(function* () {
121 const bucket = yield* GCP.Storage.Bucket("PipelineTmp", {
122 location: "US-CENTRAL1",
123 forceDestroy: true,
124 });
125 return yield* GCP.DataPipelines.Pipeline("WordCount", {
126 pipelineId: created.pipelineId,
127 location: created.location,
128 type: "PIPELINE_TYPE_BATCH",
129 displayName: "word-count-v2",
130 scheduleInfo: { schedule: "0 0 1 2 *", timeZone: "UTC" },
131 workload: {
132 dataflowLaunchTemplateRequest: {
133 projectId: project,
134 location: "us-central1",
135 gcsPath: "gs://dataflow-templates/latest/Word_Count",
136 launchParameters: {
137 jobName: "alchemy-word-count",
138 parameters: {
139 inputFile: "gs://dataflow-samples/shakespeare/kinglear.txt",
140 output: Output.interpolate`gs://${bucket.bucketName}/out-v2`,
141 },
142 environment: {
143 tempLocation: Output.interpolate`gs://${bucket.bucketName}/tmp`,
144 },
145 },
146 },
147 },
148 });
149 }),
150 );
151
152 expect(updated.name).toEqual(created.name);
153 expect(updated.displayName).toEqual("word-count-v2");
154 expect(updated.scheduleInfo?.schedule).toEqual("0 0 1 2 *");
155
156 const fetchedUpdate = yield* datapipelines.getProjectsLocationsPipelines({
157 name: updated.name,
158 });
159 expect(fetchedUpdate.displayName).toEqual("word-count-v2");
160 expect(fetchedUpdate.scheduleInfo?.schedule).toEqual("0 0 1 2 *");
161 expect(fetchedUpdate.pipelineSources?.["alchemy-id"]).toEqual(
162 expect.any(String),
163 );
164
165 yield* stack.destroy();
166
167 const gone = yield* waitUntilGone(updated.name);
168 expect(gone).toEqual("gone");
169 }).pipe(logLevel),
170 {
171 tags: ["provider:gcp", "provider:gcp:datapipelines", "live"],
172 timeout: 120_000,
173 },
174);