1import * as GCP from "@/GCP";
2import { GcpEnvironment } from "@/GCP/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as Core from "@/Test/Core";
5import * as composer from "@distilled.cloud/gcp/composer_v1";
6import * as crm from "@distilled.cloud/gcp/cloudresourcemanager_v3";
7import { describe, expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import { dockerAvailable, expectProbe } from "../bindingHost.ts";
11import ComposerBindingsHost, {
12 Airflow,
13 TaskConfig,
14 TaskSecret,
15} from "./fixtures/bindings-host.ts";
16
17const testOptions = { providers: GCP.providers() };
18const { test, beforeAll, afterAll } = Test.make(testOptions);
19const sharedStack = Core.scratchStack(testOptions, "ComposerBindings");
20
21// Composer environments take 20-45 minutes to provision.
22const runLifecycle = !!process.env.GCP_TEST_SLOW && !process.env.FAST;
23
24let baseUrl: string;
25let hostAccount: string;
26let environmentName: string;
27let configMapName: string;
28let secretName: string;
29
30/**
31 * Composer has no resource-level IAM, so its bindings grant on the project:
32 * `composer.viewer` (the Get* bindings) and `composer.editor`
33 * (ExecuteAirflowCommand — the narrowest role holding
34 * `composer.environments.executeAirflowCommand`).
35 */
36const expectProjectGrants = Effect.gen(function* () {
37 const { project } = yield* GcpEnvironment.current;
38 const policy = yield* crm.getIamPolicyProjects({
39 resource: `projects/${project}`,
40 body: { options: { requestedPolicyVersion: 3 } },
41 });
42 const grants = (policy.bindings ?? [])
43 .filter((binding) =>
44 (binding.members ?? []).includes(`serviceAccount:${hostAccount}`),
45 )
46 .map((binding) => ({
47 role: binding.role,
48 condition: binding.condition?.expression,
49 }))
50 .sort((a, b) => (a.role ?? "").localeCompare(b.role ?? ""));
51 expect(grants).toEqual([
52 { role: "roles/composer.editor", condition: undefined },
53 { role: "roles/composer.viewer", condition: undefined },
54 ]);
55});
56
57describe.skipIf(!dockerAvailable || !runLifecycle)(
58 "Composer Bindings",
59 {
60 tags: ["provider:gcp", "provider:gcp:composer", "provider:gcp:run", "live"],
61 },
62 () => {
63 beforeAll(
64 Effect.gen(function* () {
65 yield* sharedStack.destroy();
66 const out = yield* sharedStack.deploy(
67 Effect.gen(function* () {
68 const host = yield* ComposerBindingsHost;
69 const environment = yield* Airflow;
70 const config = yield* TaskConfig;
71 const secret = yield* TaskSecret;
72 return {
73 uri: host.uri,
74 serviceAccount: host.serviceAccount,
75 environment: environment.name,
76 config: config.name,
77 secret: secret.name,
78 };
79 }),
80 );
81 baseUrl = out.uri!;
82 hostAccount = out.serviceAccount!;
83 environmentName = out.environment;
84 configMapName = out.config;
85 secretName = out.secret;
86 }),
87 // Composer environment create alone can take ~45 minutes.
88 { timeout: 5_400_000 },
89 );
90
91 afterAll(sharedStack.destroy(), { timeout: 3_600_000 });
92
93 describe("GetEnvironment", () => {
94 test.provider(
95 "reads the environment as the host's service account",
96 (_stack) =>
97 Effect.gen(function* () {
98 const environment = yield* expectProbe<composer.Environment>(
99 baseUrl,
100 "getEnvironment",
101 );
102 const live = yield* composer.getProjectsLocationsEnvironments({
103 name: environmentName,
104 });
105 expect(environment.name).toEqual(environmentName);
106 expect(environment.uuid).toEqual(live.uuid);
107 yield* expectProjectGrants;
108 }),
109 {
110 tags: ["provider:gcp", "provider:gcp:composer", "live"],
111 timeout: 600_000,
112 },
113 );
114 });
115
116 describe("ExecuteAirflowCommand", () => {
117 test.provider(
118 "runs `airflow version` as the host's service account",
119 (_stack) =>
120 Effect.gen(function* () {
121 const started =
122 yield* expectProbe<composer.ExecuteAirflowCommandResponse>(
123 baseUrl,
124 "executeAirflowCommand",
125 );
126 expect(started.error ?? "").toEqual("");
127 expect(started.executionId).toEqual(expect.any(String));
128
129 // The execution is visible out of band and finishes cleanly.
130 const polled = yield* composer
131 .pollAirflowCommandProjectsLocationsEnvironments({
132 environment: environmentName,
133 body: {
134 executionId: started.executionId,
135 pod: started.pod,
136 podNamespace: started.podNamespace,
137 // The API rejects a missing or zero line number.
138 nextLineNumber: 1,
139 },
140 })
141 .pipe(
142 Effect.repeat({
143 schedule: Schedule.spaced("5 seconds"),
144 until: (response) => response.outputEnd === true,
145 times: 24,
146 }),
147 );
148 expect(polled.exitInfo?.exitCode ?? 0).toEqual(0);
149 yield* expectProjectGrants;
150 }),
151 {
152 tags: ["provider:gcp", "provider:gcp:composer", "live"],
153 timeout: 600_000,
154 },
155 );
156 });
157
158 describe("GetUserWorkloadsConfigMap", () => {
159 test.provider(
160 "reads the ConfigMap as the host's service account",
161 (_stack) =>
162 Effect.gen(function* () {
163 const configMap =
164 yield* expectProbe<composer.UserWorkloadsConfigMap>(
165 baseUrl,
166 "getUserWorkloadsConfigMap",
167 );
168 expect(configMap.name).toEqual(configMapName);
169 expect(configMap.data).toEqual({ LOG_LEVEL: "INFO" });
170 yield* expectProjectGrants;
171 }),
172 {
173 tags: ["provider:gcp", "provider:gcp:composer", "live"],
174 timeout: 600_000,
175 },
176 );
177 });
178
179 describe("GetUserWorkloadsSecret", () => {
180 test.provider(
181 "reads the Secret as the host's service account",
182 (_stack) =>
183 Effect.gen(function* () {
184 const secret = yield* expectProbe<composer.UserWorkloadsSecret>(
185 baseUrl,
186 "getUserWorkloadsSecret",
187 );
188 expect(secret.name).toEqual(secretName);
189 // Composer redacts secret values on read; the key survives.
190 expect(Object.keys(secret.data ?? {})).toEqual(["password"]);
191 yield* expectProjectGrants;
192 }),
193 {
194 tags: ["provider:gcp", "provider:gcp:composer", "live"],
195 timeout: 600_000,
196 },
197 );
198 });
199 },
200);