defmodule Microwaveprop.Workers.PropagationGridWorkerTest do @moduledoc """ Tests the Phase-3-cutover seeder behaviour. Elixir no longer runs the chain step; the hourly cron fires `perform(%Oban.Job{args: %{}})` with empty args and the worker inserts grid_tasks rows (1 analysis f00 + forecast f01..f18 hourly, f21..f48 3-hourly) for the Rust `prop-grid-rs` worker to drain. Rust owns HRRR fetch, wgrib2 decode, native duct, NEXRAD, commercial, scoring, and both the ProfilesFile and band score-file writes from here on out. """ use Microwaveprop.DataCase, async: false use Oban.Testing, repo: Microwaveprop.Repo import Ecto.Query alias Microwaveprop.Workers.PropagationGridWorker alias Microwaveprop.Workers.PropagationPruneWorker describe "pick_run_time/1 (HRRR cycle freshness selection)" do setup do original = Application.get_env(:microwaveprop, :hrrr_cycle_available_fn) on_exit(fn -> if original, do: Application.put_env(:microwaveprop, :hrrr_cycle_available_fn, original), else: Application.delete_env(:microwaveprop, :hrrr_cycle_available_fn) end) end test "picks now-1h when that cycle has fully published" do Application.put_env(:microwaveprop, :hrrr_cycle_available_fn, fn _dt -> true end) # Cron fires at HH:05, so use that as `now`. now-1h = 23:05 rounds # to 23:00 (the freshly-published cycle). now = ~U[2026-04-25 00:05:00Z] assert PropagationGridWorker.pick_run_time(now) == ~U[2026-04-24 23:00:00Z] end test "falls back to now-2h when the now-1h cycle is not yet available" do Application.put_env(:microwaveprop, :hrrr_cycle_available_fn, fn _dt -> false end) now = ~U[2026-04-25 00:05:00Z] assert PropagationGridWorker.pick_run_time(now) == ~U[2026-04-24 22:00:00Z] end test "probes the rounded now-1h value, not the raw `now`" do test_pid = self() Application.put_env(:microwaveprop, :hrrr_cycle_available_fn, fn dt -> send(test_pid, {:probed, dt}) true end) # 00:05 UTC: -1h = 23:05 UTC, rounds to 23:00 UTC. PropagationGridWorker.pick_run_time(~U[2026-04-25 00:05:00Z]) assert_receive {:probed, ~U[2026-04-24 23:00:00Z]} end end describe "queue priority" do test "PropagationGridWorker runs at the highest priority on :propagation" do assert PropagationGridWorker.__opts__()[:priority] == 0 end test "PropagationPruneWorker yields to the grid chain" do assert PropagationPruneWorker.__opts__()[:priority] > 0 end end describe "perform/1 — chain seeding (empty args)" do test "inserts 1 analysis + forecast grid_tasks rows; no Oban fan-out" do Oban.Testing.with_testing_mode(:manual, fn -> assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}}) # No chain-step Oban jobs fan out anymore — the entire chain # lives in grid_tasks from the Rust worker's perspective. jobs = all_enqueued(worker: PropagationGridWorker) assert jobs == [] {_count, [first_task]} = "grid_tasks" |> Microwaveprop.Repo.insert_all( [], returning: [:run_time, :kind, :forecast_hour] ) |> case do # When the worker seeded a run, the table has 19 fresh rows. # Pull the analysis row so the test can derive run_time for # the subsequent query. _ -> from(t in "grid_tasks", where: t.kind == "analysis", order_by: [desc: t.inserted_at], limit: 1, select: %{run_time: t.run_time, forecast_hour: t.forecast_hour, kind: t.kind} ) |> Microwaveprop.Repo.all() |> case do [row] -> {1, [row]} _ -> {0, [%{run_time: nil, forecast_hour: nil, kind: nil}]} end end assert first_task.kind == "analysis" assert first_task.forecast_hour == 0 run_time = first_task.run_time assert run_time.minute == 0 assert run_time.second == 0 rows = Microwaveprop.Repo.all( from t in "grid_tasks", where: t.run_time == ^run_time, select: %{fh: t.forecast_hour, kind: t.kind}, order_by: [t.kind, t.forecast_hour] ) expected_fhs = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 21, 24, 27, 30, 33, 36, 39, 42, 45, 48 ] kinds = rows |> Enum.map(& &1.kind) |> Enum.frequencies() assert kinds == %{"analysis" => 1, "forecast" => 28} forecast_fhs = rows |> Enum.filter(&(&1.kind == "forecast")) |> Enum.map(& &1.fh) |> Enum.sort() assert forecast_fhs == expected_fhs end) end end describe "telemetry instrumentation" do test "perform/1 emits :perform span plus :seeded event with queued_tasks count" do Oban.Testing.with_testing_mode(:manual, fn -> test_pid = self() handler_id = {__MODULE__, :grid_worker_perform} :telemetry.attach_many( handler_id, [ [:microwaveprop, :propagation, :grid_worker, :perform, :start], [:microwaveprop, :propagation, :grid_worker, :perform, :stop], [:microwaveprop, :propagation, :grid_worker, :seeded] ], fn event, measurements, metadata, _config -> send(test_pid, {:telemetry, event, measurements, metadata}) end, nil ) try do assert :ok = PropagationGridWorker.perform(%Oban.Job{args: %{}}) assert_receive {:telemetry, [:microwaveprop, :propagation, :grid_worker, :perform, :start], _, _} stop_event = [:microwaveprop, :propagation, :grid_worker, :perform, :stop] assert_receive {:telemetry, ^stop_event, stop_measurements, _} assert is_integer(stop_measurements.duration) assert stop_measurements.duration >= 0 assert_receive {:telemetry, [:microwaveprop, :propagation, :grid_worker, :seeded], %{queued_tasks: count}, _} assert count == 29 after :telemetry.detach(handler_id) end end) end end end