Check process ancestry tree for test mode override

Cascade the `with_testing_mode` block to nested processes that make use
of `:$ancestry` in the process dictionary, i.e. tasks. Now enqueuing a
job within spawned processes like `Task.async` or `Task.async_stream`
will honor the testing mode specified in `with_testing_mode/2`.

Closes #1067

lib/oban/config.ex

@@ -151,14 +151,24 @@ defmodule Oban.Config do
 
   @doc false
   @spec get_engine(t()) :: module()
+  def get_engine(%__MODULE__{engine: engine, testing: :disabled}), do: engine
+
   def get_engine(%__MODULE__{engine: engine, testing: testing}) do
-    if Process.get(:oban_testing, testing) == :inline do
+    pids = [self() | Process.get(:"$ancestors", [])]
+
+    if Enum.any?(pids, &inline_testing?(&1, testing)) do
       Oban.Engines.Inline
     else
       engine
     end
   end
 
+  defp inline_testing?(pid, default) do
+    {:dictionary, dictionary} = Process.info(pid, :dictionary)
+
+    Keyword.get(dictionary, :oban_testing, default) == :inline
+  end
+
   @doc false
   @spec node_name(%{optional(binary()) => binary()}) :: binary()
   def node_name(env \\ System.get_env()) do

test/oban/testing_test.exs

@@ -396,7 +396,23 @@ defmodule Oban.TestingTest do
 
         assert_enqueued worker: Worker
 
-        refute_received {:ok, 1}
+        refute_received {:ok, 1}, 25
+      end)
+    end
+
+    test "the temporary mode cascades down to child processes" do
+      name = start_supervised_oban!(testing: :manual)
+
+      Testing.with_testing_mode(:inline, fn ->
+        fun = fn ->
+          Oban.insert!(name, Worker.new(%{ref: 1, action: "OK"}))
+
+          assert_received {:ok, 1}
+        end
+
+        fun
+        |> Task.async()
+        |> Task.await()
       end)
     end