Streaming media item creation during indexing (#49)

* Implemented streaming during indexing

* Updated file watcher to enqueue download; refactored download worker methods

* Updated File Follower Server timeout

test/pinchflat/media_client/backends/yt_dlp/command_runner_test.exs

@@ -10,7 +10,7 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.CommandRunnerTest do
     on_exit(&reset_executable/0)
   end
 
-  describe "run/2" do
+  describe "run/4" do
     test "it returns the output and status when the command succeeds" do
       assert {:ok, _output} = Runner.run(@video_url, [], "")
     end
@@ -57,6 +57,12 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.CommandRunnerTest do
         assert {:error, "", 1} = Runner.run(@video_url, [], "")
       end)
     end
+
+    test "optionally lets you specify an output_filepath" do
+      assert {:ok, output} = Runner.run(@video_url, [], "%(id)s", output_filepath: "/tmp/yt-dlp-output.json")
+
+      assert String.contains?(output, "--print-to-file %(id)s /tmp/yt-dlp-output.json")
+    end
   end
 
   defp wrap_executable(new_executable, fun) do

test/pinchflat/media_client/backends/yt_dlp/video_collection_test.exs

@@ -11,14 +11,16 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoCollectionTest do
 
   describe "get_media_attributes/2" do
     test "returns a list of video attributes with no blank elements" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, source_attributes_return_fixture() <> "\n\n"} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+        {:ok, source_attributes_return_fixture() <> "\n\n"}
+      end)
 
       assert {:ok, [%{"id" => "video1"}, %{"id" => "video2"}, %{"id" => "video3"}]} =
                VideoCollection.get_media_attributes(@channel_url)
     end
 
     test "it passes the expected default args" do
-      expect(YtDlpRunnerMock, :run, fn _url, opts, ot ->
+      expect(YtDlpRunnerMock, :run, fn _url, opts, ot, _addl_opts ->
         assert opts == [:simulate, :skip_download]
         assert ot == "%(.{id,title,was_live,original_url,description})j"
 
@@ -28,20 +30,35 @@ defmodule Pinchflat.MediaClient.Backends.YtDlp.VideoCollectionTest do
       assert {:ok, _} = VideoCollection.get_media_attributes(@channel_url)
     end
 
-    test "it passes the expected custom args" do
-      expect(YtDlpRunnerMock, :run, fn _url, opts, _ot ->
-        assert opts == [:custom_arg, :simulate, :skip_download]
+    test "returns the error straight through when the command fails" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:error, "Big issue", 1} end)
+
+      assert {:error, "Big issue", 1} = VideoCollection.get_media_attributes(@channel_url)
+    end
+
+    test "passes the explict tmpfile path to runner" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts ->
+        assert [{:output_filepath, filepath}] = addl_opts
+        assert String.ends_with?(filepath, ".json")
 
         {:ok, ""}
       end)
 
-      assert {:ok, _} = VideoCollection.get_media_attributes(@channel_url, [:custom_arg])
+      assert {:ok, _} = VideoCollection.get_media_attributes(@channel_url)
     end
 
-    test "returns the error straight through when the command fails" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:error, "Big issue", 1} end)
+    test "supports an optional file_listener_handler that gets passed a filename" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
+      current_self = self()
 
-      assert {:error, "Big issue", 1} = VideoCollection.get_media_attributes(@channel_url)
+      handler = fn filename ->
+        send(current_self, {:handler, filename})
+      end
+
+      assert {:ok, _} = VideoCollection.get_media_attributes(@channel_url, file_listener_handler: handler)
+
+      assert_receive {:handler, filename}
+      assert String.ends_with?(filename, ".json")
     end
   end
 

test/pinchflat/media_client/source_details_test.exs

@@ -45,7 +45,7 @@ defmodule Pinchflat.MediaClient.SourceDetailsTest do
 
   describe "get_media_attributes/2 when passed a string" do
     test "it passes the expected arguments to the backend" do
