Fix & improve tests

Author: FelonEkonom

test/membrane/buffer_metric/byte_size_test.exs

@@ -13,32 +13,31 @@ defmodule Membrane.Buffer.Metric.ByteSizeTest do
 
   describe ".buffers_size/1" do
     test "should return size of all buffers" do
-      size = ByteSize.buffers_size(@buffers)
-      assert size == byte_size(@pay1) + byte_size(@pay2)
+      assert ByteSize.buffers_size(@buffers) == {:ok, byte_size(@pay1) + byte_size(@pay2)}
     end
   end
 
-  describe ".split_buffers/2" do
+  describe ".split_buffers/4" do
     test "when split position matches size of first buffer, extract only first buffer" do
-      {buf, rest} = ByteSize.split_buffers(@buffers, byte_size(@pay1), nil)
+      {buf, rest} = ByteSize.split_buffers(@buffers, byte_size(@pay1), nil, nil)
       assert buf == [@buf1]
       assert rest == [@buf2]
     end
 
     test "when there is only one buffer where split position is greater than buffer size \
       returns the buffer and an empty list" do
-      {buf, []} = ByteSize.split_buffers(@single_buffer, byte_size(@pay1) + 10, nil)
+      {buf, []} = ByteSize.split_buffers(@single_buffer, byte_size(@pay1) + 10, nil, nil)
       assert buf == [@buf1]
     end
 
     test "when there is only one buffer where split position is 0, it returns an empty \
       list and a list with the buffer" do
-      {[], rest} = ByteSize.split_buffers(@single_buffer, 0, nil)
+      {[], rest} = ByteSize.split_buffers(@single_buffer, 0, nil, nil)
       assert rest == [@buf1]
     end
 
     test "when splitting is necessary it extracts the first buffer and splits the second into two" do
-      {extracted, rest} = ByteSize.split_buffers(@buffers, byte_size(@pay1) + 1, nil)
+      {extracted, rest} = ByteSize.split_buffers(@buffers, byte_size(@pay1) + 1, nil, nil)
       <<one_byte::binary-size(1), expected_rest::binary>> = @pay2
       assert extracted == [@buf1, %Membrane.Buffer{payload: one_byte}]
       assert rest == [%Membrane.Buffer{payload: expected_rest}]

test/membrane/buffer_metric/count_test.exs

@@ -11,13 +11,13 @@ defmodule Membrane.Buffer.Metric.CountTest do
 
   describe ".buffers_size/1" do
     test "should return count of all buffers" do
-      assert Count.buffers_size(@buffers) == 2
+      assert Count.buffers_size(@buffers) == {:ok, 2}
     end
   end
 
-  describe ".split_buffers/2" do
+  describe ".split_buffers/4" do
     test "should return split buffers" do
-      {extracted, rest} = Count.split_buffers(@buffers, @count)
+      {extracted, rest} = Count.split_buffers(@buffers, @count, nil, nil)
 
       assert extracted == [@buf1]
       assert rest == [@buf2]

test/membrane/buffer_metric/timestamp_test.exs

