test: more reliable tests & test async cancel with http1 pool

lib/finch/http1/pool.ex

@@ -103,7 +103,9 @@ defmodule Finch.HTTP1.Pool do
   end
 
   @impl Finch.Pool
-  def cancel_async_request(_request_ref) do
+  def cancel_async_request({_, _, _, pid}) do
+    Process.exit(pid, :shutdown)
+    :ok
   end
 
   @impl NimblePool

test/finch/http1/integration_proxy_test.exs

@@ -4,7 +4,7 @@ defmodule Finch.HTTP1.IntegrationProxyTest do
   alias Finch.HTTP1Server
 
   setup_all do
-    port = 4002
+    port = 4004
 
     # Not quite a proper forward proxy server, but good enough
     {:ok, _} = HTTP1Server.start(port)

test/finch/http1/pool_test.exs

@@ -1,6 +1,8 @@
 defmodule Finch.HTTP1.PoolTest do
   use FinchCase, async: true
 
+  alias Finch.HTTP1Server
+
   @tag capture_log: true
   test "should terminate pool after idle timeout", %{bypass: bypass, finch_name: finch_name} do
     test_name = to_string(finch_name)
@@ -47,4 +49,32 @@ defmodule Finch.HTTP1.PoolTest do
 
     :telemetry.detach(test_name)
   end
+
+  describe "async requests" do
+    @describetag bypass: false
+
+    setup do
+      port = 4005
+      endpoint = "http://localhost:#{port}"
+
+      start_supervised!({HTTP1Server, port: port})
+
+      {:ok, endpoint: endpoint}
+    end
+
+    test "can be canceled with cancel_async_request/1", %{
+      finch_name: finch_name,
+      endpoint: endpoint
+    } do
+      start_supervised!({Finch, name: finch_name, pools: %{default: [protocol: :http1]}})
+
+      ref =
+        Finch.build(:get, endpoint <> "/stream/1/50")
+        |> Finch.async_request(finch_name)
+
+      assert_receive {^ref, {:status, 200}}
+      Finch.HTTP1.Pool.cancel_async_request(ref)
+      refute_receive {^ref, {:data, _}}
+    end
+  end
 end

test/finch_test.exs

@@ -767,32 +767,6 @@ defmodule FinchTest do
       for _ <- 1..5, do: assert_receive({^request_ref, {:data, "chunk-data"}})
       assert_receive {^request_ref, :done}
     end
-
-    @tag :skip
-    test "can be canceled with cancel_async_request/1", %{bypass: bypass, finch_name: finch_name} do
-      start_supervised!({Finch, name: finch_name})
-
-      Bypass.expect(bypass, fn conn ->
-        conn = Plug.Conn.send_chunked(conn, 200)
-
-        :timer.sleep(10)
-
-        Enum.reduce(1..5, conn, fn _, conn ->
-          {:ok, conn} = Plug.Conn.chunk(conn, "chunk-data")
-          conn
-        end)
-      end)
-
-      request_ref =
-        Finch.build(:get, endpoint(bypass))
-        |> Finch.async_request(finch_name)
-
-      assert_receive {^request_ref, {:status, 200}}, 10
-
-      Finch.cancel_async_request(request_ref)
-
-      refute_receive {^request_ref, {:data, _}}
-    end
   end
 
   defp get_pools(name, shp) do

test/support/finch_case.ex

@@ -8,8 +8,14 @@ defmodule FinchCase do
     end
   end
 
-  setup %{test: finch_name} do
-    {:ok, bypass: Bypass.open(), finch_name: finch_name}
+  setup context do
+    bypass =
+      case context do
+        %{bypass: false} -> []
+        %{} -> [bypass: Bypass.open()]
+      end
+
+    {:ok, bypass ++ [finch_name: context.test]}
   end
 
   @doc "Returns the url for a Bypass instance"

test/support/http1_server.ex

@@ -1,23 +1,23 @@
 defmodule Finch.HTTP1Server do
   @moduledoc false
 
-  def start(port) do
-    children = [
-      Plug.Adapters.Cowboy.child_spec(
-        scheme: :http,
-        plug: Finch.HTTP1Server.PlugRouter,
-        options: [
-          port: port,
-          otp_app: :finch,
-          protocol_options: [
-            idle_timeout: 3_000,
-            request_timeout: 10_000
-          ]
+  def child_spec(opts) do
+    Plug.Adapters.Cowboy.child_spec(
+      scheme: :http,
+      plug: Finch.HTTP1Server.PlugRouter,
+      options: [
+        port: Keyword.fetch!(opts, :port),
+        otp_app: :finch,
+        protocol_options: [
+          idle_timeout: 3_000,
+          request_timeout: 10_000
         ]
-      )
-    ]
+      ]
+    )
+  end
 
-    Supervisor.start_link(children, strategy: :one_for_one)
+  def start(port) do
+    Supervisor.start_link([child_spec(port: port)], strategy: :one_for_one)
   end
 end
 
@@ -37,4 +37,22 @@ defmodule Finch.HTTP1Server.PlugRouter do
     |> send_resp(200, "Hello #{name}!")
     |> halt()
   end
+
+  get "/wait/:delay" do
+    delay = conn.params["delay"] |> String.to_integer()
+    Process.sleep(delay)
+    send_resp(conn, 200, "ok")
+  end
+
+  get "/stream/:num/:delay" do
+    num = conn.params["num"] |> String.to_integer()
+    delay = conn.params["delay"] |> String.to_integer()
+    conn = send_chunked(conn, 200)
+
+    Enum.reduce(1..num, conn, fn i, conn ->
+      Process.sleep(delay)
+      {:ok, conn} = chunk(conn, "chunk-#{i}\n")
+      conn
+    end)
+  end
 end