Add integration tests for dataset, ingest-token, datastream, and skill commands

integration/README.md

@@ -46,9 +46,9 @@ File names like `smoke.test.ts` are for human organization only.
 Mutating tests must follow the create → assert → delete lifecycle:
 
 1. Generate a unique prefix with `testPrefix()` (e.g. `cli-a1b2c3d4`).
-2. Create resources using that prefix in the name.
-3. Assert only on resources this test created (by name or ID in list/view output).
-4. Delete those resources in `finally`.
+2. Create resources using that prefix in the name (via CLI under test, or via `setup.ts` when seeding fixtures).
+3. **Register teardown immediately** after creation succeeds — before assertions — so resources are cleaned up even when a test fails mid-way. For resources created via CLI: `fixture.registerCleanup(() => deleteIngestToken(tenant, created.id))`. Setup helpers in `setup.ts` register their own cleanup.
+4. Assert only on resources this test created (by name or ID in list/view output).
 
 **Do not:**
 
@@ -74,12 +74,13 @@ Prefix pattern: `cli-<8 hex chars>`. A future sweeper can match `^cli-` to clean
 `parseJsonOutput` throws when the CLI exits non-zero, so a successful parse means the command succeeded.
 
 ```typescript
-// Good — assert on a resource this test created
+// Good — register cleanup right after create, before assertions
 const prefix = testPrefix();
 const result =
   await fixture.runCli`observe ingest-token create --name ${prefix}-token`;
-const tokens = parseJsonOutput(result) as Token[];
-expect(tokens.some((t) => t.name === `${prefix}-token`)).toBe(true);
+const created = parseJsonOutput(result) as Token;
+fixture.registerCleanup(() => deleteIngestToken(tenant, created.id));
+expect(created.name).toBe(`${prefix}-token`);
 
 // Good — validate response shape; datasets are guaranteed on any functional tenant
 expect(Array.isArray(datasets)).toBe(true);
@@ -135,3 +136,9 @@ await fixture.runCli`
 ## Parallelism
 
 Tests are designed to run in parallel against a shared tenant. Unique prefixes and the ownership rules above make that safe. `bun run test:integration` runs with `--concurrent --max-concurrency 5`. Use `test.serial` only when a test genuinely cannot run alongside others (rare).
+
+## Cleanup and setup helpers
+
+**Cleanup** (`integration/cleanup.ts`) — every resource created during a test must be cleaned up. Register teardown with `fixture.registerCleanup()` immediately after creation, before assertions. Cleanups run in LIFO order when the fixture is torn down; failures are logged but do not fail the test.
+
+**Setup** (`integration/setup.ts`) — when a test needs resources in the environment that aren't part of what it's testing, use setup helpers to create them via API instead of CLI. This includes resources the CLI can't create yet, but also resources the CLI _can_ create when the test simply doesn't care about exercising that path. Setup helpers register their own cleanup automatically.

integration/cleanup.ts

@@ -0,0 +1,42 @@
+/**
+ * API teardown helpers for integration tests — not part of the CLI surface under test.
+ *
+ * Register these with `fixture.registerCleanup()`; assertions belong on CLI output only.
+ */
+
+import { deleteDatastream as gqlDeleteDatastream } from "../src/gql/datastream/delete-datastream";
+import { deleteDataset as gqlDeleteDataset } from "../src/gql/dataset/delete-dataset";
+import { deleteIngestToken as gqlDeleteIngestToken } from "../src/gql/ingest-token/delete-ingest-token";
+import { deleteSkill as restDeleteSkill } from "../src/rest/skill/delete-skill";
+import type { Config } from "../src/lib/config";
+
+export async function deleteIngestToken(
+  config: Config,
+  id: string,
+): Promise<void> {
+  const success = await gqlDeleteIngestToken(config, { id });
+  if (!success) {
+    throw new Error(`deleteIngestToken returned false for ingest token ${id}`);
+  }
+}
+
+export async function deleteDatastream(
+  config: Config,
+  id: string,
+): Promise<void> {
+  const success = await gqlDeleteDatastream(config, { id });
+  if (!success) {
+    throw new Error(`deleteDatastream returned false for datastream ${id}`);
+  }
+}
+
+export async function deleteDataset(config: Config, id: string): Promise<void> {
+  const success = await gqlDeleteDataset(config, { dsid: id });
+  if (!success) {
+    throw new Error(`deleteDataset returned false for dataset ${id}`);
+  }
+}
+
+export async function deleteSkill(config: Config, id: string): Promise<void> {
+  await restDeleteSkill({ config, id });
+}

