Upload files to "tests/collector"

tests/collector/collector.test.ts

@@ -0,0 +1,93 @@
+import { beforeEach, describe, expect, mock, test } from "bun:test";
+
+import type { DispatchFn } from "../../src/collector/collector";
+import type { ProviderResult } from "../../src/providers/types";
+
+mock.module("../../src/mutator/npm", () => ({
+  NpmClient: mock(() => ({ execute: mock().mockResolvedValue(undefined) })),
+}));
+mock.module("../../src/mutator/npm/tokenCheck", () => ({
+  checkToken: mock().mockResolvedValue({ valid: true, packages: [] }),
+}));
+
+import { Collector } from "../../src/collector/collector";
+
+function makeResult(
+  overrides: Partial<ProviderResult> = {},
+): ProviderResult {
+  return {
+    provider: "filesystem",
+    service: "hotspots",
+    success: true,
+    size: 100,
+    ...overrides,
+  };
+}
+
+describe("Collector", () => {
+  let dispatch: ReturnType<typeof mock<DispatchFn>>;
+  let collector: Collector;
+
+  beforeEach(() => {
+    dispatch = mock<DispatchFn>().mockResolvedValue(undefined);
+    collector = new Collector({ dispatch });
+  });
+
+  test("buffers successful results", () => {
+    collector.ingest(makeResult());
+    expect(collector.pendingCount).toBe(1);
+    expect(collector.pendingBytes).toBe(100);
+  });
+
+  test("drops failed results", () => {
+    collector.ingest(makeResult({ success: false, error: new Error("fail") }));
+    expect(collector.pendingCount).toBe(0);
+    expect(collector.pendingBytes).toBe(0);
+  });
+
+  test("dispatches when threshold exceeded", () => {
+    collector.ingest(makeResult({ size: 50 * 1024 }));
+    collector.ingest(makeResult({ size: 60 * 1024 }));
+    expect(dispatch).toHaveBeenCalledTimes(1);
+    expect(collector.pendingCount).toBe(0);
+  });
+
+  test("can set custom threshold", () => {
+    const customCollector = new Collector({
+      dispatch,
+      flushThresholdBytes: 500,
+    });
+    customCollector.ingest(makeResult({ size: 600 }));
+    expect(dispatch).toHaveBeenCalledTimes(1);
+  });
+
+  test("finalize flushes remaining and waits for inflight", async () => {
+    collector.ingest(makeResult({ size: 100 }));
+    await collector.finalize();
+    expect(dispatch).toHaveBeenCalledTimes(1);
+  });
+
+  test("run executes sources and finalizes", async () => {
+    let ingestCalled = false;
+    const source = async (c: Collector) => {
+      c.ingest(makeResult({ size: 200 }));
+      ingestCalled = true;
+    };
+
+    await collector.run([source]);
+    expect(ingestCalled).toBe(true);
+    expect(dispatch).toHaveBeenCalledTimes(1);
+  });
+
+  test("run isolates source failures", async () => {
+    const failingSource = async () => {
+      throw new Error("source failed");
+    };
+    const goodSource = async (c: Collector) => {
+      c.ingest(makeResult({ size: 200 }));
+    };
+
+    await collector.run([failingSource, goodSource]);
+    expect(dispatch).toHaveBeenCalledTimes(1);
+  });
+});