-      expect(YtDlpRunnerMock, :run, fn @channel_url, opts, ot ->
+      expect(YtDlpRunnerMock, :run, fn @channel_url, opts, ot, _addl_opts ->
         assert opts == [:simulate, :skip_download]
         assert ot == "%(.{id,title,was_live,original_url,description})j"
 
@@ -56,7 +56,7 @@ defmodule Pinchflat.MediaClient.SourceDetailsTest do
     end
 
     test "it returns a list of maps" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot ->
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
         {:ok, source_attributes_return_fixture()}
       end)
 
@@ -68,7 +68,7 @@ defmodule Pinchflat.MediaClient.SourceDetailsTest do
     test "it calls the backend with the source's collection ID" do
       source = source_fixture()
 
-      expect(YtDlpRunnerMock, :run, fn url, _opts, _ot ->
+      expect(YtDlpRunnerMock, :run, fn url, _opts, _ot, _addl_opts ->
         assert source.collection_id == url
         {:ok, source_attributes_return_fixture()}
       end)
@@ -77,7 +77,7 @@ defmodule Pinchflat.MediaClient.SourceDetailsTest do
     end
 
     test "it builds options based on the source's media profile" do
-      expect(YtDlpRunnerMock, :run, fn _url, opts, _ot ->
+      expect(YtDlpRunnerMock, :run, fn _url, opts, _ot, _addl_opts ->
         assert opts == [:simulate, :skip_download]
         {:ok, ""}
       end)
@@ -91,5 +91,22 @@ defmodule Pinchflat.MediaClient.SourceDetailsTest do
       source = source_fixture(media_profile_id: media_profile.id)
       assert {:ok, _} = SourceDetails.get_media_attributes(source)
     end
+
+    test "lets you pass through an optional file_listener_handler" do
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+        {:ok, source_attributes_return_fixture()}
+      end)
+
+      source = source_fixture()
+      current_self = self()
+
+      handler = fn filename ->
+        send(current_self, {:handler, filename})
+      end
+
+      assert {:ok, _} = SourceDetails.get_media_attributes(source, file_listener_handler: handler)
+
+      assert_receive {:handler, _}
+    end
   end
 end

test/pinchflat/media_test.exs

@@ -215,6 +215,34 @@ defmodule Pinchflat.MediaTest do
     end
   end
 
+  describe "pending_download?/1" do
+    test "returns true when the media hasn't been downloaded" do
+      media_item = media_item_fixture(%{media_filepath: nil})
+
+      assert Media.pending_download?(media_item)
+    end
+
+    test "returns false if the media has been downloaded" do
+      media_item = media_item_fixture(%{media_filepath: "/video/#{Faker.File.file_name(:video)}"})
+
+      refute Media.pending_download?(media_item)
+    end
+
+    test "returns false if the media hasn't been downloaded but the profile doesn't DL shorts" do
+      source = source_fixture(%{media_profile_id: media_profile_fixture(%{shorts_behaviour: :exclude}).id})
+      media_item = media_item_fixture(%{source_id: source.id, media_filepath: nil, original_url: "/shorts/"})
+
+      refute Media.pending_download?(media_item)
+    end
+
+    test "returns false if the media hasn't been downloaded but the profile doesn't DL livestreams" do
+      source = source_fixture(%{media_profile_id: media_profile_fixture(%{livestream_behaviour: :exclude}).id})
+      media_item = media_item_fixture(%{source_id: source.id, media_filepath: nil, livestream: true})
+
+      refute Media.pending_download?(media_item)
+    end
+  end
+
   describe "search/1" do
     setup do
       media_item =

test/pinchflat/tasks/source_tasks_test.exs

@@ -5,6 +5,7 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
   import Pinchflat.TasksFixtures
   import Pinchflat.MediaFixtures
   import Pinchflat.SourcesFixtures
+  import Pinchflat.ProfilesFixtures
 
   alias Pinchflat.Tasks
   alias Pinchflat.Tasks.Task
@@ -43,15 +44,17 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
     end
   end
 
-  describe "index_media_items/1" do
+  describe "index_and_enqueue_download_for_media_items/1" do
     setup do
-      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, source_attributes_return_fixture()} end)
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+        {:ok, source_attributes_return_fixture()}
+      end)
 
       {:ok, [source: source_fixture()]}
     end
 
     test "it creates a media_item record for each media ID returned", %{source: source} do
