1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import { GcpEnvironment } from "@/GCP/Environment";
9import { withKafkaClusterSlot } from "./quota.ts";
10
11const { test } = Test.make({ providers: GCP.providers() });
12
13const logLevel = Effect.provideService(
14 MinimumLogLevel,
15 process.env.DEBUG ? "Debug" : "Info",
16);
17
18// Kafka and Connect clusters take ~30 minutes to provision.
19const runLifecycle = !!process.env.GCP_TEST_SLOW && !process.env.FAST;
20
21const waitUntilGone = (name: string) =>
22 kafka.getProjectsLocationsClustersAcls({ name }).pipe(
23 Effect.as("found" as const),
24 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
25 Effect.repeat({
26 schedule: Schedule.spaced("1 second"),
27 until: (status) => status === "gone",
28 times: 10,
29 }),
30 );
31
32test.provider(
33 "getProjectsLocationsClustersAcls on a missing acl fails with a typed tag",
34 (stack) =>
35 Effect.gen(function* () {
36 const { project } = yield* GcpEnvironment.current;
37
38 yield* stack.destroy();
39
40 const error = yield* Effect.flip(
41 kafka.getProjectsLocationsClustersAcls({
42 name: `projects/${project}/locations/us-central1/clusters/alchemy-missing/acls/cluster`,
43 }),
44 );
45 // GCP has answered a missing parent cluster with both 404 NOT_FOUND
46 // and 400 FAILED_PRECONDITION "cluster must exist"
47 // (AclClusterNotFound); both are typed.
48 expect(["NotFound", "AclClusterNotFound"]).toContain(error._tag);
49
50 yield* stack.destroy();
51 }).pipe(logLevel),
52 {
53 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
54 timeout: 90_000,
55 },
56);
57
58test.provider.skipIf(!runLifecycle)(
59 "create, update, and delete a kafka acl",
60 (stack) =>
61 Effect.gen(function* () {
62 yield* stack.destroy();
63
64 const created = yield* stack.deploy(
65 Effect.gen(function* () {
66 const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {
67 location: "us-central1",
68 labels: { env: "test" },
69 });
70 const acl = yield* GCP.ManagedKafka.ClustersAcl("OrdersAcl", {
71 cluster: cluster.name,
72 aclId: "topic/orders",
73 aclEntries: [
74 {
75 principal: "User:*",
76 operation: "DESCRIBE",
77 permissionType: "ALLOW",
78 host: "*",
79 },
80 ],
81 });
82 return { cluster, acl };
83 }),
84 );
85
86 expect(created.acl.name).toContain("/acls/");
87 expect(created.acl.aclId).toEqual("topic/orders");
88 expect(created.acl.aclEntries.length).toBeGreaterThan(0);
89
90 const fetched = yield* kafka.getProjectsLocationsClustersAcls({
91 name: created.acl.name,
92 });
93 expect(fetched.name).toEqual(created.acl.name);
94
95 const updated = yield* stack.deploy(
96 Effect.gen(function* () {
97 const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {
98 clusterId: created.cluster.clusterId,
99 location: "us-central1",
100 labels: { env: "test" },
101 });
102 const acl = yield* GCP.ManagedKafka.ClustersAcl("OrdersAcl", {
103 cluster: cluster.name,
104 aclId: "topic/orders",
105 aclEntries: [
106 {
107 principal: "User:*",
108 operation: "READ",
109 permissionType: "ALLOW",
110 host: "*",
111 },
112 ],
113 });
114 return { cluster, acl };
115 }),
116 );
117 expect(updated.acl.aclId).toEqual("topic/orders");
118 expect(
119 updated.acl.aclEntries.some(
120 (entry) => (entry.operation ?? "").toUpperCase() === "READ",
121 ),
122 ).toEqual(true);
123
124 yield* stack.destroy();
125
126 const gone = yield* waitUntilGone(created.acl.name);
127 expect(gone).toEqual("gone");
128 }).pipe(logLevel, withKafkaClusterSlot),
129 {
130 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
131 timeout: 10_800_000,
132 retry: 0,
133 },
134);