defmodule MiddlewareTest do use ExUnit.Case alias Exq.Middleware.Server, as: Middleware alias Exq.Worker.Server, as: Worker import ExqTestUtil defmodule NoArgWorker do def perform do end end defmodule MissingMethodWorker do end defmodule MyMiddleware do @behaviour Exq.Middleware.Behaviour import Exq.Middleware.Pipeline def before_work(pipeline) do send :middlewaretest, :before_work assign(pipeline, :process_info, 1) end def after_processed_work(pipeline) do send :middlewaretest, :after_processed_work pipeline end def after_failed_work(pipeline) do send :middlewaretest, :after_failed_work pipeline end end defmodule HaltedMiddleware do @behaviour Exq.Middleware.Behaviour import Exq.Middleware.Pipeline def before_work(pipeline) do send :middlewaretest, :before_work_halted halt(pipeline) end def after_processed_work(pipeline) do send :middlewaretest, :after_processed_work_halted halt(pipeline) end def after_failed_work(pipeline) do halt(pipeline) end end defmodule StubServer do use GenServer def handle_cast(_msg, state) do {:noreply, state} end def handle_call(_msg, _from, state) do {:reply, {:ok, state}, state} end end def start_worker({class, args, middleware}) do job = "{ \"queue\": \"default\", \"class\": \"#{class}\", \"args\": #{args}, \"jid\": \"123\" }" work_table = :ets.new(:work_table, [:set, :public]) {:ok, stub_server} = GenServer.start_link(MiddlewareTest.StubServer, []) Worker.start_link(job, stub_server, "default", work_table, stub_server, "exq", "localhost", stub_server, middleware) end setup do Process.register(self, :middlewaretest) {:ok, middleware} = GenServer.start_link(Middleware, []) {:ok, middleware: middleware} end test "calls chain for processed work", %{middleware: middleware} do {:ok, worker} = start_worker({"MiddlewareTest.NoArgWorker", "[]", middleware}) Middleware.push(middleware, Exq.Middleware.Job) Middleware.push(middleware, MyMiddleware) Worker.work(worker) state = :sys.get_state(worker) assert state.pipeline.assigns.process_info == 1 assert_receive :before_work assert_receive :after_processed_work end test "calls chain for failed work", %{middleware: middleware} do {:ok, worker} = start_worker({"MiddlewareTest.MissingMethodWorker", "[]", middleware}) Middleware.push(middleware, Exq.Middleware.Job) Middleware.push(middleware, MyMiddleware) Worker.work(worker) state = :sys.get_state(worker) assert state.pipeline.assigns.process_info == 1 assert_receive :before_work assert_receive :after_failed_work end test "halts middleware execution", %{middleware: middleware} do {:ok, worker} = start_worker({"MiddlewareTest.NoArgWorker", "[]", middleware}) Middleware.push(middleware, Exq.Middleware.Job) Middleware.push(middleware, HaltedMiddleware) Middleware.push(middleware, MyMiddleware) Worker.work(worker) state = :sys.get_state(worker) refute Map.has_key?(state.pipeline.assigns, :process_info) assert_receive :before_work_halted assert_receive :after_processed_work_halted refute_receive :before_work refute_receive :after_processed_work end test "restores default middleware after process kill" do {:ok, _pid} = Exq.start_link chain = [Exq.Middleware.Stats, Exq.Middleware.Job, Exq.Middleware.Manager] assert Middleware.all(Middleware) == chain pid = Process.whereis(Middleware) Process.exit(pid, :kill) wait assert Middleware.all(Middleware) == chain end end