moltbot/extensions/test-utils/plugin-runtime-mock.ts

import type { PluginRuntime } from "openclaw/plugin-sdk";
import { removeAckReactionAfterReply, shouldAckReaction } from "openclaw/plugin-sdk";
import { vi } from "vitest";

type DeepPartial<T> = {
  [K in keyof T]?: T[K] extends (...args: never[]) => unknown
    ? T[K]
    : T[K] extends ReadonlyArray<unknown>
      ? T[K]
      : T[K] extends object
        ? DeepPartial<T[K]>
        : T[K];
};

function isObject(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function mergeDeep<T>(base: T, overrides: DeepPartial<T>): T {
  const result: Record<string, unknown> = { ...(base as Record<string, unknown>) };
  for (const [key, overrideValue] of Object.entries(overrides as Record<string, unknown>)) {
    if (overrideValue === undefined) {
      continue;
    }
    const baseValue = result[key];
    if (isObject(baseValue) && isObject(overrideValue)) {
      result[key] = mergeDeep(baseValue, overrideValue);
      continue;
    }
    result[key] = overrideValue;
  }
  return result as T;
}

export function createPluginRuntimeMock(overrides: DeepPartial<PluginRuntime> = {}): PluginRuntime {
  const base: PluginRuntime = {
    version: "1.0.0-test",
    config: {
      loadConfig: vi.fn(() => ({})) as unknown as PluginRuntime["config"]["loadConfig"],
      writeConfigFile: vi.fn() as unknown as PluginRuntime["config"]["writeConfigFile"],
    },
    system: {
      enqueueSystemEvent: vi.fn() as unknown as PluginRuntime["system"]["enqueueSystemEvent"],
      requestHeartbeatNow: vi.fn() as unknown as PluginRuntime["system"]["requestHeartbeatNow"],
      runCommandWithTimeout: vi.fn() as unknown as PluginRuntime["system"]["runCommandWithTimeout"],
      formatNativeDependencyHint: vi.fn(
        () => "",
      ) as unknown as PluginRuntime["system"]["formatNativeDependencyHint"],
    },
    media: {
      loadWebMedia: vi.fn() as unknown as PluginRuntime["media"]["loadWebMedia"],
      detectMime: vi.fn() as unknown as PluginRuntime["media"]["detectMime"],
      mediaKindFromMime: vi.fn() as unknown as PluginRuntime["media"]["mediaKindFromMime"],
      isVoiceCompatibleAudio:
        vi.fn() as unknown as PluginRuntime["media"]["isVoiceCompatibleAudio"],
      getImageMetadata: vi.fn() as unknown as PluginRuntime["media"]["getImageMetadata"],
      resizeToJpeg: vi.fn() as unknown as PluginRuntime["media"]["resizeToJpeg"],
    },
    tts: {
      textToSpeechTelephony: vi.fn() as unknown as PluginRuntime["tts"]["textToSpeechTelephony"],
    },
    stt: {
      transcribeAudioFile: vi.fn() as unknown as PluginRuntime["stt"]["transcribeAudioFile"],
    },
    tools: {
      createMemoryGetTool: vi.fn() as unknown as PluginRuntime["tools"]["createMemoryGetTool"],
      createMemorySearchTool:
        vi.fn() as unknown as PluginRuntime["tools"]["createMemorySearchTool"],
      registerMemoryCli: vi.fn() as unknown as PluginRuntime["tools"]["registerMemoryCli"],
    },
    channel: {
      text: {
        chunkByNewline: vi.fn((text: string) => (text ? [text] : [])),
        chunkMarkdownText: vi.fn((text: string) => [text]),
        chunkMarkdownTextWithMode: vi.fn((text: string) => (text ? [text] : [])),
        chunkText: vi.fn((text: string) => (text ? [text] : [])),
        chunkTextWithMode: vi.fn((text: string) => (text ? [text] : [])),
        resolveChunkMode: vi.fn(
          () => "length",
        ) as unknown as PluginRuntime["channel"]["text"]["resolveChunkMode"],
        resolveTextChunkLimit: vi.fn(() => 4000),
        hasControlCommand: vi.fn(() => false),
        resolveMarkdownTableMode: vi.fn(
          () => "code",
        ) as unknown as PluginRuntime["channel"]["text"]["resolveMarkdownTableMode"],
        convertMarkdownTables: vi.fn((text: string) => text),
      },
      reply: {
        dispatchReplyWithBufferedBlockDispatcher: vi.fn(
          async () => undefined,
        ) as unknown as PluginRuntime["channel"]["reply"]["dispatchReplyWithBufferedBlockDispatcher"],
        createReplyDispatcherWithTyping:
          vi.fn() as unknown as PluginRuntime["channel"]["reply"]["createReplyDispatcherWithTyping"],
        resolveEffectiveMessagesConfig:
          vi.fn() as unknown as PluginRuntime["channel"]["reply"]["resolveEffectiveMessagesConfig"],
        resolveHumanDelayConfig:
          vi.fn() as unknown as PluginRuntime["channel"]["reply"]["resolveHumanDelayConfig"],
        dispatchReplyFromConfig:
          vi.fn() as unknown as PluginRuntime["channel"]["reply"]["dispatchReplyFromConfig"],
        withReplyDispatcher: vi.fn(async ({ dispatcher, run, onSettled }) => {
          try {
            return await run();
          } finally {
            dispatcher.markComplete();
            try {
              await dispatcher.waitForIdle();
            } finally {
              await onSettled?.();
            }
          }
        }) as unknown as PluginRuntime["channel"]["reply"]["withReplyDispatcher"],
        finalizeInboundContext: vi.fn(
          (ctx: Record<string, unknown>) => ctx,
        ) as unknown as PluginRuntime["channel"]["reply"]["finalizeInboundContext"],
        formatAgentEnvelope: vi.fn(
          (opts: { body: string }) => opts.body,
        ) as unknown as PluginRuntime["channel"]["reply"]["formatAgentEnvelope"],
        formatInboundEnvelope: vi.fn(
          (opts: { body: string }) => opts.body,
        ) as unknown as PluginRuntime["channel"]["reply"]["formatInboundEnvelope"],
        resolveEnvelopeFormatOptions: vi.fn(() => ({
          template: "channel+name+time",
        })) as unknown as PluginRuntime["channel"]["reply"]["resolveEnvelopeFormatOptions"],
      },
      routing: {
        resolveAgentRoute: vi.fn(() => ({
          agentId: "main",
          accountId: "default",
          sessionKey: "agent:main:test:dm:peer",
        })) as unknown as PluginRuntime["channel"]["routing"]["resolveAgentRoute"],
      },
      pairing: {
        buildPairingReply: vi.fn(
          () => "Pairing code: TESTCODE",
        ) as unknown as PluginRuntime["channel"]["pairing"]["buildPairingReply"],
        readAllowFromStore: vi
          .fn()
          .mockResolvedValue(
            [],
          ) as unknown as PluginRuntime["channel"]["pairing"]["readAllowFromStore"],
        upsertPairingRequest: vi.fn().mockResolvedValue({
          code: "TESTCODE",
          created: true,
        }) as unknown as PluginRuntime["channel"]["pairing"]["upsertPairingRequest"],
      },
      media: {
        fetchRemoteMedia:
          vi.fn() as unknown as PluginRuntime["channel"]["media"]["fetchRemoteMedia"],
        saveMediaBuffer: vi.fn().mockResolvedValue({
          path: "/tmp/test-media.jpg",
          contentType: "image/jpeg",
        }) as unknown as PluginRuntime["channel"]["media"]["saveMediaBuffer"],
      },
      session: {
        resolveStorePath: vi.fn(
          () => "/tmp/sessions.json",
        ) as unknown as PluginRuntime["channel"]["session"]["resolveStorePath"],
        readSessionUpdatedAt: vi.fn(
          () => undefined,
        ) as unknown as PluginRuntime["channel"]["session"]["readSessionUpdatedAt"],
        recordSessionMetaFromInbound:
          vi.fn() as unknown as PluginRuntime["channel"]["session"]["recordSessionMetaFromInbound"],
        recordInboundSession:
          vi.fn() as unknown as PluginRuntime["channel"]["session"]["recordInboundSession"],
        updateLastRoute:
          vi.fn() as unknown as PluginRuntime["channel"]["session"]["updateLastRoute"],
      },
      mentions: {
        buildMentionRegexes: vi.fn(() => [
          /\bbert\b/i,
        ]) as unknown as PluginRuntime["channel"]["mentions"]["buildMentionRegexes"],
        matchesMentionPatterns: vi.fn((text: string, regexes: RegExp[]) =>
          regexes.some((regex) => regex.test(text)),
        ) as unknown as PluginRuntime["channel"]["mentions"]["matchesMentionPatterns"],
        matchesMentionWithExplicit: vi.fn(
          (params: { text: string; mentionRegexes: RegExp[]; explicitWasMentioned?: boolean }) =>
            params.explicitWasMentioned === true
              ? true
              : params.mentionRegexes.some((regex) => regex.test(params.text)),
        ) as unknown as PluginRuntime["channel"]["mentions"]["matchesMentionWithExplicit"],
      },
      reactions: {
        shouldAckReaction,
        removeAckReactionAfterReply,
      },
      groups: {
        resolveGroupPolicy: vi.fn(
          () => "open",
        ) as unknown as PluginRuntime["channel"]["groups"]["resolveGroupPolicy"],
        resolveRequireMention: vi.fn(
          () => false,
        ) as unknown as PluginRuntime["channel"]["groups"]["resolveRequireMention"],
      },
      debounce: {
        createInboundDebouncer: vi.fn(
          (params: { onFlush: (items: unknown[]) => Promise<void> }) => ({
            enqueue: async (item: unknown) => {
              await params.onFlush([item]);
            },
            flushKey: vi.fn(),
          }),
        ) as unknown as PluginRuntime["channel"]["debounce"]["createInboundDebouncer"],
        resolveInboundDebounceMs: vi.fn(
          () => 0,
        ) as unknown as PluginRuntime["channel"]["debounce"]["resolveInboundDebounceMs"],
      },
      commands: {
        resolveCommandAuthorizedFromAuthorizers: vi.fn(
          () => false,
        ) as unknown as PluginRuntime["channel"]["commands"]["resolveCommandAuthorizedFromAuthorizers"],
        isControlCommandMessage:
          vi.fn() as unknown as PluginRuntime["channel"]["commands"]["isControlCommandMessage"],
        shouldComputeCommandAuthorized:
          vi.fn() as unknown as PluginRuntime["channel"]["commands"]["shouldComputeCommandAuthorized"],
        shouldHandleTextCommands:
          vi.fn() as unknown as PluginRuntime["channel"]["commands"]["shouldHandleTextCommands"],
      },
      discord: {} as PluginRuntime["channel"]["discord"],
      activity: {} as PluginRuntime["channel"]["activity"],
      line: {} as PluginRuntime["channel"]["line"],
      slack: {} as PluginRuntime["channel"]["slack"],
      telegram: {} as PluginRuntime["channel"]["telegram"],
      signal: {} as PluginRuntime["channel"]["signal"],
      imessage: {} as PluginRuntime["channel"]["imessage"],
      whatsapp: {} as PluginRuntime["channel"]["whatsapp"],
    },
    events: {
      onAgentEvent: vi.fn(() => () => {}) as unknown as PluginRuntime["events"]["onAgentEvent"],
      onSessionTranscriptUpdate: vi.fn(
        () => () => {},
      ) as unknown as PluginRuntime["events"]["onSessionTranscriptUpdate"],
    },
    logging: {
      shouldLogVerbose: vi.fn(() => false),
      getChildLogger: vi.fn(() => ({
        info: vi.fn(),
        warn: vi.fn(),
        error: vi.fn(),
        debug: vi.fn(),
      })),
    },
    state: {
      resolveStateDir: vi.fn(() => "/tmp/openclaw"),
    },
  };

  return mergeDeep(base, overrides);
}