1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as datalineage from "@distilled.cloud/gcp/datalineage_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";
9
10const { test } = Test.make({ providers: GCP.providers() });
11
12const logLevel = Effect.provideService(
13 MinimumLogLevel,
14 process.env.DEBUG ? "Debug" : "Info",
15);
16
17const waitUntilGone = (name: string) =>
18 datalineage.getProjectsLocationsProcessesRunsLineageEvents({ name }).pipe(
19 Effect.as("found" as const),
20 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
21 Effect.repeat({
22 schedule: Schedule.spaced("1 second"),
23 until: (status) => status === "gone",
24 times: 10,
25 }),
26 );
27
28test.provider.skipIf(!!process.env.FAST)(
29 "create and delete a lineage event",
30 (stack) =>
31 Effect.gen(function* () {
32 const { project } = yield* GcpEnvironment.current;
33
34 yield* stack.destroy();
35
36 const created = yield* stack.deploy(
37 Effect.gen(function* () {
38 const process = yield* GCP.DataLineage.Process("Etl", {
39 location: "us-central1",
40 displayName: "event parent",
41 attributes: { env: "test" },
42 });
43 const run = yield* GCP.DataLineage.ProcessesRun("Nightly", {
44 process: process.name,
45 startTime: "2024-01-01T00:00:00Z",
46 state: "STARTED",
47 attributes: { env: "test" },
48 });
49 const event = yield* GCP.DataLineage.ProcessesRunsLineageEvent(
50 "Load",
51 {
52 run: run.name,
53 startTime: "2024-01-01T00:00:00Z",
54 endTime: "2024-01-01T00:05:00Z",
55 links: [
56 {
57 source: {
58 fullyQualifiedName: `custom:${project}.raw.orders`,
59 },
60 target: {
61 fullyQualifiedName: `custom:${project}.dw.orders`,
62 },
63 },
64 ],
65 },
66 );
67 return { process, run, event };
68 }),
69 );
70
71 expect(created.event.name).toContain("/lineageEvents/");
72 expect(created.event.lineageEventId).toEqual(expect.any(String));
73 expect(created.event.run).toEqual(created.run.name);
74 expect(created.event.process).toEqual(created.process.name);
75 expect(created.event.links.length).toEqual(1);
76 expect(created.event.links[0]?.source.fullyQualifiedName).toContain(
77 "raw.orders",
78 );
79 expect(created.event.links[0]?.target.fullyQualifiedName).toContain(
80 "dw.orders",
81 );
82
83 const fetched =
84 yield* datalineage.getProjectsLocationsProcessesRunsLineageEvents({
85 name: created.event.name,
86 });
87 expect(fetched.name).toEqual(created.event.name);
88 expect(fetched.links?.length).toEqual(1);
89
90 yield* stack.destroy();
91 const gone = yield* waitUntilGone(created.event.name);
92 expect(gone).toEqual("gone");
93 }).pipe(logLevel),
94 {
95 tags: ["provider:gcp", "provider:gcp:datalineage", "live"],
96 timeout: 90_000,
97 },
98);