Graham McIntire
aa6c885683
- Replace Module.safe_concat/1 with PID-based names in IemRateLimiterTest
and AggregatorTest (8+6 tests) to avoid binary_to_existing_atom errors
after Elixir 1.20 upgrade
- Fix ContactLive ShowCoverageTest assertions (3 tests) that referenced
removed template text ('in queue') and incompatible weather_status
- Use conn.remote_ip instead of init_test_session for internal_network?
tests since the store_remote_ip plug uses atom keys
98 lines
2.7 KiB
Elixir
98 lines
2.7 KiB
Elixir
defmodule Microwaveprop.Pskr.AggregatorTest do
|
|
use Microwaveprop.DataCase, async: false
|
|
|
|
alias Ecto.Adapters.SQL.Sandbox
|
|
alias Microwaveprop.Pskr.Aggregator
|
|
alias Microwaveprop.Pskr.SpotHourly
|
|
alias Microwaveprop.Repo
|
|
|
|
@sample %{
|
|
"f" => 144_174_000,
|
|
"md" => "FT8",
|
|
"rp" => -5,
|
|
"t_tx" => 1_662_407_697,
|
|
"sc" => "K5ABC",
|
|
"sl" => "EM12kl",
|
|
"rc" => "K7XYZ",
|
|
"rl" => "DM43st",
|
|
"sa" => 291,
|
|
"ra" => 291,
|
|
"b" => "2m"
|
|
}
|
|
|
|
setup do
|
|
pid =
|
|
start_supervised!({Aggregator, name: nil, flush_ms: 0})
|
|
|
|
Sandbox.allow(Repo, self(), pid)
|
|
%{agg: pid}
|
|
end
|
|
|
|
test "ingest collapses repeats on the same path-hour", %{agg: agg} do
|
|
payloads = [
|
|
Jason.encode!(@sample),
|
|
Jason.encode!(%{@sample | "rp" => -3}),
|
|
Jason.encode!(%{@sample | "rp" => -12, "md" => "FT4"})
|
|
]
|
|
|
|
Enum.each(payloads, &Aggregator.ingest(agg, &1))
|
|
|
|
assert Aggregator.flush(agg) == 1
|
|
[row] = Repo.all(SpotHourly)
|
|
|
|
assert row.spot_count == 3
|
|
assert row.max_snr_db == -3
|
|
assert row.min_snr_db == -12
|
|
assert Enum.sort(row.modes) == ["FT4", "FT8"]
|
|
assert row.band == "2m"
|
|
assert row.sender_grid == "EM12KL"
|
|
assert row.receiver_grid == "DM43ST"
|
|
assert row.distance_km > 0
|
|
end
|
|
|
|
test "two paths in the same hour produce two rows", %{agg: agg} do
|
|
Aggregator.ingest(agg, Jason.encode!(@sample))
|
|
Aggregator.ingest(agg, Jason.encode!(%{@sample | "sl" => "FN31pr"}))
|
|
|
|
assert Aggregator.flush(agg) == 2
|
|
assert Repo.aggregate(SpotHourly, :count) == 2
|
|
end
|
|
|
|
test "second flush merges into existing row via upsert", %{agg: agg} do
|
|
Aggregator.ingest(agg, Jason.encode!(@sample))
|
|
assert Aggregator.flush(agg) == 1
|
|
|
|
Aggregator.ingest(agg, Jason.encode!(%{@sample | "rp" => 5}))
|
|
assert Aggregator.flush(agg) == 1
|
|
|
|
[row] = Repo.all(SpotHourly)
|
|
assert row.spot_count == 2
|
|
assert row.max_snr_db == 5
|
|
assert row.min_snr_db == -5
|
|
end
|
|
|
|
test "malformed payloads are dropped silently", %{agg: agg} do
|
|
Aggregator.ingest(agg, "not json")
|
|
Aggregator.ingest(agg, Jason.encode!(%{"oops" => true}))
|
|
Aggregator.ingest(agg, Jason.encode!(@sample))
|
|
|
|
assert Aggregator.flush(agg) == 1
|
|
end
|
|
|
|
test "pending_count returns the number of unflushed rows", %{agg: agg} do
|
|
assert Aggregator.pending_count(agg) == 0
|
|
|
|
Aggregator.ingest(agg, Jason.encode!(@sample))
|
|
assert Aggregator.pending_count(agg) == 1
|
|
|
|
Aggregator.ingest(agg, Jason.encode!(%{@sample | "sl" => "FN31pr"}))
|
|
assert Aggregator.pending_count(agg) == 2
|
|
|
|
Aggregator.flush(agg)
|
|
assert Aggregator.pending_count(agg) == 0
|
|
end
|
|
|
|
test "flush with empty rows returns 0", %{agg: agg} do
|
|
assert Aggregator.flush(agg) == 0
|
|
end
|
|
end
|