services/retl-runner/src/console-client.test.ts

Rama newjitsu · réplica de lectura del repositorio original.

import { afterEach, describe, it, expect, vi } from "vitest";
import { ReverseRunConfig } from "@jitsu/warehouse-query/src/runtime";
import { createConsoleClient, type ConsoleRequest } from "./console-client";
import { createAdapterRegistry } from "./adapters";

const config = ReverseRunConfig.parse({
  version: 1,
  kind: "reverse",
  id: "sync",
  workspaceId: "workspace",
  fromId: "model",
  toId: "dst",
  configRevision: "a".repeat(64),
  updatedAt: new Date().toISOString(),
  model: { warehouseId: "warehouse", query: "select email from users", primaryKey: ["email"] },
  warehouse: { destinationType: "postgres" },
  destination: {
    destinationType: "google-ads",
    authorized: true,
    oauthConnectionId: "destination.dst",
    customerId: "1234567890",
  },
  options: {
    stream: "audience",
    mode: "upsert",
    mapping: { email: "email" },
    streamOptions: { audienceId: "123", customerMatchTermsAccepted: true },
  },
});
afterEach(() => {
  vi.useRealTimers();
  vi.unstubAllEnvs();
});
describe("runner console OAuth and adapter binding", () => {
  it("binds refresh admission and OAuth to the original task", async () => {
    const urls: URL[] = [];
    const client = createConsoleClient(
      "https://console.test.local",
      "service",
      config,
      async url => {
        urls.push(url);
        return new Response(
          JSON.stringify(
            url.pathname.includes("reverse-sync-oauth")
              ? { accessToken: "token", expiresAt: new Date(Date.now() + 300000).toISOString() }
              : config
          )
        );
      },
      "original-task"
    );
    await client.admit(new AbortController().signal);
    await client.accessToken(new AbortController().signal);
    expect(urls.map(url => url.searchParams.get("refreshTaskId"))).toEqual(["original-task", "original-task"]);
  });
  it("binds admission and OAuth to sync/workspace/revision, never arbitrary connection IDs", async () => {
    const request = vi.fn<ConsoleRequest>(
      async url =>
        new Response(
          JSON.stringify(
            String(url).includes("reverse-sync-oauth")
              ? { accessToken: "token", expiresAt: new Date(Date.now() + 300000).toISOString() }
              : config
          )
        )
    );
    const client = createConsoleClient("https://console.test.local", "service-token", config, request);
    const signal = new AbortController().signal;
    expect(await client.admit(signal)).toEqual(config);
    expect(await client.accessToken(signal)).toBe("token");
    expect(request.mock.calls.map(([url]) => String(url))).toEqual([
      "https://console.test.local/api/admin/reverse-syncs/sync?workspaceId=workspace",
      `https://console.test.local/api/admin/reverse-sync-oauth/sync?workspaceId=workspace&configRevision=${config.configRevision}`,
    ]);
    for (const [, init] of request.mock.calls) {
      expect(init?.redirect).toBe("error");
      expect(init?.headers).toEqual({ Authorization: "Bearer service-token" });
    }
  });
  it("caches briefly in memory, refreshes before expiry and does not cache errors", async () => {
    vi.useFakeTimers();
    const request = vi.fn<ConsoleRequest>(
      async () =>
        new Response(JSON.stringify({ accessToken: "token", expiresAt: new Date(Date.now() + 120000).toISOString() }))
    );
    const client = createConsoleClient("https://console.test.local", "service", config, request);
    const signal = new AbortController().signal;
    await client.accessToken(signal);
    await client.accessToken(signal);
    expect(request).toHaveBeenCalledTimes(1);
    vi.advanceTimersByTime(91000);
    request.mockImplementationOnce(async () => new Response("private", { status: 403 }));
    await expect(client.accessToken(signal)).rejects.toThrow("denied or unavailable");
    expect(await client.accessToken(signal)).toBe("token");
    expect(request).toHaveBeenCalledTimes(3);
    await expect(client.accessToken(AbortSignal.abort())).rejects.toThrow();
    expect(request).toHaveBeenCalledTimes(3);
  });
  it.each([{ accessToken: "token", expiresAt: new Date(0).toISOString() }, { refreshToken: "private" }])(
    "rejects expired/malformed token responses",
    async value => {
      const client = createConsoleClient(
        "https://console.test.local",
        "service",
        config,
        async () => new Response(JSON.stringify(value))
      );
      await expect(client.accessToken(new AbortController().signal)).rejects.toThrow();
    }
  );
  it("registers code-owned Microsoft, Google, Webhook, Meta, and OpenAI implementations, without fetching tokens for conversion binding", async () => {
    // Local runtime credentials must not hide an incomplete adapter fixture in CI.
    vi.stubEnv("MICROSOFT_ADS_DEVELOPER_TOKEN", undefined);
    const token = vi.fn(async () => "token");
    const registry = createAdapterRegistry(token);
    expect([...registry.keys()]).toEqual([
      "microsoft-ads",
      "google-ads",
      "webhook",
      "facebook-conversions",
      "openai-ads",
    ]);
    const adapter = await registry.get("google-ads")!(config);
    expect(adapter.targetIdentity).toBe("google-data-manager:1234567890:123");
    expect(adapter.stream.batchDelivery).toBe("asynchronous");
    expect(adapter.mirror).toBeUndefined();
    expect(adapter.recovery?.({}).reconcileBatch).toBeTypeOf("function");
    const meta = await registry.get("facebook-conversions")!({
      ...config,
      destination: { destinationType: "facebook-conversions", accessToken: "meta-test-token" },
      options: {
        ...config.options,
        stream: "conversions",
        mode: "upsert",
        mapping: { email: "email", eventTime: "event_time" },
        streamOptions: { pixelId: "456", eventName: "Purchase", actionSource: "website" },
      },
    });
    expect(meta.targetIdentity).toBe("meta-conversions:456");
    expect(meta.stream.name).toBe("conversions");
    const microsoft = await registry.get("microsoft-ads")!({
      ...config,
      destination: {
        destinationType: "microsoft-ads",
        developerToken: "microsoft-test-developer-token",
        accountId: "123",
        customerId: "456",
        oauthIntegrationId: "jitsu-cloud-dst-microsoft-ads",
        oauthConnectionId: `destination.${config.toId}`,
      },
      options: {
        ...config.options,
        stream: "offline-conversions",
        mode: "upsert",
        mapping: { email: "email", conversionTime: "time" },
        streamOptions: { conversionName: "Sale" },
      },
    });
    expect(microsoft.insertOnly).toBe(true);
    expect(microsoft.stream.name).toBe("offline-conversions");
    const openai = await registry.get("openai-ads")!({
      ...config,
      destination: { destinationType: "openai-ads", conversionsApiKey: "openai-test-key" },
      options: {
        ...config.options,
        stream: "conversions",
        mode: "upsert",
        mapping: { email: "email", sourceUrl: "url" },
        streamOptions: { pixelId: "px_1", eventType: "order_created", actionSource: "web" },
      },
    });
    expect(openai.insertOnly).toBe(true);
    expect(openai.targetIdentity).toBe("openai-conversions:px_1");
    expect(openai.stream.name).toBe("conversions");
    expect(token).not.toHaveBeenCalled();
    expect(() => registry.get("google-ads")!({ ...config, toId: "other" })).toThrow("OAuth binding");
  });
});