Fix tests

c_src/membrane_agora_plugin/sink.cpp

@@ -192,13 +192,13 @@ void handle_destroy_state(UnifexEnv *env, SinkState *state) {
       AG_LOG(ERROR, "Failed to disconnect from Agora channel!");
       return;
     }
-    AG_LOG(INFO, "Disconnected from Agora channel successfully");
+    AG_LOG(INFO, "[Sink] Disconnected from Agora channel successfully");
     state->connection = NULL;
   }
 
   if (state->service) {
     state->service->release();
-    AG_LOG(INFO, "Agora service released successfully");
+    AG_LOG(INFO, "[Sink] Agora service released successfully");
     state->service = NULL;
   }
 }

c_src/membrane_agora_plugin/source.cpp

@@ -109,13 +109,13 @@ void handle_destroy_state(UnifexEnv *env, SourceState *state) {
       AG_LOG(ERROR, "Failed to disconnect from Agora channel!");
       return;
     }
-    AG_LOG(INFO, "Disconnected from Agora channel successfully");
+    AG_LOG(INFO, "[Source] Disconnected from Agora channel successfully");
     state->connection = NULL;
   }
 
   if (state->service) {
     state->service->release();
-    AG_LOG(INFO, "Agora service released successfully");
+    AG_LOG(INFO, "[Source] Agora service released successfully");
     state->service = NULL;
   }
 }

test/integration_test.exs

@@ -2,6 +2,8 @@ defmodule Membrane.Agora.IntegrationTest do
   use ExUnit.Case, async: false
   import Membrane.Testing.Assertions
   require Membrane.Pad, as: Pad
+
+  alias Membrane.Testing
   alias Membrane.Agora.Support.{ReceiverPipeline, SenderPipeline}
 
   describe "if the data is sent to Agora properly when the audio codec is" do
@@ -15,22 +17,24 @@ defmodule Membrane.Agora.IntegrationTest do
         reference_video = input_video
         audio_codec = unquote(audio_codec)
 
-        {input_audio, reference_audio} =
+        input_audio_extension =
           case audio_codec do
-            :aac -> {"test/fixtures/in_audio.aac", "test/fixtures/in_audio.pcm"}
-            :opus -> {"test/fixtures/in_audio_opus.ogg", "test/fixtures/in_audio_opus.pcm"}
+            :aac -> "aac"
+            :opus -> "ogg"
           end
 
+        input_audio = "test/fixtures/in_audio." <> input_audio_extension
+        reference_audio = "test/fixtures/in_audio.pcm"
         output_audio = "#{dir}/audio_#{audio_codec}.pcm"
 
         {:ok, _supervisor, receiver_pipeline} =
-          Membrane.Testing.Pipeline.start_link(
+          Testing.Pipeline.start_link(
             module: ReceiverPipeline,
             custom_args: [audio: output_audio, video: output_video, framerate: framerate]
           )
 
         {:ok, _supervisor, sender_pipeline} =
-          Membrane.Testing.Pipeline.start_link(
+          Testing.Pipeline.start_link(
             module: SenderPipeline,
             custom_args: [audio: input_audio, video: input_video, framerate: framerate]
           )
@@ -55,7 +59,7 @@ defmodule Membrane.Agora.IntegrationTest do
                   |> length()) -
                    (get_h264_frames(output_video)
                     |> length())
-               ) < 40
+               ) < 50
 
         assert abs(File.stat!(reference_audio).size - File.stat!(output_audio).size) <
                  100_000

test/support/receiver_pipeline.ex

@@ -4,6 +4,16 @@ defmodule Membrane.Agora.Support.ReceiverPipeline do
   use Membrane.Pipeline
   alias Membrane.Agora.TokenGenerator
 
+  defmodule Tee do
+    use Membrane.Filter
+
+    def_input_pad :input, accepted_format: _any
+    def_output_pad :output, accepted_format: _any, availability: :on_request
+
+    @impl true
+    def handle_buffer(:input, buffer, _ctx, state), do: {[forward: buffer], state}
+  end
+
   defmodule FramerateAsserter do
     @moduledoc false
     use Membrane.Filter
@@ -49,6 +59,53 @@ defmodule Membrane.Agora.Support.ReceiverPipeline do
     end
   end
 
+
+  defmodule FrameratePrinter do
+    @moduledoc false
+    alias Membrane.Agora.Support.ReceiverPipeline.FrameratePrinter
+    use Membrane.Filter
+
+    @tolerance Membrane.Time.milliseconds(500)
+
+    def_input_pad :input, accepted_format: _any, flow_control: :auto
+    def_output_pad :output, accepted_format: _any, flow_control: :auto
+
+    def_options framerate: []
+
+    @impl true
+    def handle_init(_ctx, opts) do
+      {num, den} = opts.framerate
+      interval = Membrane.Time.seconds(Ratio.new(den, num))
+      {[], %{interval: interval, last_timestamp: nil, how_many_frames: 0}}
+    end
+
+    @impl true
+    def handle_buffer(:input, buffer, _ctx, state) do
+      if state.last_timestamp do
+        diff = buffer.dts - state.last_timestamp
+
+        unless abs(diff - state.interval) < @tolerance,
+          do:
+            raise("""
+            Framerate assertion failed.
+            Value: #{abs(diff - state.interval)}
+            Tolerance: #{@tolerance}
+            """)
+
+        unless abs(state.interval * state.how_many_frames - buffer.dts) < @tolerance,
+          do:
+            raise("""
+            Framerate assertion failed.
+            Value: #{abs(state.interval * state.how_many_frames - buffer.dts)}
+            Tolerance: #{@tolerance}
+            """)
+      end
+
+      {[buffer: {:output, buffer}],
+       %{state | last_timestamp: buffer.dts, how_many_frames: state.how_many_frames + 1}}
+    end
+  end
+
   @channel_name System.get_env("AGORA_CHANNEL_NAME", "")
   @app_id System.get_env("AGORA_APP_ID", "")
   @certificate System.get_env("AGORA_CERTIFICATE", "")
@@ -90,6 +147,7 @@ defmodule Membrane.Agora.Support.ReceiverPipeline do
     spec =
       get_child(:dispatcher_audio)
       |> via_out(Pad.ref(:output, user_id))
+      |> child(:tee, Tee)
       |> child(:audio_sink, %Membrane.File.Sink{location: state.audio})
 
     {[spec: spec], state}