diff --git a/test/microwaveprop/workers/nexrad_worker_test.exs b/test/microwaveprop/workers/nexrad_worker_test.exs new file mode 100644 index 00000000..bca0a92f --- /dev/null +++ b/test/microwaveprop/workers/nexrad_worker_test.exs @@ -0,0 +1,302 @@ +defmodule Microwaveprop.Workers.NexradWorkerTest do + use Microwaveprop.DataCase, async: false + use Oban.Testing, repo: Microwaveprop.Repo + + alias Microwaveprop.Radio.Contact + alias Microwaveprop.Weather.NexradClient + alias Microwaveprop.Weather.NexradObservation + alias Microwaveprop.Workers.NexradWorker + + # Build a minimal valid indexed-color PNG (20x20, all-zero pixels). + # The worker's PNG decoder only needs: + # - 8-byte signature + # - IHDR (width, height, bit_depth=8, color_type=3) + # - one or more IDAT chunks (zlib-compressed filtered scanlines) + # - IEND + # CRCs are read but not validated. Pixel value 0 = no echo everywhere, so + # extract_box returns pixel_count: 0 for every point. That's fine: these + # tests characterize the worker (filtering, upsert, error handling), not + # the PNG decoder (covered in NexradClientTest). + defp minimal_png(width \\ 20, height \\ 20) do + signature = <<137, 80, 78, 71, 13, 10, 26, 10>> + ihdr_data = <> + ihdr_chunk = <> + + # Each scanline: filter byte (0 = None) + width pixel bytes (all 0). + raw = for _ <- 1..height, into: <<>>, do: <<0, 0::size(width)-unit(8)>> + compressed = :zlib.compress(raw) + idat_chunk = <> + + iend_chunk = <<0::32, "IEND", 0::32>> + + signature <> ihdr_chunk <> idat_chunk <> iend_chunk + end + + defp stub_png_success do + png = minimal_png() + + Req.Test.stub(NexradClient, fn conn -> + Plug.Conn.send_resp(conn, 200, png) + end) + end + + defp insert_contact!(attrs) do + defaults = %{ + station1: "W5XD", + station2: "K5TR", + mode: "CW", + band: Decimal.new("10000") + } + + %Contact{} + |> Contact.changeset(Map.merge(defaults, attrs)) + |> Repo.insert!() + end + + describe "perform/1" do + test "fetches one n0q frame and upserts one observation per unique snapped point" do + stub_png_success() + + # Frame at 14:00 UTC. Two contacts inside the +/-30 min window, + # at two distinct positions (snapped to 3-decimal precision). + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:10:00Z], + pos1: %{"lat" => 32.9, "lon" => -97.0} + }) + + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 13:45:00Z], + pos1: %{"lat" => 33.5, "lon" => -96.5} + }) + + args = %{ + "year" => 2022, + "month" => 8, + "day" => 20, + "hour" => 14, + "minute" => 0 + } + + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + + rows = Repo.all(NexradObservation) + assert length(rows) == 2 + + coords = rows |> Enum.map(&{&1.lat, &1.lon}) |> Enum.sort() + assert coords == [{32.9, -97.0}, {33.5, -96.5}] + + for row <- rows do + assert DateTime.compare(row.observed_at, ~U[2022-08-20 14:00:00Z]) == :eq + assert row.pixel_count == 0 + assert row.mean_reflectivity_dbz == 0.0 + assert row.max_reflectivity_dbz == 0.0 + assert row.texture_variance == 0.0 + end + end + + test "re-running the same job upserts in place (no duplicates, updated fields persist)" do + stub_png_success() + + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:00:00Z], + pos1: %{"lat" => 32.9, "lon" => -97.0} + }) + + args = %{ + "year" => 2022, + "month" => 8, + "day" => 20, + "hour" => 14, + "minute" => 0 + } + + # First run inserts. + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + assert [first] = Repo.all(NexradObservation) + + # Seed a stale value on the existing row so we can prove the + # second run overwrites it via the ON CONFLICT clause. + Repo.update_all(NexradObservation, set: [mean_reflectivity_dbz: 42.0, pixel_count: 999]) + + # Second run: same (lat, lon, observed_at) => upsert on the unique + # index, not a duplicate insert. + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + + assert [second] = Repo.all(NexradObservation) + # Unique index kept us at one row. + assert second.id == first.id + # :replace_all_except [:id, :inserted_at] overwrote the stale values. + assert second.mean_reflectivity_dbz == 0.0 + assert second.pixel_count == 0 + # inserted_at preserved, updated_at refreshed. + assert DateTime.compare(second.inserted_at, first.inserted_at) == :eq + end + + test "skips contacts outside the +/- 30 minute window around the frame timestamp" do + stub_png_success() + + # Frame at 14:00. Window is [13:30, 14:30]. + # In-window contact: + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:20:00Z], + pos1: %{"lat" => 32.9, "lon" => -97.0} + }) + + # Out-of-window (too early): + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 13:00:00Z], + pos1: %{"lat" => 40.0, "lon" => -80.0} + }) + + # Out-of-window (too late): + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 15:00:00Z], + pos1: %{"lat" => 41.0, "lon" => -81.0} + }) + + # In-window but pos1 is nil — the query filters it out. + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:15:00Z], + pos1: nil + }) + + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14, "minute" => 0} + + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + + rows = Repo.all(NexradObservation) + assert length(rows) == 1 + assert hd(rows).lat == 32.9 + assert hd(rows).lon == -97.0 + end + + test "returns {:error, _} and inserts nothing when upstream returns 404" do + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:00:00Z], + pos1: %{"lat" => 32.9, "lon" => -97.0} + }) + + Req.Test.stub(NexradClient, fn conn -> + Plug.Conn.send_resp(conn, 404, "not found") + end) + + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14, "minute" => 0} + + assert {:error, reason} = NexradWorker.perform(%Oban.Job{args: args}) + assert reason =~ "HTTP 404" + assert Repo.aggregate(NexradObservation, :count, :id) == 0 + end + + test "returns {:error, _} and inserts nothing when the PNG body is garbage" do + # A 200 response with bytes that fail the PNG signature match. The + # client's decode_png_to_pixels rescues the MatchError and returns + # {:error, "PNG decode failed: ..."}; the worker surfaces that. + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:00:00Z], + pos1: %{"lat" => 32.9, "lon" => -97.0} + }) + + Req.Test.stub(NexradClient, fn conn -> + Plug.Conn.send_resp(conn, 200, "not a png") + end) + + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14, "minute" => 0} + + assert {:error, reason} = NexradWorker.perform(%Oban.Job{args: args}) + assert reason =~ "PNG decode failed" + assert Repo.aggregate(NexradObservation, :count, :id) == 0 + end + + test "returns :ok without contacting upstream when no contacts fall in the window" do + # Stub that fails loudly if touched. + Req.Test.stub(NexradClient, fn _conn -> + raise "NexradClient should not be contacted when POI list is empty" + end) + + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14, "minute" => 0} + + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + assert Repo.aggregate(NexradObservation, :count, :id) == 0 + end + + test "defaults the minute arg to 0 when omitted" do + stub_png_success() + + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:05:00Z], + pos1: %{"lat" => 32.9, "lon" => -97.0} + }) + + # No "minute" key — the worker Map.gets it with default 0. + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14} + + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + + assert [row] = Repo.all(NexradObservation) + assert DateTime.compare(row.observed_at, ~U[2022-08-20 14:00:00Z]) == :eq + end + + test "deduplicates points_of_interest when multiple contacts share a snapped coordinate" do + stub_png_success() + + # Two contacts at coordinates that snap to the same 3-decimal point. + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:10:00Z], + pos1: %{"lat" => 32.9001, "lon" => -97.0002} + }) + + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:20:00Z], + pos1: %{"lat" => 32.9004, "lon" => -97.0001} + }) + + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14, "minute" => 0} + + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + + # Both snap to (32.900, -97.000) → one row, not two. + assert [row] = Repo.all(NexradObservation) + assert row.lat == 32.9 + assert row.lon == -97.0 + end + + test ~s(accepts legacy pos1 maps that use the "lng" key instead of "lon") do + stub_png_success() + + insert_contact!(%{ + qso_timestamp: ~U[2022-08-20 14:10:00Z], + pos1: %{"lat" => 32.9, "lng" => -97.0} + }) + + args = %{"year" => 2022, "month" => 8, "day" => 20, "hour" => 14, "minute" => 0} + + assert :ok = NexradWorker.perform(%Oban.Job{args: args}) + + assert [row] = Repo.all(NexradObservation) + assert row.lat == 32.9 + assert row.lon == -97.0 + end + end + + describe "unique constraint" do + test "duplicate enqueues collapse via Oban unique on {year, month, day, hour, minute}" do + args = %{ + "year" => 2022, + "month" => 8, + "day" => 20, + "hour" => 14, + "minute" => 0 + } + + Oban.Testing.with_testing_mode(:manual, fn -> + {:ok, job1} = Oban.insert(NexradWorker.new(args)) + {:ok, job2} = Oban.insert(NexradWorker.new(args)) + + # Second insert hits the unique constraint; Oban returns the + # existing job instead of inserting a new one. + assert job1.id == job2.id + assert Repo.aggregate(Oban.Job, :count, :id) == 1 + end) + end + end +end