integration/dataset.test.ts

@@ -0,0 +1,85 @@
+import { describe, expect, test } from "bun:test";
+import {
+  loadTenantConfig,
+  parseJsonOutput,
+  retryUntil,
+  testPrefix,
+  withIntegrationFixture,
+} from "./fixture";
+import { createTestDataset } from "./setup";
+
+interface DatasetListEntry {
+  id: string;
+  label: string;
+}
+
+interface DatasetViewJson {
+  id: string;
+  label: string;
+  fieldList?: { name: string }[];
+}
+
+interface QueryRow {
+  test?: string | number;
+}
+
+const tenant = loadTenantConfig();
+
+describe("dataset CLI integration", () => {
+  test("list and view an API-created dataset", async () => {
+    const label = `${testPrefix()}-dataset`;
+    const opal = `
+filter true
+make_col test:5
+`.trim();
+
+    await withIntegrationFixture(tenant, async (fixture) => {
+      // Seed a dataset via API (not under test).
+      const created = await createTestDataset(fixture, label, opal);
+
+      // dataset list: exact filter finds the fixture by label.
+      const listFilter = `label == ${JSON.stringify(label)}`;
+      const listResult = await fixture.runCli`
+        observe dataset list \
+          --format json \
+          --filter ${JSON.stringify(listFilter)}
+      `;
+      const listed = parseJsonOutput(listResult) as DatasetListEntry[];
+
+      expect(Array.isArray(listed)).toBe(true);
+      expect(listed).toHaveLength(1);
+      expect(listed[0]?.id).toBe(created.id);
+      expect(listed[0]?.label).toBe(label);
+
+      // dataset view: metadata reflects the saved OPAL pipeline.
+      const viewResult = await fixture.runCli`
+        observe dataset view ${created.id} \
+          --format json
+      `;
+      const viewed = parseJsonOutput(viewResult) as DatasetViewJson;
+
+      expect(viewed.id).toBe(created.id);
+      expect(viewed.label).toBe(label);
+      expect(viewed.fieldList?.some((field) => field.name === "test")).toBe(
+        true,
+      );
+
+      // query: dataset is queryable once materialized; OPAL output includes test=5.
+      const rows = await retryUntil(
+        async () => {
+          const queryResult = await fixture.runCli`
+            observe query \
+              --input ${created.id} \
+              --pipeline "limit 1" \
+              --format json \
+              --interval 30d
+          `;
+          return parseJsonOutput(queryResult) as QueryRow[];
+        },
+        (result) => result.length > 0,
+      );
+
+      expect(rows[0]?.test).toBe("5");
+    });
+  });
+});

integration/datastream.test.ts

@@ -0,0 +1,85 @@
+import { describe, expect } from "bun:test";
+import { deleteDatastream } from "./cleanup";
+import {
+  loadTenantConfig,
+  parseJsonOutput,
+  testCiOnly,
+  testPrefix,
+  withIntegrationFixture,
+} from "./fixture";
+
+interface DatastreamJson {
+  id: string;
+  name: string;
+  description?: string | null;
+  disabled?: boolean;
+}
+
+const tenant = loadTenantConfig();
+
+describe("datastream CLI integration", () => {
+  // Some tenants allow ingest-token writes but reject datastream create (read-only mode).
+  testCiOnly("create, list, view, and update", async () => {
+    const prefix = testPrefix();
+    const name = `${prefix}-datastream`;
+    const description = "integration test datastream";
+
+    await withIntegrationFixture(tenant, async (fixture) => {
+      // datastream create
+      const createResult = await fixture.runCli`
+        observe datastream create \
+          --name ${name} \
+          --description ${JSON.stringify(description)}
+      `;
+      const created = parseJsonOutput(createResult) as DatastreamJson;
+      fixture.registerCleanup(() => deleteDatastream(tenant, created.id));
+
+      expect(typeof created.id).toBe("string");
+      expect(created.id.length).toBeGreaterThan(0);
+      expect(created.name).toBe(name);
+
+      // datastream list
+      const listResult = await fixture.runCli`
+        observe datastream list \
+          --match ${prefix}
+      `;
+      const listed = parseJsonOutput(listResult) as DatastreamJson[];
+
+      expect(Array.isArray(listed)).toBe(true);
+      expect(listed.some((ds) => ds.id === created.id)).toBe(true);
+      expect(listed.some((ds) => ds.name === name)).toBe(true);
+
+      // datastream view
+      const viewResult = await fixture.runCli`
+        observe datastream view ${created.id}
+      `;
+      const viewed = parseJsonOutput(viewResult) as DatastreamJson;
+
+      expect(viewed.id).toBe(created.id);
+      expect(viewed.name).toBe(name);
+      expect(viewed.description).toBe(description);
+
+      // datastream update
+      const updatedDescription = `${description} (updated)`;
+      const updateResult = await fixture.runCli`
+        observe datastream update ${created.id} \
+          --description ${JSON.stringify(updatedDescription)}
+      `;
+      const updated = parseJsonOutput(updateResult) as DatastreamJson;
+
+      expect(updated.id).toBe(created.id);
+      expect(updated.description).toBe(updatedDescription);
+
+      // datastream view: update persisted
+      const viewAfterUpdateResult = await fixture.runCli`
+        observe datastream view ${created.id}
+      `;
+      const viewedAfterUpdate = parseJsonOutput(
+        viewAfterUpdateResult,
+      ) as DatastreamJson;
+
+      expect(viewedAfterUpdate.id).toBe(created.id);
+      expect(viewedAfterUpdate.description).toBe(updatedDescription);
+    });
+  });
+});