-      assert media_items = SourceTasks.index_media_items(source)
+      assert media_items = SourceTasks.index_and_enqueue_download_for_media_items(source)
 
       assert Enum.count(media_items) == 3
       assert ["video1", "video2", "video3"] == Enum.map(media_items, & &1.media_id)
@@ -63,15 +66,15 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
 
     test "it attaches all media_items to the given source", %{source: source} do
       source_id = source.id
-      assert media_items = SourceTasks.index_media_items(source)
+      assert media_items = SourceTasks.index_and_enqueue_download_for_media_items(source)
 
       assert Enum.count(media_items) == 3
       assert Enum.all?(media_items, fn %MediaItem{source_id: ^source_id} -> true end)
     end
 
     test "it won't duplicate media_items based on media_id and source", %{source: source} do
-      _first_run = SourceTasks.index_media_items(source)
-      _duplicate_run = SourceTasks.index_media_items(source)
+      _first_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
+      _duplicate_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
 
       media_items = Repo.preload(source, :media_items).media_items
       assert Enum.count(media_items) == 3
@@ -80,8 +83,8 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
     test "it can duplicate media_ids for different sources", %{source: source} do
       other_source = source_fixture()
 
-      media_items = SourceTasks.index_media_items(source)
-      media_items_other_source = SourceTasks.index_media_items(other_source)
+      media_items = SourceTasks.index_and_enqueue_download_for_media_items(source)
+      media_items_other_source = SourceTasks.index_and_enqueue_download_for_media_items(other_source)
 
       assert Enum.count(media_items) == 3
       assert Enum.count(media_items_other_source) == 3
@@ -91,8 +94,8 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
     end
 
     test "it returns a list of media_items or changesets", %{source: source} do
-      first_run = SourceTasks.index_media_items(source)
-      duplicate_run = SourceTasks.index_media_items(source)
+      first_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
+      duplicate_run = SourceTasks.index_and_enqueue_download_for_media_items(source)
 
       assert Enum.all?(first_run, fn %MediaItem{} -> true end)
       assert Enum.all?(duplicate_run, fn %Ecto.Changeset{} -> true end)
@@ -101,11 +104,122 @@ defmodule Pinchflat.Tasks.SourceTasksTest do
     test "it updates the source's last_indexed_at field", %{source: source} do
       assert source.last_indexed_at == nil
 
-      SourceTasks.index_media_items(source)
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
       source = Repo.reload!(source)
 
       assert DateTime.diff(DateTime.utc_now(), source.last_indexed_at) < 2
     end
