prop/test/microwaveprop/weather/era5_client_test.exs
Graham McIntire 1ec10bec1f
Split ERA5 backfill into submit/poll workers with persistent CDS state
The single Era5MonthBatchWorker pinned an Oban slot for the full 30-45
min a CDS month-tile takes to assemble, and lost the work entirely on a
rolling deploy because the in-flight Task died with the pod. This splits
the flow into two tiny workers and persists the CDS job IDs so deploys
survive.

New pipeline:

1. Era5SubmitWorker (:era5_submit queue) — POSTs both CDS requests in
   parallel, writes an `era5_cds_jobs` row with the returned job IDs,
   and enqueues an Era5PollWorker scheduled +5 min. ~1s of real work.
   Short-circuits when the month-tile is already cached or when a row
   already exists (in-flight from a previous attempt).

2. Era5PollWorker (:era5_poll queue) — reads the row, calls
   Era5Client.check_status/1 for both CDS job IDs, and:
     - returns {:snooze, 300} if either job is still running (Oban
       re-schedules without counting an attempt and releases the slot
       immediately — a pod can keep dozens of tile-months in flight
       without pinning workers)
     - streams both GRIB files to disk via Req into: File.stream!,
       decodes via Wgrib2.extract_grid_messages_from_file, bulk-inserts
       via Era5BatchClient.decode_and_insert/6, deletes the row, and
       DELETEs both completed jobs from CDS to free server-side quota
     - if either leg CDS-reports failed, deletes the row + both CDS
       jobs and returns {:error, reason}

Era5Client gains four testable building blocks:
  submit_job/2               (bare POST → {:ok, job_id})
  check_status/1             (GET → :running | {:done, src} | {:failed, reason})
  download_source_to_file/3  (streams {:url, href} or writes {:body, bin})
  delete_job/1               (DELETE /jobs/:id, treats 200/202/204/404 as :ok)

All Req calls now route through `era5_req_options` so tests can stub
CDS responses via Req.Test.stub(Era5Client, fn).

Era5MonthBatchWorker is retained as a thin forwarder to Era5SubmitWorker
so any jobs already in the :era5_batch queue on prod pods drain cleanly
on the next rolling deploy. Safe to delete in a follow-up.

Adds era5_cds_jobs table with a unique index on
(year, month, tile_lat, tile_lon) so duplicate submits collapse.

New queue config in runtime.exs:
  era5_submit: local_limit 4, rate_limit 30/hour (burst protection)
  era5_poll:   local_limit 20 (polls are cheap GETs)
  era5_batch:  kept at 1 for legacy job drain, delete next cycle
2026-04-13 16:26:26 -05:00

176 lines
6 KiB
Elixir