@@ -0,0 +1,143 @@
+defmodule Membrane.Buffer.Metric.TimestampTest do
+  use ExUnit.Case, async: true
+
+  import ExUnit.CaptureLog
+
+  alias Membrane.Buffer
+  alias Membrane.Buffer.Metric.Timestamp.{DTS, DTSOrPTS, PTS}
+
+  @t0 0 |> Membrane.Time.milliseconds()
+  @t1 100 |> Membrane.Time.milliseconds()
+  @t2 250 |> Membrane.Time.milliseconds()
+  @t3 350 |> Membrane.Time.milliseconds()
+  @t4 500 |> Membrane.Time.milliseconds()
+
+  defp buf(ts_field, ts), do: struct(%Buffer{payload: <<>>}, [{ts_field, ts}])
+
+  test ".init_manual_demand_size_value/0 returns -1 as the no-demand sentinel" do
+    assert PTS.init_manual_demand_size_value() == -1
+    assert DTS.init_manual_demand_size_value() == -1
+    assert DTSOrPTS.init_manual_demand_size_value() == -1
+  end
+
+  test ".reduce_demand/2 always returns the demand unchanged regardless of consumed size" do
+    for module <- [PTS, DTS, DTSOrPTS] do
+      assert module.reduce_demand(1_000, 5) == 1_000
+      assert module.reduce_demand(1_000, nil) == 1_000
+    end
+  end
+
+  test ".buffers_size/1 returns {:error, :operation_not_supported}" do
+    for module <- [PTS, DTS, DTSOrPTS] do
+      assert module.buffers_size([]) == {:error, :operation_not_supported}
+      assert module.buffers_size([%Buffer{payload: <<>>}]) == {:error, :operation_not_supported}
+    end
+  end
+
+  describe ".split_buffers/4" do
+    for {module, name, ts_field} <- [{PTS, "PTS", :pts}, {DTS, "DTS", :dts}] do
+      test "returns {[], buffers} when demand is the initial sentinel value (-1) for #{name}" do
+        buffers = Enum.map([@t0, @t1, @t2], &buf(unquote(ts_field), &1))
+        assert unquote(module).split_buffers(buffers, -1, nil, nil) == {[], buffers}
+      end
+
+      test "uses first buffer's #{name} as offset when no buffers have been consumed yet" do
+        buffers = Enum.map([@t0, @t1, @t2, @t3, @t4], &buf(unquote(ts_field), &1))
+        # offset = @t0 = 0; demand = 300 ns → consume until ts - 0 >= 300 → stops at @t3 = 350
+        {consumed, remaining} = unquote(module).split_buffers(buffers, 300, nil, nil)
+        assert Enum.map(consumed, &Map.get(&1, unquote(ts_field))) == [@t0, @t1, @t2, @t3]
+        assert Enum.map(remaining, &Map.get(&1, unquote(ts_field))) == [@t4]
+      end
+
+      test "uses first_consumed_buffer's #{name} as offset when buffers have been consumed" do
+        buffers = Enum.map([@t0, @t1, @t2, @t3, @t4], &buf(unquote(ts_field), &1))
+        # first_consumed at @t0 → same offset as above, same split point
+        first_consumed = buf(unquote(ts_field), @t0)
+        last_consumed = buf(unquote(ts_field), @t1)
+
+        {consumed, remaining} =
+          unquote(module).split_buffers(buffers, 300, first_consumed, last_consumed)
+
+        assert Enum.map(consumed, &Map.get(&1, unquote(ts_field))) == [@t0, @t1, @t2, @t3]
+        assert Enum.map(remaining, &Map.get(&1, unquote(ts_field))) == [@t4]
+      end
+
+      test "returns all buffers for #{name} when demand exceeds the available timestamp range" do
+        buffers = Enum.map([@t0, @t1, @t2, @t3, @t4], &buf(unquote(ts_field), &1))
+        {consumed, remaining} = unquote(module).split_buffers(buffers, 10_000, nil, nil)
+        assert consumed == buffers
+        assert remaining == []
+      end
+
+      test "emits a warning and returns {[], buffers} for #{name} when elapsed duration already meets demand" do
+        buffers = Enum.map([@t0, @t1, @t2, @t3, @t4], &buf(unquote(ts_field), &1))
+        # last_consumed - first_consumed = @t4 - @t0 = 500 >= demand = 300
+        first_consumed = buf(unquote(ts_field), @t0)
+        last_consumed = buf(unquote(ts_field), @t4)
+
+        log =
+          capture_log(fn ->
+            assert unquote(module).split_buffers(buffers, 300, first_consumed, last_consumed) ==
+                     {[], buffers}
+          end)
+
+        assert log =~ "warning"
+      end
+    end
+
+    test "DTSOrPTS prefers DTS when present, falls back to PTS" do
+      # first buffer has dts=@t0 and pts=@t4: metric should use dts
+      buf_with_dts = %Buffer{payload: <<>>, dts: @t0, pts: @t4}
+      buf_pts_only = %Buffer{payload: <<>>, pts: @t3}
+
+      # offset = dts=@t0=0; buf_with_dts: 0-0=0 < 300 → include
+      # buf_pts_only uses pts=@t3=350: 350-0=350 >= 300 → include, stop
+      {consumed, remaining} = DTSOrPTS.split_buffers([buf_with_dts, buf_pts_only], 300, nil, nil)
+      assert consumed == [buf_with_dts, buf_pts_only]
+      assert remaining == []
+    end
+  end
+
+  describe ".generate_metric_specific_warnings/1" do
+    test "returns :ok for an empty list" do
+      assert PTS.generate_metric_specific_warnings([]) == :ok
+      assert DTS.generate_metric_specific_warnings([]) == :ok
+      assert DTSOrPTS.generate_metric_specific_warnings([]) == :ok
+    end
+
+    for {module, name, ts_field} <- [{PTS, "PTS", :pts}, {DTS, "DTS", :dts}] do
+      test "emits no warning for monotonically increasing #{name}s" do
+        buffers = Enum.map([@t0, @t1, @t2, @t3], &buf(unquote(ts_field), &1))
+
+        log =
+          capture_log(fn ->
+            assert unquote(module).generate_metric_specific_warnings(buffers) == :ok
+          end)
+
+        assert log == ""
+      end
+
+      test "emits a warning for non-monotonic #{name}s" do
+        buffers = Enum.map([@t0, @t3, @t1, @t4], &buf(unquote(ts_field), &1))
+
+        log =
+          capture_log(fn ->
+            assert unquote(module).generate_metric_specific_warnings(buffers) == :ok
+          end)
+
+        assert log =~ "warning"
+      end
+    end
+
+    test "DTSOrPTS emits a warning for non-monotonic DTS-or-PTS values" do
+      buffers = Enum.map([@t0, @t3, @t1, @t4], &%Buffer{payload: <<>>, dts: &1})
+
+      log =
+        capture_log(fn ->
+          assert DTSOrPTS.generate_metric_specific_warnings(buffers) == :ok
+        end)
+
+      assert log =~ "warning"
+    end
+  end
+end