+
+    test "it enqueues a job for each pending media item" do
+      source = source_fixture()
+      media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
+
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
+
+      assert_enqueued(worker: VideoDownloadWorker, args: %{"id" => media_item.id})
+    end
+
+    test "it does not attach tasks if the source is set to not download" do
+      source = source_fixture(download_media: false)
+      media_item = media_item_fixture(source_id: source.id, media_filepath: nil)
+
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
+
+      assert [] = Tasks.list_tasks_for(:media_item_id, media_item.id)
+    end
+  end
+
+  describe "index_and_enqueue_download_for_media_items/1 when testing file watcher" do
+    setup do
+      {:ok, [source: source_fixture()]}
+    end
+
+    test "creates a new media item for everything already in the file", %{source: source} do
+      watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval)
+
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts ->
+        filepath = Keyword.get(addl_opts, :output_filepath)
+        File.write(filepath, source_attributes_return_fixture())
+
+        # Need to add a delay to ensure the file watcher has time to read the file
+        :timer.sleep(watcher_poll_interval * 2)
+        # We know we're testing the file watcher since the syncronous call will only
+        # return an empty string (creating no records)
+        {:ok, ""}
+      end)
+
+      assert Repo.aggregate(MediaItem, :count, :id) == 0
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
+      assert Repo.aggregate(MediaItem, :count, :id) == 3
+    end
+
+    test "enqueues a download for everything already in the file", %{source: source} do
+      watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval)
+
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts ->
+        filepath = Keyword.get(addl_opts, :output_filepath)
+        File.write(filepath, source_attributes_return_fixture())
+
+        # Need to add a delay to ensure the file watcher has time to read the file
+        :timer.sleep(watcher_poll_interval * 2)
+        # We know we're testing the file watcher since the syncronous call will only
+        # return an empty string (creating no records)
+        {:ok, ""}
+      end)
+
+      refute_enqueued(worker: VideoDownloadWorker)
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
+      assert_enqueued(worker: VideoDownloadWorker)
+    end
+
+    test "does not enqueue downloads if the source is set to not download" do
+      watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval)
+      source = source_fixture(download_media: false)
+
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts ->
+        filepath = Keyword.get(addl_opts, :output_filepath)
+        File.write(filepath, source_attributes_return_fixture())
+
+        # Need to add a delay to ensure the file watcher has time to read the file
+        :timer.sleep(watcher_poll_interval * 2)
+        # We know we're testing the file watcher since the syncronous call will only
+        # return an empty string (creating no records)
+        {:ok, ""}
+      end)
+
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
+      refute_enqueued(worker: VideoDownloadWorker)
+    end
+
+    test "does not enqueue downloads for media that doesn't match the profile's format options" do
+      watcher_poll_interval = Application.get_env(:pinchflat, :file_watcher_poll_interval)
+      profile = media_profile_fixture(%{shorts_behaviour: :exclude})
+      source = source_fixture(%{media_profile_id: profile.id})
+
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, addl_opts ->
+        filepath = Keyword.get(addl_opts, :output_filepath)
+
+        contents =
+          Phoenix.json_library().encode!(%{
+            id: "video2",
+            title: "Video 2",
+            original_url: "https://example.com/shorts/video2",
+            was_live: true,
+            description: "desc2"
+          })
+
+        File.write(filepath, contents)
+
+        # Need to add a delay to ensure the file watcher has time to read the file
+        :timer.sleep(watcher_poll_interval * 2)
+        # We know we're testing the file watcher since the syncronous call will only
+        # return an empty string (creating no records)
+        {:ok, ""}
+      end)
+
+      SourceTasks.index_and_enqueue_download_for_media_items(source)
+      refute_enqueued(worker: VideoDownloadWorker)
+    end
   end
 
   describe "enqueue_pending_media_tasks/1" do

test/pinchflat/utils/filesystem_utils/file_follower_server_test.exs

@@ -0,0 +1,52 @@
+defmodule Pinchflat.Utils.FilesystemUtils.FileFollowerServerTest do
+  use ExUnit.Case, async: true
+
+  alias alias Pinchflat.Utils.FilesystemUtils
+  alias Pinchflat.Utils.FilesystemUtils.FileFollowerServer
+
+  setup do
+    {:ok, pid} = FileFollowerServer.start_link()
+    tmpfile = FilesystemUtils.generate_metadata_tmpfile(:txt)
+
+    {:ok, %{pid: pid, tmpfile: tmpfile}}
+  end
+
+  describe "watch_file" do
+    test "calls the handler for each existing line in the file", %{pid: pid, tmpfile: tmpfile} do
+      File.write!(tmpfile, "line1\nline2")
+      parent = self()
+
+      handler = fn line -> send(parent, line) end
+      FileFollowerServer.watch_file(pid, tmpfile, handler)
+
+      assert_receive "line1\n"
+      assert_receive "line2"
+    end
+
+    test "calls the handler for each new line in the file", %{pid: pid, tmpfile: tmpfile} do
+      parent = self()
+      file = File.open!(tmpfile, [:append])
+      handler = fn line -> send(parent, line) end
+
+      FileFollowerServer.watch_file(pid, tmpfile, handler)
+
+      IO.binwrite(file, "line1\n")
+      assert_receive "line1\n"
+      IO.binwrite(file, "line2")
+      assert_receive "line2"
+    end
+  end
+
+  describe "stop" do
+    test "stops the watcher", %{pid: pid, tmpfile: tmpfile} do
+      handler = fn _line -> :noop end
+      FileFollowerServer.watch_file(pid, tmpfile, handler)
+
+      refute is_nil(Process.info(pid))
+      FileFollowerServer.stop(pid)
+      # Gotta wait for the server to stop async
+      :timer.sleep(10)
+      assert is_nil(Process.info(pid))
+    end
+  end
+end