defmodule Microwaveprop.Weather.Era5ClientTest do
# async: false — tests mutate the global ERA5_CDS_API_KEY env var to match
# production's lookup path. Running concurrently with Era5BatchClientTest
# (which *deletes* the env var to exercise the missing-key error path)
# leaks state in both directions.
use ExUnit.Case, async: false
alias Microwaveprop.Weather.Era5Client
setup do
# A real key — stubs don't care about value, only presence.
System.put_env("ERA5_CDS_API_KEY", "test-key")
on_exit(fn -> System.delete_env("ERA5_CDS_API_KEY") end)
:ok
end
describe "submit_job/2" do
test "returns {:ok, job_id} when CDS accepts the request" do
Req.Test.stub(Era5Client, fn conn ->
assert conn.method == "POST"
assert String.ends_with?(conn.request_path, "/reanalysis-era5-single-levels/execution")
Req.Test.json(conn, %{"jobID" => "cds-job-123", "status" => "accepted"})
end)
assert {:ok, "cds-job-123"} =
Era5Client.submit_job("reanalysis-era5-single-levels", %{"foo" => "bar"})
end
test "returns {:error, _} when CDS returns 400" do
Req.Test.stub(Era5Client, fn conn ->
conn
|> Plug.Conn.put_resp_content_type("application/json")
|> Plug.Conn.resp(400, ~s({"detail":"bad request"}))
end)
assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
assert reason =~ "HTTP 400"
end
test "returns {:error, _} when API key is missing" do
System.delete_env("ERA5_CDS_API_KEY")
assert {:error, reason} = Era5Client.submit_job("reanalysis-era5-single-levels", %{})
assert reason =~ "ERA5_CDS_API_KEY"
end
end
describe "check_status/1" do
test "returns :running while CDS job is accepted" do
Req.Test.stub(Era5Client, fn conn ->
assert conn.method == "GET"
assert String.ends_with?(conn.request_path, "/jobs/cds-job-123")
Req.Test.json(conn, %{"status" => "accepted"})
end)
assert :running = Era5Client.check_status("cds-job-123")
end
test "returns :running while CDS job is running" do
Req.Test.stub(Era5Client, fn conn ->
Req.Test.json(conn, %{"status" => "running"})
end)
assert :running = Era5Client.check_status("cds-job-123")
end
test "returns {:done, {:url, href}} when CDS job completes with an asset href" do
Req.Test.stub(Era5Client, fn conn ->
case conn.request_path do
"/api/retrieve/v1/jobs/cds-job-123" ->
Req.Test.json(conn, %{"status" => "successful"})
"/api/retrieve/v1/jobs/cds-job-123/results" ->
Req.Test.json(conn, %{
"asset" => %{"value" => %{"href" => "https://example.com/file.grib"}}
})
end
end)
assert {:done, {:url, "https://example.com/file.grib"}} =
Era5Client.check_status("cds-job-123")
end
test "returns {:failed, _} when CDS reports the job failed" do
Req.Test.stub(Era5Client, fn conn ->
assert conn.request_path == "/api/retrieve/v1/jobs/cds-job-123"
Req.Test.json(conn, %{"status" => "failed", "message" => "input out of range"})
end)
assert {:failed, reason} = Era5Client.check_status("cds-job-123")
assert reason =~ "input out of range"
end
end
describe "delete_job/1" do
test "sends DELETE to the CDS jobs endpoint and returns :ok on 204" do
test_pid = self()
Req.Test.stub(Era5Client, fn conn ->
send(test_pid, {:delete_request, conn.method, conn.request_path})
Plug.Conn.resp(conn, 204, "")
end)
assert :ok = Era5Client.delete_job("cds-job-123")
assert_received {:delete_request, "DELETE", "/api/retrieve/v1/jobs/cds-job-123"}
end
test "also treats 200/202/404 as success (CDS sometimes returns 200 or the job may already be gone)" do
for status <- [200, 202, 404] do
Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, status, "") end)
assert :ok = Era5Client.delete_job("cds-job-#{status}")
end
end
test "returns {:error, _} for unexpected statuses" do
Req.Test.stub(Era5Client, fn conn -> Plug.Conn.resp(conn, 500, "server error") end)
assert {:error, reason} = Era5Client.delete_job("cds-job-500")
assert reason =~ "HTTP 500"
end
end
describe "download_source_to_file/3" do
test "writes a {:body, binary} source directly to the target path" do
path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
try do
assert :ok = Era5Client.download_source_to_file({:body, "grib-payload"}, nil, path)
assert File.read!(path) == "grib-payload"
after
File.rm(path)
end
end
test "streams a {:url, href} source through Req into the target path" do
Req.Test.stub(Era5Client, fn conn ->
assert conn.method == "GET"
conn
|> Plug.Conn.put_resp_content_type("application/octet-stream")
|> Plug.Conn.resp(200, "streamed-grib-data")
end)
path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
try do
assert :ok =
Era5Client.download_source_to_file(
{:url, "https://example.com/file.grib"},
"test-key",
path
)
assert File.read!(path) == "streamed-grib-data"
after
File.rm(path)
end
end
test "removes partial file and returns {:error, _} on HTTP error" do
Req.Test.stub(Era5Client, fn conn ->
Plug.Conn.resp(conn, 503, "")
end)
path = Path.join(System.tmp_dir!(), "era5_test_#{System.unique_integer([:positive])}.grib")
assert {:error, reason} =
Era5Client.download_source_to_file(
{:url, "https://example.com/file.grib"},
"test-key",
path
)
assert reason =~ "HTTP 503"
refute File.exists?(path)
end
end
end