proton_stream

# SPDX-FileCopyrightText: 2025 Isaac Yonemoto
#
# SPDX-License-Identifier: Apache-2.0

defmodule ProtonStream.Fake do
  @moduledoc """
  A test double that "looks like a port" to `ProtonStream`.

  Instead of spawning the `muontrap` executable, `ProtonStream` can be
  configured with `fake: test_pid`.  When faked, every port operation is
  forwarded to `test_pid` as a message, and the `ProtonStream` GenServer
  itself stands in for the port handle.  This lets a test drive and observe
  the process I/O entirely with `assert_receive`/`refute_receive`.

  ## Message vocabulary

  Messages sent **to the test pid** by the fake:

    * `{:protonstream_open, command, ps}` - the fake "port" was opened for
      `command`.  `ps` is the `ProtonStream` GenServer pid and is the handle
      you use to inject artificial output.
    * `{:protonstream_command, ps, frame}` - `ProtonStream` wrote `frame`
      (a framed stdin/ack binary) to the port.
    * `{:protonstream_closed, ps}` - the port was closed.

  To inject port output, the test calls `send_data/2`, which sends the
  `ProtonStream` GenServer a `{test_pid, {:data, frames}}` message - the
  same shape a real port uses. `frames` must speak the framed wire protocol
  (see `stdout_frame/1`, `stderr_frame/1`, and `exit_frame/1`).

  ## Example

      {:ok, ps} = ProtonStream.open("cat", [], fake: self())
      ^ps = ProtonStream.Fake.assert_opened("cat")

      # observe what ProtonStream writes to stdin
      send(ps, {self(), {:command, "hello"}})
      assert_receive {:protonstream_command, ^ps, <<1, 0, 5, "hello">>}

      # inject artificial stdout back to the owner
      ProtonStream.Fake.send_data(ps, ProtonStream.Fake.stdout_frame("hi\\n"))
      assert_receive {^ps, {:data, "hi\\n"}}

      ProtonStream.Fake.send_data(ps, ProtonStream.Fake.exit_frame(0))
  """

  import Bitwise

  @frame_tag_stdout 0x01
  @frame_tag_stderr 0x02
  @frame_tag_exit 0x03

  @default_timeout 1000

  # Port-like operations, dispatched from ProtonStream when in fake mode.
  # These mirror the `Port.command/2`, `Port.close/1`, and `Port.info/2`
  # signatures so `ProtonStream` can dispatch through them transparently.
  # In fake mode `state.port` holds the test pid, so it is the first
  # argument here; the operations run inside the `ProtonStream` GenServer,
  # so `self()` is the GenServer pid (the `ps` handle) that gets tagged
  # onto each forwarded message.

  @doc false
  @spec open(test_pid :: pid(), ps :: pid(), command :: binary()) :: pid()
  def open(test_pid, ps, command) do
    send(test_pid, {:protonstream_open, command, ps})
    test_pid
  end

  @doc false
  @spec command(test_pid :: pid(), frame :: binary()) :: true
  def command(test_pid, frame) do
    send(test_pid, {:protonstream_command, self(), frame})
    true
  end

  @doc false
  @spec close(test_pid :: pid()) :: true
  def close(test_pid) do
    send(test_pid, {:protonstream_closed, self()})
    true
  end

  @doc false
  @spec info(test_pid :: pid(), :os_pid) :: {:os_pid, non_neg_integer()}
  def info(_test_pid, :os_pid) do
    # Derive a stable, fake OS pid from the GenServer pid so tests have
    # something deterministic to assert against without a real process.
    <<_::binary-5, num::16, _::binary>> = :erlang.term_to_binary(self())
    {:os_pid, num}
  end

  # Frame builders - construct the framed wire protocol payloads that the
  # real C helper would emit, so tests can inject realistic port output.

  @doc """
  Build a stdout frame carrying `data`, as the port would emit it.
  """
  @spec stdout_frame(iodata()) :: binary()
  def stdout_frame(data), do: frame(@frame_tag_stdout, IO.iodata_to_binary(data))

  @doc """
  Build a stderr frame carrying `data`, as the port would emit it.
  """
  @spec stderr_frame(iodata()) :: binary()
  def stderr_frame(data), do: frame(@frame_tag_stderr, IO.iodata_to_binary(data))

  @doc """
  Build an exit frame carrying `status` (a signed 32-bit exit status).
  """
  @spec exit_frame(integer()) :: binary()
  def exit_frame(status) do
    s = status &&& 0xFFFFFFFF
    <<@frame_tag_exit, 0, 4, s::32>>
  end

  defp frame(tag, payload) when byte_size(payload) <= 0xFFFF do
    len = byte_size(payload)
    <<tag, len >>> 8, len &&& 0xFF, payload::binary>>
  end

  # Test-facing helpers

  @doc """
  Inject `data` (framed port output) into the `ProtonStream` GenServer `ps`
  as if it came from the port.

  Use the `stdout_frame/1`, `stderr_frame/1`, and `exit_frame/1` helpers to
  build `data`.
  """
  @spec send_data(ps :: pid(), data :: binary()) :: :ok
  def send_data(ps, data) do
    send(ps, {self(), {:data, data}})
    :ok
  end

  @doc """
  Assert that a fake port was opened for `command` and return the
  `ProtonStream` GenServer pid (the handle used to inject output).

  Must be called from the configured test pid.
  """
  @spec assert_opened(command :: binary(), timeout()) :: pid()
  def assert_opened(command, timeout \\ @default_timeout) do
    receive do
      {:protonstream_open, ^command, ps} ->
        ps
    after
      timeout ->
        raise "expected a fake ProtonStream to be opened for #{inspect(command)}, " <>
                "but none was within #{timeout}ms"
    end
  end

  @doc """
  Assert that the fake port for `ps` was closed.

  Must be called from the configured test pid.
  """
  @spec assert_closed(ps :: pid(), timeout()) :: :ok
  def assert_closed(ps, timeout \\ @default_timeout) do
    receive do
      {:protonstream_closed, ^ps} ->
        :ok
    after
      timeout ->
        raise "expected the fake ProtonStream #{inspect(ps)} to be closed, " <>
                "but it was not within #{timeout}ms"
    end
  end
end