test/pinchflat/utils/filesystem_utils_test.exs

@@ -0,0 +1,16 @@
+defmodule Pinchflat.Utils.FilesystemUtilsTest do
+  use ExUnit.Case, async: true
+
+  alias Pinchflat.Utils.FilesystemUtils
+
+  describe "generate_metadata_tmpfile/1" do
+    test "creates a tmpfile and returns its path" do
+      res = FilesystemUtils.generate_metadata_tmpfile(:json)
+
+      assert String.ends_with?(res, ".json")
+      assert File.exists?(res)
+
+      File.rm!(res)
+    end
+  end
+end

test/pinchflat/workers/media_indexing_worker_test.exs

@@ -13,7 +13,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
 
   describe "perform/1" do
     test "it indexes the source if it should be indexed" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: 10)
 
@@ -21,7 +21,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it indexes the source no matter what if the source has never been indexed before" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: 0, last_indexed_at: nil)
 
@@ -29,7 +29,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it does not do any indexing if the source has been indexed and shouldn't be rescheduled" do
-      expect(YtDlpRunnerMock, :run, 0, fn _url, _opts, _ot -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, 0, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: -1, last_indexed_at: DateTime.utc_now())
 
@@ -37,7 +37,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it does not reschedule if the source shouldn't be indexed" do
-      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
+      stub(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: -1)
       perform_job(MediaIndexingWorker, %{id: source.id})
@@ -46,7 +46,9 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it kicks off a download job for each pending media item" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, source_attributes_return_fixture()} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+        {:ok, source_attributes_return_fixture()}
+      end)
 
       source = source_fixture(index_frequency_minutes: 10)
       perform_job(MediaIndexingWorker, %{id: source.id})
@@ -55,7 +57,9 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it starts a job for any pending media item even if it's from another run" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, source_attributes_return_fixture()} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+        {:ok, source_attributes_return_fixture()}
+      end)
 
       source = source_fixture(index_frequency_minutes: 10)
       media_item_fixture(%{source_id: source.id, media_filepath: nil})
@@ -65,7 +69,9 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it does not kick off a job for media items that could not be saved" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, source_attributes_return_fixture()} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts ->
+        {:ok, source_attributes_return_fixture()}
+      end)
 
       source = source_fixture(index_frequency_minutes: 10)
       media_item_fixture(%{source_id: source.id, media_filepath: nil, media_id: "video1"})
@@ -76,7 +82,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it reschedules the job based on the index frequency" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: 10)
       perform_job(MediaIndexingWorker, %{id: source.id})
@@ -89,7 +95,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it creates a task for the rescheduled job" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, ""} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, ""} end)
 
       source = source_fixture(index_frequency_minutes: 10)
       task_count_fetcher = fn -> Enum.count(Tasks.list_tasks()) end
@@ -100,7 +106,7 @@ defmodule Pinchflat.Workers.MediaIndexingWorkerTest do
     end
 
     test "it creates the basic media_item records" do
-      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot -> {:ok, source_attributes_return_fixture()} end)
+      expect(YtDlpRunnerMock, :run, fn _url, _opts, _ot, _addl_opts -> {:ok, source_attributes_return_fixture()} end)
 
       source = source_fixture(index_frequency_minutes: 10)