1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as Core from "@/Test/Core";
4import * as crm from "@distilled.cloud/gcp/cloudresourcemanager_v3";
5import * as dataproc from "@distilled.cloud/gcp/dataproc_v1";
6import { describe, expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { dockerAvailable, expectProbe } from "../bindingHost.ts";
9import { CAPACITY_REGION } from "../zones.ts";
10import DataprocBindingsHost, { Jobs } from "./fixtures/bindings-host.ts";
11
12const testOptions = { providers: GCP.providers() };
13const { test, beforeAll, afterAll } = Test.make(testOptions);
14const sharedStack = Core.scratchStack(testOptions, "DataprocBindings");
15
16// Cluster create + delete takes ~8 minutes (observed 475s).
17const runLifecycle = !!process.env.GCP_TEST_SLOW && !process.env.FAST;
18
19let baseUrl: string;
20let hostAccount: string;
21let clusterName: string;
22let clusterResource: string;
23let project: string;
24
25const rolesOf = (
26 bindings: ReadonlyArray<{ role?: string; members?: ReadonlyArray<string> }>,
27) =>
28 bindings
29 .filter((binding) =>
30 (binding.members ?? []).includes(`serviceAccount:${hostAccount}`),
31 )
32 .map((binding) => binding.role)
33 .sort();
34
35const projectRoles = Effect.gen(function* () {
36 const policy = yield* crm.getIamPolicyProjects({
37 resource: `projects/${project}`,
38 body: { options: { requestedPolicyVersion: 3 } },
39 });
40 return rolesOf(policy.bindings ?? []);
41});
42
43describe.skipIf(!dockerAvailable || !runLifecycle)(
44 "Dataproc Bindings",
45 {
46 tags: ["provider:gcp", "provider:gcp:dataproc", "provider:gcp:run", "live"],
47 },
48 () => {
49 beforeAll(
50 Effect.gen(function* () {
51 yield* sharedStack.destroy();
52 const out = yield* sharedStack.deploy(
53 Effect.gen(function* () {
54 const host = yield* DataprocBindingsHost;
55 const cluster = yield* Jobs;
56 return {
57 uri: host.uri,
58 serviceAccount: host.serviceAccount,
59 clusterName: cluster.clusterName,
60 name: cluster.name,
61 project: cluster.project,
62 };
63 }),
64 );
65 baseUrl = out.uri!;
66 hostAccount = out.serviceAccount!;
67 clusterName = out.clusterName;
68 clusterResource = out.name;
69 project = out.project;
70 }),
71 { timeout: 1_800_000 },
72 );
73
74 afterAll(sharedStack.destroy(), { timeout: 1_200_000 });
75
76 describe("GetCluster", () => {
77 test.provider(
78 "reads the cluster as the host's service account, granted on the cluster only",
79 (_stack) =>
80 Effect.gen(function* () {
81 const cluster = yield* expectProbe<dataproc.Cluster>(
82 baseUrl,
83 "getCluster",
84 );
85 expect(cluster.clusterName).toEqual(clusterName);
86 expect(cluster.status?.state).toEqual("RUNNING");
87
88 const policy = yield* dataproc.getIamPolicyProjectsRegionsClusters({
89 resource: clusterResource,
90 });
91 expect(rolesOf(policy.bindings ?? [])).toEqual([
92 "roles/dataproc.viewer",
93 ]);
94 expect(yield* projectRoles).not.toContain("roles/dataproc.viewer");
95 }),
96 {
97 tags: ["provider:gcp", "provider:gcp:dataproc", "live"],
98 timeout: 600_000,
99 },
100 );
101 });
102
103 describe("SubmitJob", () => {
104 test.provider(
105 "submits a job to the cluster as the host's service account",
106 (_stack) =>
107 Effect.gen(function* () {
108 const job = yield* expectProbe<dataproc.Job>(baseUrl, "submitJob");
109 expect(job.placement?.clusterName).toEqual(clusterName);
110 expect(job.reference?.jobId).toEqual(expect.any(String));
111
112 const live = yield* dataproc.getProjectsRegionsJobs({
113 projectId: project,
114 region: CAPACITY_REGION,
115 jobId: job.reference!.jobId!,
116 });
117 expect(live.placement?.clusterName).toEqual(clusterName);
118 expect(live.pigJob?.queryList?.queries).toEqual(["DUMP;"]);
119
120 // dataproc.jobs.create is checked on the project.
121 expect(yield* projectRoles).toEqual(["roles/dataproc.editor"]);
122 }),
123 {
124 tags: ["provider:gcp", "provider:gcp:dataproc", "live"],
125 timeout: 600_000,
126 },
127 );
128 });
129 },
130);