test/membrane/core/element/input_queue_test.exs

@@ -373,7 +373,15 @@ defmodule Membrane.Core.Element.InputQueueTest do
   end
 
   defp prepare_input_queue(fields) do
-    struct(InputQueue, [stalker_metrics: %{size: :atomics.new(1, [])}] ++ fields)
+    outbound_metric = Keyword.get(fields, :outbound_metric)
+
+    defaults = [
+      stalker_metrics: %{size: :atomics.new(1, [])},
+      outbound_metric_demand_init_size:
+        if(outbound_metric, do: outbound_metric.init_manual_demand_size_value(), else: 0)
+    ]
+
+    struct(InputQueue, defaults ++ fields)
   end
 
   defp new_atomic_demand(),
@@ -390,10 +398,13 @@ defmodule Membrane.Core.Element.InputQueueTest do
   defp bufs_size(output, unit) do
     {_state, bufs} = output
 
-    Enum.flat_map(bufs, fn {:buffers, bufs_list, _inbound_metric_size, _outbound_metric_size} ->
-      bufs_list
-    end)
-    |> Membrane.Buffer.Metric.from_unit(unit).buffers_size()
+    {:ok, size} =
+      Enum.flat_map(bufs, fn {:buffers, bufs_list, _inbound_metric_size, _outbound_metric_size} ->
+        bufs_list
+      end)
+      |> Membrane.Buffer.Metric.from_unit(unit).buffers_size()
+
+    size
   end
 
   defp bufs(n), do: Enum.to_list(1..n)