1import * as GCP from "@/GCP";
2import { GcpEnvironment } from "@/GCP/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as Core from "@/Test/Core";
5import * as resourcemanager from "@distilled.cloud/gcp/cloudresourcemanager_v3";
6import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
7import { describe, expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import { dockerAvailable, expectProbe } from "../bindingHost.ts";
10import KafkaBindingsHost, {
11 Brokers,
12 Connect,
13 Events,
14 Schemas,
15} from "./fixtures/bindings-host.ts";
16import { releaseKafkaClusterSlot, takeKafkaClusterSlot } from "./quota.ts";
17
18const testOptions = { providers: GCP.providers() };
19const { test, beforeAll, afterAll } = Test.make(testOptions);
20const sharedStack = Core.scratchStack(testOptions, "ManagedKafkaBindings");
21
22// Kafka and Connect clusters take ~30 minutes to provision.
23const runLifecycle = !!process.env.GCP_TEST_SLOW && !process.env.FAST;
24
25let baseUrl: string;
26let hostAccount: string;
27let clusterName: string;
28let topicName: string;
29let connectName: string;
30let registryName: string;
31
32/**
33 * Managed Kafka has no per-resource IAM policy: bindings grant on the
34 * project under an IAM Condition matching the bound resource.
35 */
36const hostProjectGrants = Effect.gen(function* () {
37 const { project } = yield* GcpEnvironment.current;
38 const policy = yield* resourcemanager.getIamPolicyProjects({
39 resource: `projects/${project}`,
40 body: { options: { requestedPolicyVersion: 3 } },
41 });
42 return (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});
51
52/** Granted on the project, unconditioned (see GetSchemaRegistryHttp). */
53const SCHEMA_REGISTRY_ROLE = "roles/managedkafka.schemaRegistryViewer";
54
55const scopedTo = (name: string) =>
56 `resource.name == "${name}" || resource.name.startsWith("${name}/")`;
57
58/** The host holds `role` only under the condition scoped to `name`. */
59const expectScopedGrant = (role: string, name: string) =>
60 Effect.gen(function* () {
61 const grants = yield* hostProjectGrants;
62 expect(grants).toContainEqual({ role, condition: scopedTo(name) });
63 expect(grants).toEqual(
64 expect.arrayContaining([
65 { role: SCHEMA_REGISTRY_ROLE, condition: undefined },
66 ]),
67 );
68 // Every grant except the schema-registry viewer is condition-scoped.
69 expect(
70 grants
71 .filter((grant) => grant.role !== SCHEMA_REGISTRY_ROLE)
72 .every((grant) => grant.condition !== undefined),
73 ).toBe(true);
74 expect(grants).toHaveLength(4);
75 });
76
77describe.skipIf(!dockerAvailable || !runLifecycle)(
78 "ManagedKafka Bindings",
79 {
80 tags: [
81 "provider:gcp",
82 "provider:gcp:managedkafka",
83 "provider:gcp:run",
84 "live",
85 ],
86 },
87 () => {
88 beforeAll(
89 Effect.gen(function* () {
90 yield* takeKafkaClusterSlot;
91 yield* sharedStack.destroy();
92 const out = yield* sharedStack.deploy(
93 Effect.gen(function* () {
94 const host = yield* KafkaBindingsHost;
95 const cluster = yield* Brokers;
96 const topic = yield* Events;
97 const connect = yield* Connect;
98 const registry = yield* Schemas;
99 return {
100 uri: host.uri,
101 serviceAccount: host.serviceAccount,
102 cluster: cluster.name,
103 topic: topic.name,
104 connect: connect.name,
105 registry: registry.name,
106 };
107 }),
108 );
109 baseUrl = out.uri!;
110 hostAccount = out.serviceAccount!;
111 clusterName = out.cluster;
112 topicName = out.topic;
113 connectName = out.connect;
114 registryName = out.registry;
115 }),
116 { timeout: 10_800_000 },
117 );
118
119 afterAll(
120 sharedStack.destroy().pipe(Effect.ensuring(releaseKafkaClusterSlot)),
121 { timeout: 3_600_000 },
122 );
123
124 describe("GetCluster", () => {
125 test.provider(
126 "reads the cluster as the host's service account",
127 (_stack) =>
128 Effect.gen(function* () {
129 const out = yield* expectProbe<kafka.Cluster>(
130 baseUrl,
131 "getCluster",
132 );
133 const direct = yield* kafka.getProjectsLocationsClusters({
134 name: clusterName,
135 });
136 expect(out.name).toEqual(clusterName);
137 expect(out.state).toEqual(direct.state);
138 yield* expectScopedGrant("roles/managedkafka.viewer", clusterName);
139 }),
140 {
141 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
142 timeout: 600_000,
143 },
144 );
145 });
146
147 describe("GetTopic", () => {
148 test.provider(
149 "reads the topic as the host's service account",
150 (_stack) =>
151 Effect.gen(function* () {
152 const out = yield* expectProbe<kafka.Topic>(baseUrl, "getTopic");
153 const direct = yield* kafka.getProjectsLocationsClustersTopics({
154 name: topicName,
155 });
156 expect(out.name).toEqual(topicName);
157 expect(out.partitionCount).toEqual(direct.partitionCount);
158 yield* expectScopedGrant("roles/managedkafka.viewer", topicName);
159 }),
160 {
161 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
162 timeout: 600_000,
163 },
164 );
165 });
166
167 describe("GetConnectCluster", () => {
168 test.provider(
169 "reads the Connect cluster as the host's service account",
170 (_stack) =>
171 Effect.gen(function* () {
172 const out = yield* expectProbe<kafka.ConnectCluster>(
173 baseUrl,
174 "getConnectCluster",
175 );
176 const direct = yield* kafka.getProjectsLocationsConnectClusters({
177 name: connectName,
178 });
179 expect(out.name).toEqual(connectName);
180 expect(out.kafkaCluster).toEqual(direct.kafkaCluster);
181 yield* expectScopedGrant("roles/managedkafka.viewer", connectName);
182 }),
183 {
184 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
185 timeout: 600_000,
186 },
187 );
188 });
189
190 describe("GetSchemaRegistry", () => {
191 test.provider(
192 "reads the schema registry as the host's service account",
193 (_stack) =>
194 Effect.gen(function* () {
195 const out = yield* expectProbe<kafka.SchemaRegistry>(
196 baseUrl,
197 "getSchemaRegistry",
198 );
199 const direct = yield* kafka.getProjectsLocationsSchemaRegistries({
200 name: registryName,
201 });
202 expect(out.name).toEqual(registryName);
203 expect(out.contexts).toEqual(direct.contexts);
204 const grants = yield* hostProjectGrants;
205 expect(grants).toContainEqual({
206 role: SCHEMA_REGISTRY_ROLE,
207 condition: undefined,
208 });
209 expect(grants).toHaveLength(4);
210 }),
211 {
212 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
213 timeout: 600_000,
214 },
215 );
216 });
217 },
218);