Write function to get child pid in tests

CHANGELOG.md

@@ -5,6 +5,7 @@
   * Add children groups - a mechanism that allows refering to multiple children with a single identifier. 
   * Rename `remove_child_t` action into `remove_children_t` and allow for removing a children group with a single action.
   * Add an ability to spawn anonymous children.
+  * Add `Membrane.Testing.Pipeline.get_child_pid/2`
 
 ## 0.11.0
  * Separate element_name and pad arguments in handle_element_{start, end}_of_stream signature [#219](https://github.com/membraneframework/membrane_core/issues/219)

lib/membrane/core/bin.ex

@@ -257,4 +257,16 @@ defmodule Membrane.Core.Bin do
   def handle_call(Message.new(:get_clock), _from, state) do
     {:reply, state.synchronization.clock, state}
   end
+
+  @impl GenServer
+  def handle_call(Message.new(:get_child_pid, child_ref), _from, state) do
+    reply =
+      with %State{children: %{^child_ref => %{pid: child_pid}}} <- state do
+        {:ok, child_pid}
+      else
+        _other -> {:error, :child_not_found}
+      end
+
+    {:reply, reply, state}
+  end
 end

lib/membrane/core/element.ex

@@ -148,6 +148,11 @@ defmodule Membrane.Core.Element do
     {:reply, :ok, state}
   end
 
+  @impl GenServer
+  def handle_call(Message.new(:get_child_pid, _child_ref), _from, state) do
+    {:reply, {:error, :element_cannot_have_children}, state}
+  end
+
   @impl GenServer
   def handle_call(message, {pid, _tag}, _state) do
     raise Membrane.ElementError,

lib/membrane/core/pipeline.ex

@@ -133,6 +133,18 @@ defmodule Membrane.Core.Pipeline do
     {:noreply, state}
   end
 
+  @impl GenServer
+  def handle_call(Message.new(:get_child_pid, child_ref), _from, state) do
+    reply =
+      with %State{children: %{^child_ref => %{pid: child_pid}}} <- state do
+        {:ok, child_pid}
+      else
+        _other -> {:error, :child_not_found}
+      end
+
+    {:reply, reply, state}
+  end
+
   @impl GenServer
   def handle_call(message, from, state) do
     context = &CallbackContext.Call.from_state(&1, from: from)

lib/membrane/testing/pipeline.ex

@@ -79,11 +79,14 @@ defmodule Membrane.Testing.Pipeline do
 
   use Membrane.Pipeline
 
+  alias Membrane.Child
   alias Membrane.ChildrenSpec
+  alias Membrane.Core.Message
   alias Membrane.{Element, Pipeline}
   alias Membrane.Testing.Notification
 
   require Membrane.Logger
+  require Membrane.Core.Message
 
   defmodule State do
     @moduledoc false
@@ -218,6 +221,65 @@ defmodule Membrane.Testing.Pipeline do
     :ok
   end
 
+  @doc """
+  Returns the pid of the children process.
+
+  Accepts pipeline pid as a first argument and a child reference or a list
+  of child references representing a path as a second argument.
+
+  If second argument is a child reference, function gets pid of this child
+  from pipeline.
+
+  If second argument is a path of child references, function gets pid of
+  last a component pointed by this path.
+
+  Returns
+   * `{:ok, child_pid}`, if a child was succesfully found
+   * `{:error, reason}`, if, for example, pipeline is not alive or children path is invalid
+  """
+  @spec get_child_pid(pid(), child_ref_path :: Child.ref_t() | [Child.ref_t()]) ::
+          {:ok, pid()} | {:error, reason :: term()}
+  def get_child_pid(pipeline, [_head | _tail] = child_ref_path) do
+    do_get_child_pid(pipeline, child_ref_path)
+  end
+
+  def get_child_pid(pipeline, child_ref) when not is_list(child_ref) do
+    do_get_child_pid(pipeline, [child_ref])
+  end
+
+  @doc """
+  Returns the pid of the children process.
+
+  Works as get_child_pid/2, but raises an error instead of returning
+  `{:error, reason}` tuple.
+  """
+  @spec get_child_pid!(pid(), child_ref_path :: Child.ref_t() | [Child.ref_t()]) :: pid()
+  def get_child_pid!(parent_pid, child_ref_path) do
+    {:ok, child_pid} = get_child_pid(parent_pid, child_ref_path)
+    child_pid
+  end
+
+  defp do_get_child_pid(component_pid, child_ref_path, is_pipeline? \\ true)
+
+  defp do_get_child_pid(component_pid, [], _is_pipeline?) do
+    {:ok, component_pid}
+  end
+
+  defp do_get_child_pid(component_pid, [child_ref | child_ref_path_tail], is_pipeline?) do
+    case Message.call(component_pid, :get_child_pid, child_ref) do
+      {:ok, child_pid} ->
+        do_get_child_pid(child_pid, child_ref_path_tail, false)
+
+      {:error, {:call_failure, {:noproc, _call_info}}} ->
+        if is_pipeline?,
+          do: {:error, :pipeline_not_alive},
+          else: {:error, :component_not_alive}
+
+      {:error, _reason} = error ->
+        error
+    end
+  end
+
   @impl true
   def handle_init(ctx, options) do
     case Keyword.get(options, :module, :default) do

test/membrane/testing/pipeline_test.exs

@@ -1,7 +1,10 @@
 defmodule Membrane.Testing.PipelineTest do
   use ExUnit.Case
 
-  alias Membrane.ChildrenSpec
+  import Membrane.ChildrenSpec
+  import Membrane.Testing.Assertions
+
+  alias Membrane.Child
   alias Membrane.Testing.Pipeline
 
   defmodule Elem do
@@ -17,7 +20,6 @@ defmodule Membrane.Testing.PipelineTest do
 
   describe "Testing pipeline creation" do
     test "works with :default implementation" do
-      import ChildrenSpec
       elements = [elem: Elem, elem2: Elem]
       links = [get_child(:elem) |> get_child(:elem2)]
       options = [module: :default, spec: elements ++ links, test_process: nil]
@@ -29,7 +31,6 @@ defmodule Membrane.Testing.PipelineTest do
     end
 
     test "by default chooses :default implementation" do
-      import ChildrenSpec
       links = [child(:elem, Elem) |> child(:elem2, Elem)]
       options = [module: :default, spec: links, test_process: nil]
       assert {[spec: spec, playback: :playing], state} = Pipeline.handle_init(%{}, options)
@@ -53,8 +54,6 @@ defmodule Membrane.Testing.PipelineTest do
 
   describe "When initializing Testing Pipeline" do
     test "uses prepared links if they were provided" do
-      import ChildrenSpec
-
       links = [child(:elem, Elem) |> child(:elem2, Elem)]
       options = [module: :default, spec: links, test_process: nil]
       assert {[spec: spec, playback: :playing], state} = Pipeline.handle_init(%{}, options)
@@ -75,4 +74,99 @@ defmodule Membrane.Testing.PipelineTest do
       assert_raise RuntimeError, ~r/Unknown module./, fn -> reraise exception, stacktrace end
     end
   end
+
+  test "get_child_pid/3" do
+    defmodule Element do
+      use Membrane.Filter
+
+      @impl true
+      def handle_parent_notification({:get_pid, msg_id}, _ctx, state) do
+        {[notify_parent: {:pid, self(), msg_id}], state}
+      end
+    end
+
+    defmodule Bin do
+      use Membrane.Bin
+
+      @impl true
+      def handle_init(_ctx, _opts) do
+        spec = [
+          child(:element_1, Element),
+          child(:element_2, Element),
+          child(:element_3, Element)
+        ]
+
+        {[spec: spec], %{}}
+      end
+
+      @impl true
+      def handle_parent_notification({:get_pid, msg_id}, _ctx, state) do
+        {[notify_parent: {:pid, self(), msg_id}], state}
+      end
+
+      @impl true
+      def handle_parent_notification({:get_child_pid, child, msg_id}, _ctx, state) do
+        {[notify_child: {child, {:get_pid, msg_id}}], state}
+      end
+
+      @impl true
+      def handle_child_notification(msg, _child, _ctx, state) do
+        {[notify_parent: msg], state}
+      end
+    end
+
+    spec = [
+      child(:bin_1, Bin),
+      child(:bin_2, Bin),
+      child(:bin_3, Bin)
+    ]
+
+    pipeline = Pipeline.start_supervised!(spec: spec)
+
+    assert_pipeline_play(pipeline)
+
+    # getting children pids from pipeline
+    for bin <- [:bin_1, :bin_2, :bin_3] do
+      Pipeline.execute_actions(pipeline, notify_child: {bin, {:get_pid, bin}})
+      assert_pipeline_notified(pipeline, bin, {:pid, bin_pid, ^bin})
+
+      assert {:ok, bin_pid} == Pipeline.get_child_pid(pipeline, bin)
+    end
+
+    # getting children pids from bins
+    for bin <- [:bin_1, :bin_2, :bin_3], element <- [:element_1, :element_2, :element_3] do
+      Pipeline.execute_actions(pipeline,
+        notify_child: {bin, {:get_child_pid, element, {bin, element}}}
+      )
+
+      assert_pipeline_notified(pipeline, bin, {:pid, element_pid, {^bin, ^element}})
+
+      assert {:ok, element_pid} == Pipeline.get_child_pid(pipeline, [bin, element])
+    end
+
+    # getting pid of child from child group
+    Pipeline.execute_actions(pipeline, spec: {child(:element, Element), group: :group})
+
+    element_ref = Child.ref(:element, group: :group)
+
+    Pipeline.execute_actions(pipeline,
+      notify_child: {element_ref, {:get_pid, element_ref}}
+    )
+
+    assert_pipeline_notified(pipeline, element_ref, {:pid, element_pid, ^element_ref})
+
+    assert {:ok, element_pid} == Pipeline.get_child_pid(pipeline, element_ref)
+
+    # returning error tuple with proper reason
+    assert {:error, :child_not_found} = Pipeline.get_child_pid(pipeline, :nonexisting_child)
+
+    assert {:error, :element_cannot_have_children} =
+             Pipeline.get_child_pid(pipeline, [element_ref, :child])
+
+    monitor_ref = Process.monitor(pipeline)
+    Pipeline.terminate(pipeline)
+    assert_receive {:DOWN, ^monitor_ref, :process, ^pipeline, _reason}
+
+    assert {:error, :pipeline_not_alive} = Pipeline.get_child_pid(pipeline, :bin_1)
+  end
 end