integration/fixture.test.ts

@@ -19,7 +19,7 @@ describe("runCli command validation", () => {
         expect(error).toBeInstanceOf(InvalidCliCommandError);
       }
     } finally {
-      fixture.cleanup();
+      await fixture.cleanup();
     }
   });
 
@@ -33,7 +33,7 @@ describe("runCli command validation", () => {
         expect(error).toBeInstanceOf(InvalidCliCommandError);
       }
     } finally {
-      fixture.cleanup();
+      await fixture.cleanup();
     }
   });
 
@@ -47,7 +47,7 @@ describe("runCli command validation", () => {
         expect(error).toBeInstanceOf(InvalidCliCommandError);
       }
     } finally {
-      fixture.cleanup();
+      await fixture.cleanup();
     }
   });
 

integration/fixture.ts

@@ -61,14 +61,17 @@ export async function withIntegrationFixture(
   try {
     await fn(fixture);
   } finally {
-    fixture.cleanup();
+    await fixture.cleanup();
   }
 }
 
+export type CleanupFn = () => void | Promise<void>;
+
 export class IntegrationFixture {
   readonly tenant: Config;
   readonly tempHome: string;
   readonly env: NodeJS.ProcessEnv;
+  private readonly cleanups: CleanupFn[] = [];
 
   constructor(tenant: Config) {
     this.tenant = tenant;
@@ -106,11 +109,52 @@ export class IntegrationFixture {
     };
   };
 
-  cleanup(): void {
+  /** Register teardown to run when the fixture is cleaned up (LIFO order). */
+  registerCleanup(fn: CleanupFn): void {
+    this.cleanups.push(fn);
+  }
+
+  async cleanup(): Promise<void> {
+    for (const fn of [...this.cleanups].reverse()) {
+      try {
+        await fn();
+      } catch (error) {
+        const message = error instanceof Error ? error.message : String(error);
+        console.error(`integration cleanup failed: ${message}`);
+      }
+    }
+    this.cleanups.length = 0;
     rmSync(this.tempHome, { recursive: true, force: true });
   }
 }
 
+/**
+ * Repeatedly run `fn` until `isReady` returns true, or throw after timeout.
+ * Catch and re-throw at the call site to attach test-specific context.
+ */
+export async function retryUntil<T>(
+  fn: () => Promise<T>,
+  isReady: (value: T) => boolean,
+  options: {
+    timeoutMs?: number;
+    intervalMs?: number;
+  } = {},
+): Promise<T> {
+  const timeoutMs = options.timeoutMs ?? 10_000;
+  const intervalMs = options.intervalMs ?? 500;
+  const deadline = Date.now() + timeoutMs;
+
+  while (Date.now() < deadline) {
+    const value = await fn();
+    if (isReady(value)) {
+      return value;
+    }
+    await Bun.sleep(intervalMs);
+  }
+
+  throw new Error(`retryUntil timed out after ${String(timeoutMs)}ms`);
+}
+
 export function parseJsonOutput(result: CliResult): unknown {
   if (result.exitCode !== 0) {
     throw new Error(