defmodule VavPipeline.AzazelIngestTest do @moduledoc """ Tests for the cross-language Azazel ingest GenServer (Pass 4 / Finding F, commit 7). These tests verify the file-watcher logic and the AzazelRecord changeset validation; they do NOT require PostgreSQL because the recorder is injected as a mock. L3-side tests (the actual record_disposal write) live in l3_store_test.exs tagged :l3. """ use ExUnit.Case, async: false alias VavPipeline.AzazelIngest alias VavPipeline.L3.AzazelRecord # ============================================================ # Test helpers # ============================================================ setup do # Per-test pending directory so tests don't share state. dir = System.tmp_dir!() |> Path.join("azazel-ingest-test-#{:erlang.unique_integer([:positive])}") File.mkdir_p!(dir) on_exit(fn -> File.rm_rf!(dir) end) %{pending_dir: dir} end defp valid_record_json(case_ref \\ "case-001") do %{ "case_ref" => case_ref, "vidui" => %{ "avonot" => [], "peshaim" => [], "chatot" => ["empty-vital offering: test"], "confessor" => "kohen-gadol", "confessed_at" => "2026-04-28T12:00:00Z" }, "routing" => "eretz_gezerah", "authorized_by" => "kohen-gadol", "shaliach" => "aharon", "triggering_call" => "external_offering", "goral_outcome_id" => nil, "disposed_at" => "2026-04-28T12:00:01Z" } |> Jason.encode!() end defp drop_file(dir, name, body) do File.write!(Path.join(dir, name), body) end defp start_ingest(dir, opts \\ []) do name = :"ingest_#{:erlang.unique_integer([:positive])}" captured = :ets.new(:captured, [:public, :bag]) recorder = fn attrs -> :ets.insert(captured, {:record, attrs}) {:ok, attrs} end {:ok, pid} = AzazelIngest.start_link( Keyword.merge( [ name: name, pending_dir: dir, poll_interval_ms: 60_000, # don't auto-poll during tests recorder: recorder ], opts ) ) on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) %{server: name, captured: captured, recorder: recorder} end # ============================================================ # Happy path # ============================================================ test "poll_now ingests a valid file and shreds it", %{pending_dir: dir} do drop_file(dir, "case-001.json", valid_record_json()) %{server: server, captured: captured} = start_ingest(dir) AzazelIngest.poll_now(server) # File was shredded refute File.exists?(Path.join(dir, "case-001.json")) # Record was passed to the recorder assert [{:record, attrs}] = :ets.tab2list(captured) assert attrs.case_ref == "case-001" assert attrs.routing == "eretz_gezerah" assert attrs.authorized_by == "kohen-gadol" assert attrs.shaliach == "aharon" assert %DateTime{} = attrs.disposed_at end test "poll_now ingests multiple files in sorted order", %{pending_dir: dir} do drop_file(dir, "case-002.json", valid_record_json("case-002")) drop_file(dir, "case-001.json", valid_record_json("case-001")) drop_file(dir, "case-003.json", valid_record_json("case-003")) %{server: server, captured: captured} = start_ingest(dir) AzazelIngest.poll_now(server) refs = :ets.tab2list(captured) |> Enum.map(fn {:record, attrs} -> attrs.case_ref end) assert refs == ["case-001", "case-002", "case-003"] end test "ingest preserves the vidui as a map (jsonb column)", %{pending_dir: dir} do drop_file(dir, "case-vidui.json", valid_record_json()) %{server: server, captured: captured} = start_ingest(dir) AzazelIngest.poll_now(server) [{:record, attrs}] = :ets.tab2list(captured) assert is_map(attrs.vidui) assert attrs.vidui["chatot"] == ["empty-vital offering: test"] assert attrs.vidui["confessor"] == "kohen-gadol" end # ============================================================ # Retry semantics — failed records left in place # ============================================================ test "file is left in place when recorder returns error", %{pending_dir: dir} do drop_file(dir, "case-fail.json", valid_record_json()) name = :"ingest_fail_#{:erlang.unique_integer([:positive])}" failing_recorder = fn _ -> {:error, :db_unavailable} end {:ok, pid} = AzazelIngest.start_link( name: name, pending_dir: dir, poll_interval_ms: 60_000, recorder: failing_recorder ) on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) AzazelIngest.poll_now(name) # File still present for retry assert File.exists?(Path.join(dir, "case-fail.json")) state = AzazelIngest.state(name) assert state.errors == 1 assert state.records_ingested == 0 end test "file is left in place when JSON is malformed", %{pending_dir: dir} do drop_file(dir, "case-malformed.json", "not-json{") %{server: server} = start_ingest(dir) AzazelIngest.poll_now(server) # File still present (operator can investigate) assert File.exists?(Path.join(dir, "case-malformed.json")) state = AzazelIngest.state(server) assert state.errors == 1 end test "file is left in place when disposed_at is missing", %{pending_dir: dir} do raw = %{"case_ref" => "case-no-ts", "vidui" => %{}, "routing" => "eretz_gezerah"} |> Jason.encode!() drop_file(dir, "case-no-ts.json", raw) %{server: server} = start_ingest(dir) AzazelIngest.poll_now(server) assert File.exists?(Path.join(dir, "case-no-ts.json")) end # ============================================================ # State accounting # ============================================================ test "state tracks polls_completed, records_ingested, errors", %{pending_dir: dir} do drop_file(dir, "case-001.json", valid_record_json("case-001")) drop_file(dir, "case-002.json", valid_record_json("case-002")) drop_file(dir, "case-bad.json", "garbage") %{server: server} = start_ingest(dir) AzazelIngest.poll_now(server) state = AzazelIngest.state(server) assert state.polls_completed == 1 assert state.records_ingested == 2 assert state.errors == 1 end test "state survives multiple polls", %{pending_dir: dir} do %{server: server} = start_ingest(dir) AzazelIngest.poll_now(server) AzazelIngest.poll_now(server) AzazelIngest.poll_now(server) state = AzazelIngest.state(server) assert state.polls_completed == 3 end test "non-json files are ignored", %{pending_dir: dir} do File.write!(Path.join(dir, "README.txt"), "not a record") drop_file(dir, "case-001.json", valid_record_json()) %{server: server, captured: captured} = start_ingest(dir) AzazelIngest.poll_now(server) # JSON file processed; README left alone refute File.exists?(Path.join(dir, "case-001.json")) assert File.exists?(Path.join(dir, "README.txt")) assert length(:ets.tab2list(captured)) == 1 end # ============================================================ # Property 6 — out-of-band acknowledgment via Chatzotzrot # ============================================================ test "successful ingest fires Chatzotzrot :moadim signal", %{pending_dir: dir} do # The supervision-tree-started Chatzotzrot is already running. # Subscribe this test process as an :operators recipient and # observe the signal AzazelIngest fires. VavPipeline.Chatzotzrot.subscribe(:all) drop_file(dir, "case-ack-001.json", valid_record_json("case-ack-001")) %{server: server} = start_ingest(dir) AzazelIngest.poll_now(server) # Drain any unrelated signals from prior tests; we want OUR signal. receive_until_ours("case-ack-001") end test "chatzotzrot fire does NOT include vidui item content (privacy)" do # Property 9's privacy boundary applies to the chatzotzrot # summary too — the operator-side signal channel must not leak # the operator-internal vidui items. secret_phrase = "must-not-leak-via-chatzotzrot-summary-channel" body = %{ "case_ref" => "leak-001", "vidui" => %{ "avonot" => [], "peshaim" => [], "chatot" => [secret_phrase], "confessor" => "kohen", "confessed_at" => "2026-04-28T12:00:00Z" }, "routing" => "eretz_gezerah", "authorized_by" => "kohen-gadol", "shaliach" => "aharon", "triggering_call" => "test", "goral_outcome_id" => nil, "disposed_at" => "2026-04-28T12:00:01Z" } |> Jason.encode!() VavPipeline.Chatzotzrot.subscribe(:all) dir = System.tmp_dir!() |> Path.join("azazel-leak-#{:erlang.unique_integer([:positive])}") File.mkdir_p!(dir) on_exit(fn -> File.rm_rf!(dir) end) File.write!(Path.join(dir, "leak-001.json"), body) %{server: server} = start_ingest(dir) AzazelIngest.poll_now(server) signal = receive_until_ours("leak-001") refute signal.details.reason =~ secret_phrase end # Drain any unrelated chatzotzrot signals (from other tests sharing # the supervision-tree Chatzotzrot) until we see ours, identified by # case_ref in the reason. Returns the matching signal. defp receive_until_ours(case_ref) do receive do {:chatzotzrot, signal} -> if String.contains?(signal.details.reason || "", case_ref) and signal.signal_type == :moadim do assert signal.details.reason =~ "Azazel disposal completed" signal else receive_until_ours(case_ref) end after 2_000 -> flunk("Did not receive chatzotzrot signal for #{case_ref}") end end # ============================================================ # Changeset validation (no DB) # ============================================================ describe "AzazelRecord.changeset/2" do test "valid attrs produce a valid changeset" do attrs = %{ case_ref: "case-cs-001", vidui: %{ "avonot" => [], "peshaim" => [], "chatot" => ["something"], "confessor" => "kohen", "confessed_at" => "2026-04-28T12:00:00Z" }, routing: "eretz_gezerah", authorized_by: "kohen-gadol", shaliach: "aharon", disposed_at: ~U[2026-04-28 12:00:01Z] } cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) assert cs.valid? end test "rejects routing other than eretz_gezerah" do attrs = base_attrs(%{routing: "void"}) cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) refute cs.valid? assert {:routing, _} = List.keyfind(cs.errors, :routing, 0) end test "rejects vidui missing avonot slot" do attrs = base_attrs(%{ vidui: %{ "peshaim" => [], "chatot" => ["x"], "confessor" => "kohen", "confessed_at" => "2026-04-28T12:00:00Z" } }) cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) refute cs.valid? assert {:vidui, {msg, _}} = List.keyfind(cs.errors, :vidui, 0) assert msg =~ "avonot" end test "rejects vidui with all three categories empty" do attrs = base_attrs(%{ vidui: %{ "avonot" => [], "peshaim" => [], "chatot" => [], "confessor" => "kohen", "confessed_at" => "2026-04-28T12:00:00Z" } }) cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) refute cs.valid? assert {:vidui, {msg, _}} = List.keyfind(cs.errors, :vidui, 0) assert msg =~ "untagged" end test "rejects missing required fields" do cs = AzazelRecord.changeset(%AzazelRecord{}, %{}) refute cs.valid? missing = Enum.map(cs.errors, fn {k, _} -> k end) for required <- [:case_ref, :vidui, :routing, :authorized_by, :shaliach, :disposed_at] do assert required in missing, "expected #{required} to be missing" end end test "accepts optional triggering_call and goral_outcome_id" do attrs = base_attrs(%{ triggering_call: "external_offering", goral_outcome_id: String.duplicate("a", 64) }) cs = AzazelRecord.changeset(%AzazelRecord{}, attrs) assert cs.valid? end defp base_attrs(overrides \\ %{}) do Map.merge( %{ case_ref: "case-base-001", vidui: %{ "avonot" => [], "peshaim" => [], "chatot" => ["something"], "confessor" => "kohen", "confessed_at" => "2026-04-28T12:00:00Z" }, routing: "eretz_gezerah", authorized_by: "kohen-gadol", shaliach: "aharon", disposed_at: ~U[2026-04-28 12:00:01Z] }, overrides ) end end end