1import * as AWS from "@/AWS";
2import { AWSEnvironment } from "@/AWS/Environment";
3import * as Output from "@/Output";
4import * as Test from "@/Test/Alchemy";
5import * as osis from "@distilled.cloud/aws/osis";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9
10const { test } = Test.make({ providers: AWS.providers() });
11
12// Ungated typed-error probe: proves the distilled error union carries the
13// not-found tag this provider's read/delete paths depend on.
14test.provider(
15 "getPipeline on a nonexistent pipeline fails with ResourceNotFoundException",
16 () =>
17 Effect.gen(function* () {
18 const error = yield* Effect.flip(
19 osis.getPipeline({ PipelineName: "alchemy-nonexistent-probe" }),
20 );
21 expect(error._tag).toBe("ResourceNotFoundException");
22 }),
23 { tags: ["provider:aws", "provider:aws:osis", "live"] },
24);
25
26// Ungated probe for the ResourcePolicy provider's read path: OSIS reports
27// "no policy" as a SUCCESS response with the empty document `"{}"` (not a
28// ResourceNotFoundException) — the provider's absent-detection depends on
29// this observed behavior.
30test.provider(
31 "getResourcePolicy on a nonexistent pipeline returns the empty document",
32 () =>
33 Effect.gen(function* () {
34 const { region, accountId } = yield* AWSEnvironment.current;
35 const response = yield* osis.getResourcePolicy({
36 ResourceArn: `arn:aws:osis:${region}:${accountId}:pipeline/alchemy-nonexistent-probe`,
37 });
38 expect(response.Policy === undefined || response.Policy === "{}").toBe(
39 true,
40 );
41 }),
42 { tags: ["provider:aws", "provider:aws:osis", "live"] },
43);
44
45// Ungated probe for the PipelineEndpoint provider's observe path: the list
46// API answers (typed) even when the account has no endpoints.
47test.provider(
48 "listPipelineEndpoints succeeds on an empty account",
49 () =>
50 Effect.gen(function* () {
51 const response = yield* osis.listPipelineEndpoints({});
52 expect(Array.isArray(response.PipelineEndpoints ?? [])).toBe(true);
53 }),
54 { tags: ["provider:aws", "provider:aws:osis", "live"] },
55);
56
57// Data Prepper configuration: HTTP source draining to an S3 sink. The role
58// and bucket are provisioned by the same stack; interpolation defers the
59// concrete values until deploy time.
60const pipelineConfig = (
61 roleArn: Output.Output<string>,
62 bucketName: Output.Output<string>,
63 region: string,
64) => Output.interpolate`version: "2"
65log-pipeline:
66 source:
67 http:
68 path: "/logs/ingest"
69 sink:
70 - s3:
71 aws:
72 sts_role_arn: "${roleArn}"
73 region: "${region}"
74 bucket: "${bucketName}"
75 threshold:
76 event_collect_timeout: "60s"
77 codec:
78 ndjson:
79`;
80
81// Deletion is verified as INITIATED (status DELETING, irreversible) or fully
82// gone. Full disappearance takes a few more minutes server-side.
83const assertPipelineDeleting = (name: string) =>
84 Effect.gen(function* () {
85 const status = yield* osis.getPipeline({ PipelineName: name }).pipe(
86 Effect.map((response) => response.Pipeline?.Status ?? "gone"),
87 Effect.catchTag("ResourceNotFoundException", () =>
88 Effect.succeed("gone" as const),
89 ),
90 );
91 if (status !== "gone" && status !== "DELETING") {
92 return yield* Effect.fail(
93 new Error(`pipeline '${name}' still exists (status: ${status})`),
94 );
95 }
96 }).pipe(
97 Effect.retry({
98 schedule: Schedule.max([
99 Schedule.fixed("10 seconds"),
100 Schedule.recurs(18),
101 ]),
102 }),
103 );
104
105// OSIS pipelines take ~5-10 minutes to provision and are billed per
106// Ingestion-OCU-hour while they exist (minimum 1 OCU). The full lifecycle —
107// pipeline + resource policy + VPC pipeline endpoint — is gated behind
108// AWS_TEST_SLOW=1 and always destroys what it created.
109test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
110 "create http-to-s3 pipeline with resource policy and VPC endpoint, verify, destroy",
111 (stack) =>
112 Effect.gen(function* () {
113 yield* stack.destroy();
114
115 const { region, accountId } = yield* AWSEnvironment.current;
116
117 const { pipeline, policy, endpoint } = yield* stack.deploy(
118 Effect.gen(function* () {
119 const bucket = yield* AWS.S3.Bucket("Sink", {
120 forceDestroy: true,
121 });
122 const role = yield* AWS.IAM.Role("PipelineRole", {
123 assumeRolePolicyDocument: {
124 Version: "2012-10-17",
125 Statement: [
126 {
127 Effect: "Allow",
128 Principal: { Service: ["osis-pipelines.amazonaws.com"] },
129 Action: ["sts:AssumeRole"],
130 },
131 ],
132 },
133 inlinePolicies: {
134 "s3-sink": {
135 Version: "2012-10-17",
136 Statement: [
137 {
138 Effect: "Allow",
139 Action: ["s3:PutObject", "s3:AbortMultipartUpload"],
140 Resource: [Output.interpolate`${bucket.bucketArn}/*`],
141 },
142 {
143 Effect: "Allow",
144 Action: ["s3:GetBucketLocation", "s3:ListBucket"],
145 Resource: [bucket.bucketArn],
146 },
147 ],
148 },
149 },
150 });
151 const pipeline = yield* AWS.OSIS.Pipeline("Logs", {
152 minUnits: 1,
153 maxUnits: 1,
154 pipelineConfigurationBody: pipelineConfig(
155 role.roleArn,
156 bucket.bucketName,
157 region,
158 ),
159 tags: { fixture: "osis-pipeline" },
160 });
161
162 // Resource-based policy: allow this account root to ingest.
163 const policy = yield* AWS.OSIS.ResourcePolicy("IngestPolicy", {
164 resourceArn: pipeline.pipelineArn,
165 policy: Output.interpolate`{
166 "Version": "2012-10-17",
167 "Statement": [
168 {
169 "Effect": "Allow",
170 "Principal": { "AWS": "arn:aws:iam::${accountId}:root" },
171 "Action": ["osis:Ingest"],
172 "Resource": "${pipeline.pipelineArn}"
173 }
174 ]
175 }`,
176 });
177
178 // VPC pipeline endpoint: private ingest into the pipeline.
179 const vpc = yield* AWS.EC2.Vpc("EndpointVpc", {
180 cidrBlock: "10.42.0.0/16",
181 });
182 const subnet = yield* AWS.EC2.Subnet("EndpointSubnet", {
183 vpcId: vpc.vpcId,
184 cidrBlock: "10.42.1.0/24",
185 availabilityZone: `${region}a`,
186 });
187 const securityGroup = yield* AWS.EC2.SecurityGroup("EndpointSG", {
188 vpcId: vpc.vpcId,
189 description: "osis pipeline endpoint fixture",
190 });
191 const endpoint = yield* AWS.OSIS.PipelineEndpoint("Private", {
192 pipelineArn: pipeline.pipelineArn,
193 vpcOptions: {
194 subnetIds: [subnet.subnetId],
195 securityGroupIds: [securityGroup.groupId],
196 },
197 });
198
199 return { pipeline, policy, endpoint };
200 }),
201 );
202
203 expect(pipeline.pipelineName).toBeDefined();
204 expect(pipeline.pipelineArn).toContain(":pipeline/");
205 expect(pipeline.status).toBe("ACTIVE");
206 expect(pipeline.minUnits).toBe(1);
207 expect(pipeline.maxUnits).toBe(1);
208 expect(pipeline.ingestEndpointUrls?.length).toBeGreaterThan(0);
209
210 // Out-of-band verification via distilled.
211 const described = yield* osis.getPipeline({
212 PipelineName: pipeline.pipelineName,
213 });
214 expect(described.Pipeline?.Status).toBe("ACTIVE");
215 expect(described.Pipeline?.MinUnits).toBe(1);
216 expect(described.Pipeline?.PipelineConfigurationBody).toContain(
217 "log-pipeline",
218 );
219
220 // Resource policy: attached and readable out-of-band.
221 expect(policy.resourceArn).toBe(pipeline.pipelineArn);
222 const readPolicy = yield* osis.getResourcePolicy({
223 ResourceArn: pipeline.pipelineArn,
224 });
225 expect(readPolicy.Policy).toContain("osis:Ingest");
226
227 // VPC pipeline endpoint: created and visible out-of-band.
228 expect(endpoint.endpointId).toMatch(/^pe-/);
229 expect(endpoint.pipelineArn).toBe(pipeline.pipelineArn);
230 expect(endpoint.status).toBe("ACTIVE");
231 const endpoints = yield* osis.listPipelineEndpoints({});
232 expect(
233 (endpoints.PipelineEndpoints ?? []).map((e) => e.EndpointId),
234 ).toContain(endpoint.endpointId);
235
236 // Destroy immediately — pipelines bill per OCU-hour — and verify
237 // deletion was initiated out-of-band.
238 yield* stack.destroy();
239 yield* assertPipelineDeleting(pipeline.pipelineName);
240 }),
241 // pipeline create (~5-10 min) + endpoint create (~5 min) + delete
242 // initiation, one test.
243 {
244 tags: [
245 "provider:aws",
246 "provider:aws:ec2",
247 "provider:aws:iam",
248 "provider:aws:osis",
249 "provider:aws:s3",
250 "live",
251 ],
252 timeout: 1_800_000,
253 },
254);