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 state 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/2`. 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: ``` children = [ el1: MembraneElement1, el2: MembraneElement2, ... ] options = [ module: :default # :default is the default value for this parameter, so you do not need to pass it here links: Membrane.ParentSpec.link_linear(children) ] {:ok, pipeline} = Membrane.Testing.Pipeline.start_link(options) ``` Note, that we have used `Membrane.Testing.ParentSpec.link_linear/1` function, that creates the list of links for the given list of children, linking them in linear manner (that means - children are linked in a way that `:output` pad of a given child is linked to `:input` pad of subsequent child). That is the case which is often used while creating testing pipelines. Be aware, that `Membrane.Testing.ParentSpec.link_linear/1` creates also a children specification itself, which means, that you cannot pass that children specification as another option's argument (adding `children: children` option would lead to a duplication of children specifications). If you need to link children in a different manner, you can of course do it by passing an appropriate list of links as a `:links` option, just as you would do with a regular pipeline. 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 ] {:ok, pipeline} = Membrane.Testing.Pipeline.start_link(options) ``` See `Membrane.Testing.Pipeline.pipeline_keyword_list_t()` 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 `message_child/3` for more details. ## Example usage Firstly, we can start the pipeline providing its options as a keyword list: children = [ source: %Membrane.Testing.Source{}, tested_element: TestedElement, sink: %Membrane.Testing.Sink{} ] {:ok, pipeline} = Membrane.Testing.Pipeline.start_link(links: Membrane.ParentSpec.link_linear(children)) 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.{Element, Pipeline} alias Membrane.ParentSpec alias Membrane.Testing.Notification require Membrane.Logger defmodule Options do @moduledoc """ @deprecated Structure representing `options` passed to testing pipeline. ## Struct fields - `:test_process` - `pid` of process that shall receive messages from testing pipeline, e.g. when pipeline's playback state changes. This allows using `Membrane.Testing.Assertions` - `:elements` - a list of element specs. Allows to create a simple pipeline without defining a module for it. - `:links` - a list describing the links between children. If ommited (or set to `nil`), they will be populated automatically based on the children order using default pad names. - `:module` - pipeline module with custom callbacks - useful if a simple list of children is not enough. - `:custom_args`- arguments for the module's `handle_init` callback. """ defstruct [:elements, :links, :test_process, :module, :custom_args] @type t :: %__MODULE__{ test_process: pid() | nil, elements: ParentSpec.children_spec_t() | nil, links: ParentSpec.links_spec_t() | nil, module: module() | nil, custom_args: Pipeline.pipeline_options_t() | nil } end defmodule State do @moduledoc """ Structure representing `state`. ## Test Process `pid` of process that shall receive messages when Pipeline invokes playback state change callback and receives notification. ## Module Pipeline Module with custom callbacks. ## Custom Pipeline State State of the pipeline defined by Module. """ @enforce_keys [:test_process, :module] defstruct @enforce_keys ++ [:custom_pipeline_state] @type t :: %__MODULE__{ test_process: pid() | nil, module: module() | nil, custom_pipeline_state: any } end @type default_pipeline_keyword_list_t :: [ module: :default, children: ParentSpec.children_spec_t(), links: ParentSpec.links_spec_t(), test_process: pid() ] @type custom_pipeline_keyword_list_t :: [ module: module(), custom_args: Pipeline.pipeline_options_t(), test_process: pid() ] @type pipeline_keyword_list_t :: default_pipeline_keyword_list_t() | custom_pipeline_keyword_list_t() @spec start_link(Options.t() | pipeline_keyword_list_t(), GenServer.options()) :: GenServer.on_start() def start_link(pipeline_options, process_options \\ []) def start_link(pipeline_options, process_options) when is_struct(pipeline_options, Options) do Membrane.Logger.warn( "Please pass options to Membrane.Testing.Pipeline.start_link/2 as keyword list, instead of using Membrane.Testing.Options" ) do_start(:start_link, pipeline_options, process_options) end def start_link(pipeline_options, process_options) do pipeline_options = transform_pipeline_options(pipeline_options) do_start(:start_link, pipeline_options, process_options) end @spec start(Options.t() | pipeline_keyword_list_t(), GenServer.options()) :: GenServer.on_start() def start(pipeline_options, process_options \\ []) def start(pipeline_options, process_options) when is_struct(pipeline_options, Options) do Membrane.Logger.warn( "Please pass options to Membrane.Testing.Pipeline.start/2 as keyword list, instead of using Membrane.Testing.Options" ) do_start(:start, pipeline_options, process_options) end def start(pipeline_options, process_options) do pipeline_options = transform_pipeline_options(pipeline_options) do_start(:start, pipeline_options, process_options) end defp transform_pipeline_options(pipeline_options) do module = Keyword.get(pipeline_options, :module, :default) case module do :default -> children = Keyword.get(pipeline_options, :children, []) links = Keyword.get(pipeline_options, :links, []) test_process = Keyword.get(pipeline_options, :test_process) %{module: :default, children: children, links: links, test_process: test_process} module when is_atom(module) -> case Code.ensure_compiled(module) do {:module, _} -> custom_args = Keyword.get(pipeline_options, :custom_args) test_process = Keyword.get(pipeline_options, :test_process) %{module: module, custom_args: custom_args, test_process: test_process} {:error, _} -> raise "Unknown module: #{inspect(module)}" end not_a_module -> raise "Not a module: #{inspect(not_a_module)}" end end defp do_start(_type, %Options{elements: nil, module: nil}, _process_options) do raise """ You provided no information about pipeline contents. Please provide either: - list of children via `:children` field of Options struct with optional links between them via `:links` field of `Options` struct - module that implements `Membrane.Pipeline` callbacks via `module` field of `Options` struct """ end defp do_start(_type, %Options{elements: children, module: module}, _process_options) when is_atom(module) and module != nil and children != nil do raise """ When working with Membrane.Testing.Pipeline you can't provide both override module and children list in the Membrane.Testing.Pipeline.Options struct. """ end defp do_start(type, options, process_options) do pipeline_options = default_options(options) args = [__MODULE__, pipeline_options, process_options] apply(Pipeline, type, args) end @doc """ 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"}) """ @spec message_child(pid(), Element.name_t(), any()) :: :ok def message_child(pipeline, child, message) do send(pipeline, {:for_element, child, message}) :ok end @doc """ Executes specified actions in the pipeline. The actions are returned from the `handle_other` callback. """ @spec execute_actions(pid(), Keyword.t()) :: :ok def execute_actions(pipeline, actions) do send(pipeline, {__MODULE__, :__execute_actions__, actions}) :ok end @impl true def handle_init(%Options{links: nil, module: nil} = options) do {links, children} = if length(options.elements) <= 1 do {[], options.elements} else {ParentSpec.link_linear(options.elements), []} end options_map = %{children: children, links: links, test_process: options.test_process} do_handle_init_for_default_implementation(options_map) end @impl true def handle_init(%Options{module: nil} = options) do options_map = %{ children: options.elements, links: options.links, test_process: options.test_process } do_handle_init_for_default_implementation(options_map) end @impl true def handle_init(%Options{links: nil, elements: nil} = options) do options_map = %{ test_process: options.test_process, module: options.module, custom_args: options.custom_args } do_handle_init_with_custom_module(options_map) end @impl true def handle_init(%{module: :default} = options) do do_handle_init_for_default_implementation(options) end @impl true def handle_init(%{module: _module} = options) do do_handle_init_with_custom_module(options) end defp do_handle_init_for_default_implementation(options) do spec = %Membrane.ParentSpec{ children: options.children, links: options.links } new_state = %State{test_process: options.test_process, module: nil} {{:ok, [spec: spec, playback: :playing]}, new_state} end defp do_handle_init_with_custom_module(options) do new_state = %State{ test_process: options.test_process, module: options.module, custom_pipeline_state: options.custom_args } injected_module_result = eval_injected_module_callback(:handle_init, [], new_state) testing_pipeline_result = {:ok, new_state} combine_results(injected_module_result, testing_pipeline_result) end @impl true def handle_stopped_to_prepared(ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_stopped_to_prepared, [ctx], state ) :ok = notify_playback_state_changed(state.test_process, :stopped, :prepared) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_prepared_to_playing(ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_prepared_to_playing, [ctx], state ) :ok = notify_playback_state_changed(state.test_process, :prepared, :playing) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_playing_to_prepared(ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_playing_to_prepared, [ctx], state ) :ok = notify_playback_state_changed(state.test_process, :playing, :prepared) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_prepared_to_stopped(ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_prepared_to_stopped, [ctx], state ) :ok = notify_playback_state_changed(state.test_process, :prepared, :stopped) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_notification( %Notification{payload: notification}, from, _ctx, %State{} = state ) do :ok = notify_test_process(state.test_process, {:handle_notification, {notification, from}}) {:ok, state} end @impl true def handle_notification(notification, from, ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_notification, [notification, from, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_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 @impl true def handle_other({__MODULE__, :__execute_actions__, actions}, _ctx, %State{} = state) do {{:ok, actions}, state} end @impl true def handle_other({:for_element, element, message}, ctx, %State{} = state) do injected_module_result = eval_injected_module_callback( :handle_other, [{:for_element, element, message}, ctx], state ) testing_pipeline_result = {{:ok, forward: {element, message}}, state} combine_results(injected_module_result, testing_pipeline_result) end @impl true def handle_other(message, ctx, %State{} = state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_other, [message, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_other, message}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_element_start_of_stream(endpoint, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_element_start_of_stream, [endpoint, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_element_start_of_stream, endpoint}) {custom_actions, Map.put(state, :custom_pipeline_state, custom_state)} end @impl true def handle_element_end_of_stream(endpoint, ctx, state) do {custom_actions, custom_state} = eval_injected_module_callback( :handle_element_end_of_stream, [endpoint, ctx], state ) :ok = notify_test_process(state.test_process, {:handle_element_end_of_stream, endpoint}) {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_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 default_options(%Options{test_process: nil} = options), do: %Options{options | test_process: self()} defp default_options(%{test_process: nil} = options), do: %{options | test_process: self()} defp default_options(options), do: options defp eval_injected_module_callback(callback, args, state) defp eval_injected_module_callback(_callback, _args, %State{module: nil} = state), do: {:ok, state} |> unify_result() defp eval_injected_module_callback(callback, args, state) do apply(state.module, callback, args ++ [state.custom_pipeline_state]) |> unify_result() end defp notify_playback_state_changed(test_process, previous, current) do notify_test_process(test_process, {:playback_state_changed, previous, current}) end defp notify_test_process(test_process, message) do send(test_process, {__MODULE__, self(), message}) :ok end defp unify_result({:ok, state}), do: {{:ok, []}, state} defp unify_result({{_, _}, _} = result), do: result defp combine_results({custom_actions, custom_state}, {actions, state}) do {combine_actions(custom_actions, actions), Map.put(state, :custom_pipeline_state, custom_state)} end defp combine_actions(l, r) do case {l, r} do {l, :ok} -> l {{:ok, actions_l}, {:ok, actions_r}} -> {:ok, actions_l ++ actions_r} {_l, r} -> r end end end