diff --git a/config/test.exs b/config/test.exs index 6243114e..1bd74983 100644 --- a/config/test.exs +++ b/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 diff --git a/lib/microwaveprop/weather/mrms_client.ex b/lib/microwaveprop/weather/mrms_client.ex index c36244d2..0780bdca 100644 --- a/lib/microwaveprop/weather/mrms_client.ex +++ b/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 diff --git a/test/microwaveprop/weather/mrms_cache_test.exs b/test/microwaveprop/weather/mrms_cache_test.exs new file mode 100644 index 00000000..905d9106 --- /dev/null +++ b/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 diff --git a/test/microwaveprop/workers/mrms_fetch_worker_test.exs b/test/microwaveprop/workers/mrms_fetch_worker_test.exs new file mode 100644 index 00000000..c1cb4951 --- /dev/null +++ b/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, "
no files here") + 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 = """ + f + """ + + 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 = """ + f + """ + + 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 = """ + f + """ + + 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