1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as datastream from "@distilled.cloud/gcp/datastream_v1";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import {
7 LOCATION,
8 logLevel,
9 currentProject,
10 runLifecycle,
11 waitUntilGone,
12} from "./common.ts";
13
14const { test } = Test.make({ providers: GCP.providers() });
15
16test.provider(
17 "getProjectsLocationsStreams on a missing stream fails with a typed tag",
18 (stack) =>
19 Effect.gen(function* () {
20 const project = yield* currentProject;
21 yield* stack.destroy();
22
23 const error = yield* Effect.flip(
24 datastream.getProjectsLocationsStreams({
25 name: `projects/${project}/locations/${LOCATION}/streams/alchemy-missing-stream`,
26 }),
27 );
28 expect(error._tag).toEqual("NotFound");
29
30 yield* stack.destroy();
31 }).pipe(logLevel),
32 {
33 tags: ["provider:gcp", "provider:gcp:datastream", "live"],
34 timeout: 90_000,
35 },
36);
37
38test.provider.skipIf(!runLifecycle)(
39 "create and delete a mysql-to-bigquery stream",
40 (stack) =>
41 Effect.gen(function* () {
42 const project = yield* currentProject;
43 yield* stack.destroy();
44
45 const created = yield* stack.deploy(
46 Effect.gen(function* () {
47 const source = yield* GCP.Datastream.ConnectionProfile("MysqlSrc", {
48 location: LOCATION,
49 displayName: "mysql-src",
50 mysqlProfile: {
51 hostname: "10.0.0.8",
52 port: 3306,
53 username: "alchemy",
54 password: "AlchemyTestPass1",
55 },
56 staticServiceIpConnectivity: {},
57 force: true,
58 });
59 const dest = yield* GCP.Datastream.ConnectionProfile("BqDest", {
60 location: LOCATION,
61 displayName: "bq-dest",
62 bigqueryProfile: {},
63 });
64 const stream = yield* GCP.Datastream.Stream("MysqlToBq", {
65 location: LOCATION,
66 displayName: "mysql-to-bq",
67 labels: { env: "test" },
68 sourceConfig: {
69 sourceConnectionProfile: source.name,
70 mysqlSourceConfig: {
71 includeObjects: {
72 mysqlDatabases: [{ database: "alchemy" }],
73 },
74 },
75 },
76 destinationConfig: {
77 destinationConnectionProfile: dest.name,
78 bigqueryDestinationConfig: {
79 singleTargetDataset: {
80 datasetId: `${project}:alchemy_ds`,
81 },
82 dataFreshness: "900s",
83 },
84 },
85 backfillNone: {},
86 force: true,
87 });
88 return { source, dest, stream };
89 }),
90 );
91
92 expect(created.stream.streamId).toEqual(expect.any(String));
93 expect(created.stream.name).toEqual(
94 `projects/${project}/locations/${LOCATION}/streams/${created.stream.streamId}`,
95 );
96 expect(created.stream.location).toEqual(LOCATION);
97 expect(created.stream.displayName).toEqual("mysql-to-bq");
98 expect(created.stream.labels).toMatchObject({ env: "test" });
99 expect(created.stream.sourceConfig?.sourceConnectionProfile).toContain(
100 created.source.connectionProfileId,
101 );
102 expect(
103 created.stream.destinationConfig?.destinationConnectionProfile,
104 ).toContain(created.dest.connectionProfileId);
105
106 const fetched = yield* datastream.getProjectsLocationsStreams({
107 name: created.stream.name,
108 });
109 expect(fetched.name).toEqual(created.stream.name);
110 expect(fetched.displayName).toEqual("mysql-to-bq");
111 expect(fetched.labels?.env).toEqual("test");
112
113 yield* stack.destroy();
114 const gone = yield* waitUntilGone(
115 datastream.getProjectsLocationsStreams({
116 name: created.stream.name,
117 }),
118 );
119 expect(gone).toEqual("gone");
120 }).pipe(logLevel),
121 {
122 tags: ["provider:gcp", "provider:gcp:datastream", "live"],
123 timeout: 600_000,
124 },
125);