1import * as GCP from "@/GCP";
2import { GcpEnvironment } from "@/GCP/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as Core from "@/Test/Core";
5import * as crm from "@distilled.cloud/gcp/cloudresourcemanager_v3";
6import * as dataflow from "@distilled.cloud/gcp/dataflow_v1b3";
7import * as datapipelines from "@distilled.cloud/gcp/datapipelines_v1";
8import { describe, expect } from "alchemy-test";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import { dockerAvailable, expectProbe } from "../bindingHost.ts";
12import DataPipelinesBindingsHost, {
13 BindBatch,
14} from "./fixtures/bindings-host.ts";
15
16const testOptions = { providers: GCP.providers() };
17const { test, beforeAll, afterAll } = Test.make(testOptions);
18const sharedStack = Core.scratchStack(testOptions, "DataPipelinesBindings");
19
20let baseUrl: string;
21let pipelineName: string;
22let hostAccount: string;
23
24/** Roles the host's service account holds on the project policy. */
25const projectRoles = Effect.gen(function* () {
26 const { project } = yield* GcpEnvironment.current;
27 const policy = yield* crm.getIamPolicyProjects({
28 resource: `projects/${project}`,
29 body: { options: { requestedPolicyVersion: 3 } },
30 });
31 return (policy.bindings ?? [])
32 .filter((binding) =>
33 (binding.members ?? []).includes(`serviceAccount:${hostAccount}`),
34 )
35 .map((binding) => ({
36 role: binding.role,
37 condition: binding.condition?.expression,
38 }))
39 .sort((a, b) => (a.role ?? "").localeCompare(b.role ?? ""));
40});
41
42// Data Pipelines has no resource-level IAM, so both bindings grant on the
43// project (unconditioned).
44const expectedProjectRoles = [
45 { role: "roles/datapipelines.admin", condition: undefined },
46 { role: "roles/datapipelines.invoker", condition: undefined },
47];
48
49// RunPipeline must run before StopPipeline archives the pipeline.
50describe.skipIf(!dockerAvailable)(
51 "DataPipelines Bindings",
52 {
53 sequential: true,
54 tags: [
55 "provider:gcp",
56 "provider:gcp:datapipelines",
57 "provider:gcp:run",
58 "live",
59 ],
60 },
61 () => {
62 beforeAll(
63 Effect.gen(function* () {
64 yield* sharedStack.destroy();
65 const out = yield* sharedStack.deploy(
66 Effect.gen(function* () {
67 const host = yield* DataPipelinesBindingsHost;
68 const pipeline = yield* BindBatch;
69 return {
70 uri: host.uri,
71 serviceAccount: host.serviceAccount,
72 pipeline: pipeline.name,
73 };
74 }),
75 );
76 baseUrl = out.uri!;
77 pipelineName = out.pipeline;
78 hostAccount = out.serviceAccount!;
79 }),
80 { timeout: 900_000 },
81 );
82
83 afterAll(sharedStack.destroy(), { timeout: 600_000 });
84
85 describe("RunPipeline", () => {
86 test.provider(
87 "launches a pipeline job as the host's service account",
88 (_stack) =>
89 Effect.gen(function* () {
90 const ran =
91 yield* expectProbe<datapipelines.GoogleCloudDatapipelinesV1RunPipelineResponse>(
92 baseUrl,
93 "runPipeline",
94 );
95 expect(ran.job?.id).toEqual(expect.any(String));
96
97 const jobIds = yield* datapipelines
98 .listProjectsLocationsPipelinesJobs({ parent: pipelineName })
99 .pipe(
100 Effect.map((page) => (page.jobs ?? []).map((job) => job.id)),
101 Effect.repeat({
102 schedule: Schedule.spaced("3 seconds"),
103 until: (ids) => ids.includes(ran.job?.id),
104 times: 20,
105 }),
106 );
107 expect(jobIds).toContain(ran.job?.id);
108
109 // The pipeline job is a real Dataflow job (Word Count on one file:
110 // it finishes, or fails once the scratch bucket is destroyed, on
111 // its own within minutes).
112 const { project } = yield* GcpEnvironment.current;
113 const job = yield* dataflow.getProjectsLocationsJobs({
114 projectId: project,
115 location: "us-central1",
116 jobId: ran.job!.id!,
117 });
118 expect(job.id).toEqual(ran.job?.id);
119
120 expect(yield* projectRoles).toEqual(expectedProjectRoles);
121 }),
122 {
123 tags: ["provider:gcp", "provider:gcp:datapipelines", "live"],
124 timeout: 600_000,
125 },
126 );
127 });
128
129 describe("StopPipeline", () => {
130 test.provider(
131 "archives the pipeline as the host's service account",
132 (_stack) =>
133 Effect.gen(function* () {
134 const stopped =
135 yield* expectProbe<datapipelines.GoogleCloudDatapipelinesV1Pipeline>(
136 baseUrl,
137 "stopPipeline",
138 );
139 expect(stopped.name).toEqual(pipelineName);
140 expect(stopped.state).toEqual("STATE_ARCHIVED");
141
142 const live = yield* datapipelines.getProjectsLocationsPipelines({
143 name: pipelineName,
144 });
145 expect(live.state).toEqual("STATE_ARCHIVED");
146
147 expect(yield* projectRoles).toEqual(expectedProjectRoles);
148 }),
149 {
150 tags: ["provider:gcp", "provider:gcp:datapipelines", "live"],
151 timeout: 600_000,
152 },
153 );
154 });
155 },
156);