defmodule Membrane.Testing.Pipeline do @moduledoc """ This Pipeline was created to reduce testing boilerplate and ease communication with its children. It also provides a utility for informing testing process about playback changes and received notifications. When you want a build Pipeline to test your children you need three things: - Pipeline Module - List of children - Links between those children To start a testing pipeline you need to build a keyword list representing the options used to determine the pipeline's behaviour and then pass that options list to the `Membrane.Testing.Pipeline.start_link_supervised!/1`. The testing pipeline can be started in one of two modes - either with its `:default` behaviour, or by injecting a custom module behaviour. The usage of a `:default` pipeline implementation is presented below: ``` spec = [ child(:el1, MembraneElement1) |> child(:el2, MembraneElement2) ... ] options = [ module: :default # :default is the default value for this parameter, so you do not need to pass it here spec: spec ] pipeline = Membrane.Testing.Pipeline.start_link_supervised!(options) ``` You can also pass your custom pipeline's module as a `:module` option of the options list. Every callback of the module will be executed before the callbacks of Testing.Pipeline. Passed module has to return a proper spec. There should be no children nor links specified in options passed to test pipeline as that would result in a failure. ``` options = [ module: Your.Module ] pipeline = Membrane.Testing.Pipeline.start_link_supervised!(options) ``` See `t:Membrane.Testing.Pipeline.options/0` for available options. ## Assertions This pipeline is designed to work with `Membrane.Testing.Assertions`. Check them out or see example below for more details. ## Messaging children You can send messages to children using their names specified in the children list. Please check `notify_child/3` for more details. ## Example usage Firstly, we can start the pipeline providing its options as a keyword list: ``` import Membrane.ChildrenSpec spec = [ child(:source, %Membrane.Testing.Source{output: [1, 2, 3]}) |> child(:tested_element, TestedElement) |> child(:sink, Membrane.Testing.Sink) ] {:ok, pipeline} = Membrane.Testing.Pipeline.start_link(spec: spec) ``` We can now wait till the end of the stream reaches the sink element (don't forget to import `Membrane.Testing.Assertions`): assert_end_of_stream(pipeline, :sink) We can also assert that the `Membrane.Testing.Sink` processed a specific buffer: assert_sink_buffer(pipeline, :sink, %Membrane.Buffer{payload: 1}) """ use Membrane.Pipeline alias Membrane.Child alias Membrane.ChildrenSpec alias Membrane.Core.Message alias Membrane.{Element, Pipeline} alias Membrane.Testing.Notification require Membrane.Core.Message defmodule State do @moduledoc false @enforce_keys [:test_process, :module] defstruct @enforce_keys ++ [:custom_pipeline_state, :raise_on_child_pad_removed?] @typedoc """ Structure for holding state ## Test Process `pid` of process that shall receive messages from the pipeline ## Module Pipeline Module with custom callbacks. ## Custom Pipeline State State of the pipeline defined by Module. """ @type t :: %__MODULE__{ test_process: pid() | nil, module: module() | nil, custom_pipeline_state: Pipeline.state(), raise_on_child_pad_removed?: boolean() | nil } end @type options :: [ module: :default, spec: [ChildrenSpec.builder()], test_process: pid(), name: Pipeline.name(), raise_on_child_pad_removed?: boolean() ] | [ module: module(), custom_args: Pipeline.pipeline_options(), test_process: pid(), name: Pipeline.name() ] @spec child_spec(options) :: Supervisor.child_spec() def child_spec(options) do id = Keyword.get(options, :name, make_ref()) super(options) |> Map.merge(%{restart: :transient, id: id}) end @spec start_link(options) :: Pipeline.on_start() def start_link(options) do do_start(:start_link, options) end @spec start(options) :: Pipeline.on_start() def start(options) do do_start(:start, options) end defp do_start(type, options) do :ok = validate_options!(options) {process_options, options} = Keyword.split(options, [:name]) options = Keyword.put_new(options, :test_process, self()) apply(Pipeline, type, [__MODULE__, options, process_options]) end @doc """ Starts the pipeline under the ExUnit test supervisor and links it to the current process. Can be used only in tests. """ @spec start_link_supervised(options) :: Pipeline.on_start() def start_link_supervised(pipeline_options \\ []) do pipeline_options = Keyword.put_new(pipeline_options, :test_process, self()) # TODO use start_link_supervised when added with {:ok, supervisor, pipeline} <- ex_unit_start_supervised({__MODULE__, pipeline_options}) do Process.link(pipeline) {:ok, supervisor, pipeline} else {:error, {error, _child_info}} -> {:error, error} end end @spec start_link_supervised!(options) :: pipeline_pid :: pid def start_link_supervised!(pipeline_options \\ []) do {:ok, _supervisor, pipeline} = start_link_supervised(pipeline_options) pipeline end @doc """ Starts the pipeline under the ExUnit test supervisor. Can be used only in tests. """ @spec start_supervised(options) :: Pipeline.on_start() def start_supervised(pipeline_options \\ []) do pipeline_options = Keyword.put_new(pipeline_options, :test_process, self()) ex_unit_start_supervised({__MODULE__, pipeline_options}) end @spec start_supervised!(options) :: pipeline_pid :: pid def start_supervised!(pipeline_options \\ []) do {:ok, _supervisor, pipeline} = start_supervised(pipeline_options) pipeline end defp ex_unit_start_supervised(child_spec) do # It's not a 'normal' call to keep dialyzer quiet # credo:disable-for-next-line Credo.Check.Refactor.Apply apply(ExUnit.Callbacks, :start_supervised, [child_spec]) end @spec terminate(pid, Keyword.t()) :: :ok | {:ok, pid()} | {:error, :timeout} defdelegate terminate(pipeline, opts \\ []), to: Pipeline @doc """ Sends notification to a child by Element name. ## Example Knowing that `pipeline` has child named `sink`, notification can be sent as follows: notify_child(pipeline, :sink, {:notification, "to handle"}) """ @spec notify_child(pid(), Element.name(), any()) :: :ok def notify_child(pipeline, child, notification) do send(pipeline, {:for_element, child, notification}) :ok end @doc """ Deprecated since `v1.1.0`, use `notify_child/3` instead. Sends message to a child by Element name. ## Example Knowing that `pipeline` has child named `sink`, message can be sent as follows: message_child(pipeline, :sink, {:message, "to handle"}) """ @deprecated "Use #{inspect(__MODULE__)}.notify_child/3 instead" @spec message_child(pid(), Element.name(), any()) :: :ok def message_child(pipeline, child, message) do notify_child(pipeline, child, message) end @doc """ Executes specified actions in the pipeline. The actions are returned from the `handle_info` callback. """ @spec execute_actions(pid(), Keyword.t()) :: :ok def execute_actions(pipeline, actions) do send(pipeline, {__MODULE__, :__execute_actions__, actions}) :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_name_path :: Child.name() | [Child.name()]) :: {:ok, pid()} | {:error, reason :: term()} def get_child_pid(pipeline, [_head | _tail] = child_name_path) do do_get_child_pid(pipeline, child_name_path) end def get_child_pid(pipeline, child_name) when not is_list(child_name) do do_get_child_pid(pipeline, [child_name]) 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_name_path :: Child.name() | [Child.name()]) :: pid() def get_child_pid!(parent_pid, child_name_path) do {:ok, child_pid} = get_child_pid(parent_pid, child_name_path) child_pid end defp do_get_child_pid(component_pid, child_name_path, pipeline? \\ true) defp do_get_child_pid(component_pid, [], _pipeline?) do {:ok, component_pid} end defp do_get_child_pid(component_pid, [child_name | child_name_path_tail], pipeline?) do case Message.call(component_pid, :get_child_pid, child_name) do {:ok, child_pid} -> do_get_child_pid(child_pid, child_name_path_tail, false) {:error, {:call_failure, {:noproc, _call_info}}} -> if 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 :default -> spec = Bunch.listify(Keyword.get(options, :spec, [])) test_process = Keyword.fetch!(options, :test_process) raise? = Keyword.get(options, :raise_on_child_pad_removed?, true) new_state = %State{ test_process: test_process, raise_on_child_pad_removed?: raise?, module: nil } {[spec: spec], new_state} module when is_atom(module) -> case Code.ensure_compiled(module) do {:module, _module} -> :ok {:error, _error} -> raise "Unknown module: #{inspect(module)}" end new_state = %State{ module: module, custom_pipeline_state: Keyword.get(options, :custom_args), test_process: Keyword.fetch!(options, :test_process) } injected_module_result = eval_injected_module_callback(:handle_init, [ctx], new_state) testing_pipeline_result = {[], new_state} combine_results(injected_module_result, testing_pipeline_result) not_a_module -> raise "Not a module: #{inspect(not_a_module)}" end end @impl true def handle_setup(ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_setup, [ctx], state ) :ok = notify_test_process(state.test_process, :setup) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_playing(ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_playing, [ctx], state ) :ok = notify_test_process(state.test_process, :play) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_child_notification( %Notification{payload: notification}, from, _ctx, %State{} = state ) do :ok = notify_test_process(state.test_process, {:handle_child_notification, {notification, from}}) {[], state} end @impl true def handle_child_notification(notification, from, ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_child_notification, [notification, from, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_child_notification, {notification, from}}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_spec_started(children, ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_spec_started, [children, ctx], state ) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end [ :handle_child_setup_completed, :handle_child_playing, :handle_child_terminated ] |> Enum.map(fn callback -> @impl true def unquote(callback)(child, ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( unquote(callback), [child, ctx], state ) :ok = notify_test_process(state.test_process, {unquote(callback), child}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end end) @impl true def handle_info({__MODULE__, :__execute_actions__, actions}, _ctx, %State{} = state) do {actions, state} end @impl true def handle_info({:for_element, element, message}, ctx, %State{} = state) do injected_module_result = eval_injected_module_callback( :handle_info, [{:for_element, element, message}, ctx], state ) testing_pipeline_result = {[notify_child: {element, message}], state} combine_results(injected_module_result, testing_pipeline_result) end @impl true def handle_info(message, ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_info, [message, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_info, message}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_call(message, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_call, [message, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_call, message}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_element_start_of_stream(element, pad, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_element_start_of_stream, [element, pad, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_element_start_of_stream, {element, pad}}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_element_end_of_stream(element, pad, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_element_end_of_stream, [element, pad, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_element_end_of_stream, {element, pad}}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_tick(timer, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_tick, [timer, ctx], state ) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_child_pad_removed(child, pad, ctx, state) do if state.raise_on_child_pad_removed? do raise """ Child #{inspect(child)} removed it's pad #{inspect(pad)}. If you want to handle such a scenario, pass `raise_on_child_pad_removed?: false` option to `Membrane.Testing.Pipeline.start_*/2` or pass there a pipeline module implementing this callback via `:module` option. """ end {custom_actions, custom_state} = eval_injected_module_callback( :handle_child_pad_removed, [child, pad, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_child_pad_removed, {child, pad}}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_crash_group_down(group_name, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_crash_group_down, [group_name, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_crash_group_down, group_name}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end defp eval_injected_module_callback(callback, args, state) defp eval_injected_module_callback(_callback, _args, %State{module: nil}), do: {[], nil} defp eval_injected_module_callback(callback, args, state) do if callback != :handle_spec_started or function_exported?(state.module, :handle_spec_started, 3) do apply(state.module, callback, args ++ [state.custom_pipeline_state]) else {[], state.custom_pipeline_state} end end defp notify_test_process(test_process, message) do send(test_process, {__MODULE__, self(), message}) :ok end defp combine_results({custom_actions, custom_state}, {actions, state}) do {Enum.concat(custom_actions, actions), Map.put(state, :custom_pipeline_state, custom_state)} end defp validate_options!(options) do allowed_keys_1 = [:module, :mode, :spec, :test_process, :name, :raise_on_child_pad_removed?] allowed_keys_2 = [:module, :custom_args, :test_process, :name] with {:error, _keys} <- Keyword.validate(options, allowed_keys_1), {:error, _keys} <- Keyword.validate(options, allowed_keys_2) do raise """ Options passed to #{inspect(__MODULE__)} start function has to fulfill type #{inspect(__MODULE__)}.options, while they are #{inspect(options)} """ else {:ok, _keyword} -> :ok end end end