测试从衍生进程捕获的 IO

Question

我想通过以下方法测试 return 值和 IO 输出：

defmodule Speaker do
  def speak do
    receive do
      { :say, msg } ->
        IO.puts(msg)
        speak
      _other ->
        speak # throw away the message
    end
  end
end

在 ExUnit.CaptureIO docs 中，有一个示例测试执行此操作，如下所示：

test "checking the return value and the IO output" do
  fun = fn ->
    assert Enum.each(["some", "example"], &(IO.puts &1)) == :ok
  end
  assert capture_io(fun) == "some\nexample\n"
end

鉴于此，我想我可以编写以下测试来执行类似的操作，但使用 spawned 过程：

test ".speak with capture io" do
  pid = Kernel.spawn(Speaker, :speak, [])
  fun = fn ->
    assert send(pid, { :say, "Hello" }) == { :say, "Hello" }
  end
  assert capture_io(fun) == "Hello\n"
end

但是，我收到以下错误消息，告诉我没有输出，即使我可以在终端上看到输出：

1) test .speak with capture io (SpeakerTest)
   test/speaker_test.exs:25
   Assertion with == failed
   code: capture_io(fun) == "Hello\n"
   lhs:  ""
   rhs:  "Hello\n"
   stacktrace:
     test/speaker_test.exs:30: (test)

那么，关于测试使用 receive 宏的 spawned 进程或方法，我是否遗漏了什么？我怎样才能改变我的测试以使其通过？

Answer 1

CaptureIO 可能不适合您在此处尝试执行的操作。它运行一个函数，并且 returns 在该函数 returns 时捕获输出。但是你的功能从来没有 returns，所以这似乎行不通。我想出了以下解决方法：

test ".speak with capture io" do
  test_process = self()
  pid = spawn(fn ->
    Process.group_leader(self(), test_process)
    Speaker.speak
  end)

  send(pid, {:say, "Hello"})

  assert_receive {:io_request, _, _, {:put_chars, :unicode, "Hello\n"}}

  # Just to cleanup pid which dies upon not receiving a correct response
  # to the :io_request after a timeout
  Process.exit(pid, :kill)
end

它使用Process.group_leader将当前进程设置为测试进程的IO消息的接收者，然后断言这些消息到达。

Answer 2

我有一个类似的问题，我在我的 Application 上注册了一个进程，该进程每 10 秒超时一次并使用 IO.binwrite 写入 stdio，以模拟我在@Pawel 上遇到的多个超时 - Obrok 回答，但将其更改为使用 :io_reply 回复 :io_request，这样进程就不会挂起，允许我发送多条消息。

defp assert_io() do
  send(MyProcess, :timeout)
  receive do
    {:io_request, _, reply_as, {:put_chars, _, msg}} ->
      assert msg == "Some IO message"
      send(Stats, {:io_reply, reply_as, :ok})

    _ ->
      flunk
  end
end

test "get multiple messages" do
  Process.group_leader(Process.whereis(MyProcess), self())
  assert_io()
  assert_io()
end

如果您想了解更多关于 IO 协议的信息，请查看 the erlang docs about it.

测试从衍生进程捕获的 IO

Testing captured IO from a spawned process

io

elixir

spawn

ex-unit