defmodule Microwaveprop.Workers.RadarFrameWorkerTest do use Microwaveprop.DataCase, async: false use Oban.Testing, repo: Microwaveprop.Repo alias Microwaveprop.Radio.Contact alias Microwaveprop.Radio.ContactCommonVolumeRadar alias Microwaveprop.Repo alias Microwaveprop.Weather.NexradClient alias Microwaveprop.Workers.RadarFrameWorker setup do Req.Test.set_req_test_from_context(%{async: false}) :ok end defp insert_contact(attrs) do default = %{ station1: "W5XD", station2: "K5TR", qso_timestamp: ~U[2024-09-15 18:30:00Z], mode: "CW", band: Decimal.new("10000"), grid1: "EM12", grid2: "EM00", pos1: %{"lat" => 32.9, "lon" => -97.0}, pos2: %{"lat" => 33.1, "lon" => -96.8}, distance_km: Decimal.new("30") } {:ok, contact} = %Contact{} |> Contact.changeset(Map.merge(default, attrs)) |> Repo.insert() contact end describe "process_frame/4" do test "writes a ContactCommonVolumeRadar row and marks :complete for every contact in the batch" do c1 = insert_contact(%{station1: "A1"}) c2 = insert_contact(%{station1: "A2"}) # In-memory pixel buffer: 10x10 empty frame. The aggregator will # observe zero echoes but still emit a stats row per contact. pixels = :binary.copy(<<0>>, 100) width = 10 frame_ts = ~U[2024-09-15 18:30:00Z] assert :ok = RadarFrameWorker.process_frame( [Repo.reload!(c1), Repo.reload!(c2)], frame_ts, pixels, width ) assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id) assert Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id) assert %Contact{radar_status: :complete} = Repo.get!(Contact, c1.id) assert %Contact{radar_status: :complete} = Repo.get!(Contact, c2.id) end test "re-running process_frame/4 upserts in place (one row per contact, observed_at updated)" do # Pins the on_conflict: {:replace, [...]} clause on contact_id. If a # column gets added to ContactCommonVolumeRadar and the explicit list # isn't updated, or if the list names a non-existent column, this # test fails loudly. contact = insert_contact(%{station1: "RR1"}) pixels = :binary.copy(<<0>>, 100) width = 10 :ok = RadarFrameWorker.process_frame( [Repo.reload!(contact)], ~U[2024-09-15 18:30:00Z], pixels, width ) first = Repo.get_by!(ContactCommonVolumeRadar, contact_id: contact.id) :ok = RadarFrameWorker.process_frame( [Repo.reload!(contact)], ~U[2024-09-15 18:35:00Z], pixels, width ) rows = Repo.all(from(r in ContactCommonVolumeRadar, where: r.contact_id == ^contact.id)) assert length(rows) == 1 [second] = rows assert second.id == first.id assert second.observed_at == ~U[2024-09-15 18:35:00Z] end end describe "perform/1" do test "marks every contact in the batch :unavailable when the frame 404s" do c1 = insert_contact(%{station1: "B1"}) c2 = insert_contact(%{station1: "B2"}) Req.Test.stub(NexradClient, fn conn -> Plug.Conn.send_resp(conn, 404, "not found") end) assert :ok = perform_job(RadarFrameWorker, %{ "frame_ts" => "2024-09-15T18:30:00Z", "contact_ids" => [c1.id, c2.id] }) refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c1.id) refute Repo.get_by(ContactCommonVolumeRadar, contact_id: c2.id) assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c1.id) assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c2.id) end test "returns {:error, _} and leaves contacts :queued on a 5xx server error (so Oban retries)" do c1 = insert_contact(%{station1: "S1"}) c2 = insert_contact(%{station1: "S2"}) Req.Test.stub(NexradClient, fn conn -> Plug.Conn.send_resp(conn, 503, "service unavailable") end) assert {:error, _reason} = perform_job(RadarFrameWorker, %{ "frame_ts" => "2024-09-15T18:30:00Z", "contact_ids" => [c1.id, c2.id] }) # Neither contact should be pinned :unavailable — a retry might succeed. assert %Contact{radar_status: status1} = Repo.get!(Contact, c1.id) assert %Contact{radar_status: status2} = Repo.get!(Contact, c2.id) refute status1 == :unavailable refute status2 == :unavailable end test "returns {:error, _} on a transport error without pinning contacts :unavailable" do c1 = insert_contact(%{station1: "T1"}) Req.Test.stub(NexradClient, fn conn -> Req.Test.transport_error(conn, :timeout) end) assert {:error, _reason} = perform_job(RadarFrameWorker, %{ "frame_ts" => "2024-09-15T18:30:00Z", "contact_ids" => [c1.id] }) assert %Contact{radar_status: status} = Repo.get!(Contact, c1.id) refute status == :unavailable end test "skips contacts already at a terminal status (idempotent replay)" do c_done = insert_contact(%{station1: "DONE"}) {:ok, _} = c_done |> Ecto.Changeset.change(%{radar_status: :complete}) |> Repo.update() c_new = insert_contact(%{station1: "NEW"}) Req.Test.stub(NexradClient, fn conn -> Plug.Conn.send_resp(conn, 404, "not found") end) assert :ok = perform_job(RadarFrameWorker, %{ "frame_ts" => "2024-09-15T18:30:00Z", "contact_ids" => [c_done.id, c_new.id] }) # Already-complete contact stays complete; new one is marked unavailable. assert %Contact{radar_status: :complete} = Repo.get!(Contact, c_done.id) assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_new.id) end test "processes every eligible contact in the frame's 5-min bucket, not just contact_ids" do # Both contacts land in the 18:30 bucket (18:30 ≤ ts < 18:35). c_passed = insert_contact(%{station1: "PASSED", qso_timestamp: ~U[2024-09-15 18:30:00Z]}) c_omitted = insert_contact(%{station1: "OMITTED", qso_timestamp: ~U[2024-09-15 18:34:00Z]}) # Outside the 18:30 bucket — must not be touched. c_other_bucket = insert_contact(%{station1: "OTHER", qso_timestamp: ~U[2024-09-15 18:35:00Z]}) Req.Test.stub(NexradClient, fn conn -> Plug.Conn.send_resp(conn, 404, "not found") end) # Only c_passed is explicitly listed, but c_omitted is in the same # bucket and still :pending — the worker must pick it up too. assert :ok = perform_job(RadarFrameWorker, %{ "frame_ts" => "2024-09-15T18:30:00Z", "contact_ids" => [c_passed.id] }) assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_passed.id) assert %Contact{radar_status: :unavailable} = Repo.get!(Contact, c_omitted.id) # Outside-bucket contact stays :pending. assert %Contact{radar_status: :pending} = Repo.get!(Contact, c_other_bucket.id) end end describe "NexradClient error contract" do # Pins the exact "NEXRAD n0q HTTP " format emitted by # NexradClient.fetch_decoded_frame/1 on a non-200 response. RadarFrameWorker # pattern-matches on this literal prefix in `permanent_error?/1` to # classify 4xx as permanent (stop retrying) and everything else as # transient (let Oban retry). A silent format change would quietly turn # permanent 404s into infinite retries — fail loudly here instead. test "fetch_decoded_frame/1 returns error starting with \"NEXRAD n0q HTTP \" on 4xx" do Req.Test.stub(NexradClient, fn conn -> Plug.Conn.send_resp(conn, 404, "not found") end) assert {:error, reason} = NexradClient.fetch_decoded_frame(~U[2024-09-15 18:30:00Z]) assert is_binary(reason) assert String.starts_with?(reason, "NEXRAD n0q HTTP ") assert reason == "NEXRAD n0q HTTP 404" end end describe "unique constraint" do test "rejects a second insert with the same frame_ts while one is pending" do frame_ts = "2024-09-15T18:30:00Z" {:ok, first} = Oban.insert(RadarFrameWorker.new(%{"frame_ts" => frame_ts, "contact_ids" => ["a"]})) # Second insert with a different contact_ids list but same frame_ts # must collapse into the existing job — otherwise backfill sweeps # duplicate 5 MB PNG fetches per frame. {:ok, second} = Oban.insert(RadarFrameWorker.new(%{"frame_ts" => frame_ts, "contact_ids" => ["b"]})) assert first.id == second.id end end end