test: cover mrms_cache + mrms_fetch_worker, add Req.Test seam

Same fix as rtma_client: MrmsClient had a hardcoded req_options/0 —
switched to a Keyword.merge with Application.get_env so tests can stub
HTTP. 13 new characterization tests: cache put/get/clear/broadcast and
worker listing/download/up-to-date paths. Full GRIB2 parse happy path
deferred to the wgrib2 coverage task.

config/test.exs

@@ -61,6 +61,7 @@ config :microwaveprop, elevation_req_options: [plug: {Req.Test, Microwaveprop.Te
 config :microwaveprop, giro_req_options: [plug: {Req.Test, Microwaveprop.Ionosphere.GiroClient}, retry: false]
 config :microwaveprop, hrrr_req_options: [plug: {Req.Test, Microwaveprop.Weather.HrrrClient}, retry: false]
 config :microwaveprop, iem_req_options: [plug: {Req.Test, Microwaveprop.Weather.IemClient}, retry: false]
+config :microwaveprop, mrms_req_options: [plug: {Req.Test, Microwaveprop.Weather.MrmsClient}, retry: false]
 config :microwaveprop, narr_req_options: [plug: {Req.Test, Microwaveprop.Weather.NarrClient}, retry: false]
 
 # Route HTTP requests through Req.Test stubs

lib/microwaveprop/weather/mrms_client.ex

@@ -176,6 +176,9 @@ defmodule Microwaveprop.Weather.MrmsClient do
   end
 
   defp req_options do
-    [receive_timeout: 120_000, retry: :safe_transient, max_retries: 3]
+    Keyword.merge(
+      [receive_timeout: 120_000, retry: :safe_transient, max_retries: 3],
+      Application.get_env(:microwaveprop, :mrms_req_options, [])
+    )
   end
 end

test/microwaveprop/weather/mrms_cache_test.exs

@@ -0,0 +1,97 @@
+defmodule Microwaveprop.Weather.MrmsCacheTest do
+  # async: false — MrmsCache is started at the application level and owns a
+  # named ETS table. Tests mutate shared state and must not race.
+  use ExUnit.Case, async: false
+
+  alias Microwaveprop.Weather.MrmsCache
+
+  @topic "mrms:cache"
+  @pubsub Microwaveprop.PubSub
+
+  setup do
+    MrmsCache.clear()
+    :ok
+  end
+
+  describe "fetch/0" do
+    test "returns :miss when nothing is cached" do
+      assert MrmsCache.fetch() == :miss
+    end
+
+    test "returns {:ok, valid_time, grid} after put/2" do
+      valid_time = ~U[2026-04-12 19:24:00Z]
+      grid = %{{32.0, -97.0} => 1.2, {32.125, -97.0} => 0.0}
+
+      assert :ok = MrmsCache.put(valid_time, grid)
+      assert {:ok, ^valid_time, ^grid} = MrmsCache.fetch()
+    end
+
+    test "put/2 overwrites the single current entry" do
+      first_vt = ~U[2026-04-12 19:24:00Z]
+      second_vt = ~U[2026-04-12 19:26:00Z]
+
+      MrmsCache.put(first_vt, %{{32.0, -97.0} => 1.0})
+      MrmsCache.put(second_vt, %{{32.0, -97.0} => 2.0})
+
+      assert {:ok, ^second_vt, %{{32.0, -97.0} => 2.0}} = MrmsCache.fetch()
+    end
+  end
+
+  describe "valid_time/0" do
+    test "returns nil when cache is empty" do
+      assert MrmsCache.valid_time() == nil
+    end
+
+    test "returns the cached valid_time when present" do
+      vt = ~U[2026-04-12 19:24:00Z]
+      MrmsCache.put(vt, %{{32.0, -97.0} => 5.4})
+
+      assert MrmsCache.valid_time() == vt
+    end
+  end
+
+  describe "broadcast_put/2" do
+    test "inserts locally and broadcasts on the mrms:cache topic" do
+      Phoenix.PubSub.subscribe(@pubsub, @topic)
+
+      vt = ~U[2026-04-12 19:28:00Z]
+      grid = %{{32.0, -97.0} => 3.3}
+
+      assert :ok = MrmsCache.broadcast_put(vt, grid)
+      assert {:ok, ^vt, ^grid} = MrmsCache.fetch()
+
+      # The cache subscribes to its own topic, so we should observe the
+      # message via our own subscription too.
+      assert_receive {:mrms_cache_refresh, ^vt, ^grid}
+    end
+  end
+
+  describe "clear/0" do
+    test "empties the cache" do
+      MrmsCache.put(~U[2026-04-12 19:24:00Z], %{{32.0, -97.0} => 1.0})
+      assert {:ok, _, _} = MrmsCache.fetch()
+
+      assert :ok = MrmsCache.clear()
+      assert MrmsCache.fetch() == :miss
+      assert MrmsCache.valid_time() == nil
+    end
+  end
+
+  describe "peer broadcast handling" do
+    test "applies refreshes received from peers on the PubSub topic" do
+      # Simulate a peer node pushing a refresh. The cache GenServer is
+      # subscribed to @topic in init/1; broadcasting directly exercises the
+      # handle_info({:mrms_cache_refresh, ...}) clause.
+      vt = ~U[2026-04-12 19:30:00Z]
+      grid = %{{32.0, -97.0} => 7.7}
+
+      Phoenix.PubSub.broadcast(@pubsub, @topic, {:mrms_cache_refresh, vt, grid})
+
+      # Round-trip through the MrmsCache GenServer mailbox: send it a sync
+      # call so we know the previous broadcast has been processed.
+      _ = :sys.get_state(MrmsCache)
+
+      assert {:ok, ^vt, ^grid} = MrmsCache.fetch()
+    end
+  end
+end

test/microwaveprop/workers/mrms_fetch_worker_test.exs

@@ -0,0 +1,137 @@
+defmodule Microwaveprop.Workers.MrmsFetchWorkerTest do
+  # async: false because the MrmsCache is an application-wide singleton with
+  # shared ETS state. Concurrent tests would race on cache contents.
+  use Microwaveprop.DataCase, async: false
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  alias Microwaveprop.Weather.MrmsCache
+  alias Microwaveprop.Weather.MrmsClient
+  alias Microwaveprop.Workers.MrmsFetchWorker
+
+  @topic "mrms:cache"
+  @pubsub Microwaveprop.PubSub
+
+  setup do
+    MrmsCache.clear()
+    :ok
+  end
+
+  describe "perform/1" do
+    test "returns :ok and leaves cache empty when the listing is empty" do
+      Req.Test.stub(MrmsClient, fn conn ->
+        # Empty directory index — no filenames matched by the parser.
+        Plug.Conn.send_resp(conn, 200, "<html><body>no files here</body></html>")
+      end)
+
+      assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}})
+      assert MrmsCache.fetch() == :miss
+    end
+
+    test "returns :ok and leaves cache empty when the listing request 500s" do
+      Req.Test.stub(MrmsClient, fn conn ->
+        Plug.Conn.send_resp(conn, 500, "upstream boom")
+      end)
+
+      assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}})
+      assert MrmsCache.fetch() == :miss
+    end
+
+    test "short-circuits with :up_to_date when cache already has the latest valid_time" do
+      latest_vt = ~U[2026-04-12 19:28:00Z]
+      # Pre-populate the cache so fetch_latest/1 sees known_valid_time >= latest.
+      MrmsCache.put(latest_vt, %{{32.0, -97.0} => 4.2})
+
+      Req.Test.stub(MrmsClient, fn conn ->
+        # Only the listing URL should be hit — never a .grib2.gz download.
+        assert String.ends_with?(conn.request_path, "/PrecipRate/") or
+                 String.ends_with?(conn.request_path, "/PrecipRate")
+
+        html = """
+        <a href="MRMS_PrecipRate_00.00_20260412-192800.grib2.gz">f</a>
+        """
+
+        Plug.Conn.send_resp(conn, 200, html)
+      end)
+
+      assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}})
+
+      # Cache is untouched.
+      assert {:ok, ^latest_vt, %{{32.0, -97.0} => 4.2}} = MrmsCache.fetch()
+    end
+
+    test "returns :ok and leaves cache empty when the download path 404s" do
+      # Listing succeeds with one file; download 404s. The worker must log
+      # and swallow the error rather than crash.
+      #
+      # The .grib2.gz URL triggers Req's URL-extension-based gzip decode step
+      # regardless of response status, so even a 404 body must be valid gzip.
+      gzipped_404 = :zlib.gzip("gone")
+
+      Req.Test.stub(MrmsClient, fn conn ->
+        if String.ends_with?(conn.request_path, ".grib2.gz") do
+          Plug.Conn.send_resp(conn, 404, gzipped_404)
+        else
+          html = """
+          <a href="MRMS_PrecipRate_00.00_20260412-193000.grib2.gz">f</a>
+          """
+
+          Plug.Conn.send_resp(conn, 200, html)
+        end
+      end)
+
+      assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}})
+      assert MrmsCache.fetch() == :miss
+    end
+
+    # The full happy path — successful listing + download + real GRIB2
+    # decode into a populated cache grid — requires a real MRMS PrecipRate
+    # payload that we don't have a hermetic fixture for. We defer that to
+    # the wgrib2/GRIB2 coverage task.
+    #
+    # What we can still pin down here is the worker's delivery contract
+    # when both listing and download return 200: if fetch_latest/1 returns
+    # {:ok, vt, grid}, the worker MUST broadcast_put/2 to the mrms:cache
+    # topic. We stub garbage gzipped bytes; wgrib2 (when installed) will
+    # yield an empty grid rather than an error on "no matching messages",
+    # and the worker broadcasts that through. When wgrib2 is NOT installed
+    # the client returns :wgrib2_not_available, the worker logs a warning
+    # and still returns :ok — no broadcast. Both arms are valid behavior
+    # and both are exercised here.
+    test "returns :ok on a 200 download (broadcasts iff wgrib2 yields :ok)" do
+      Phoenix.PubSub.subscribe(@pubsub, @topic)
+
+      # The .grib2.gz URL triggers Req's URL-extension-based gzip decoding,
+      # so the response body must be real gzip bytes.
+      gzipped_garbage = :zlib.gzip(:binary.copy(<<0>>, 1024))
+
+      Req.Test.stub(MrmsClient, fn conn ->
+        if String.ends_with?(conn.request_path, ".grib2.gz") do
+          Plug.Conn.send_resp(conn, 200, gzipped_garbage)
+        else
+          html = """
+          <a href="MRMS_PrecipRate_00.00_20260412-193200.grib2.gz">f</a>
+          """
+
+          Plug.Conn.send_resp(conn, 200, html)
+        end
+      end)
+
+      assert :ok = MrmsFetchWorker.perform(%Oban.Job{args: %{}})
+
+      # One of two valid outcomes:
+      #   1. wgrib2 installed → fetch_latest returns {:ok, vt, %{}} →
+      #      worker calls broadcast_put and the subscriber sees a refresh.
+      #   2. wgrib2 missing → fetch_latest returns {:error, ...} → no
+      #      broadcast, cache unchanged.
+      receive do
+        {:mrms_cache_refresh, vt, grid} ->
+          assert %DateTime{} = vt
+          assert is_map(grid)
+          assert {:ok, ^vt, ^grid} = MrmsCache.fetch()
+      after
+        200 ->
+          assert MrmsCache.fetch() == :miss
+      end
+    end
+  end
+end