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
