527 lines
16 KiB
Elixir
527 lines
16 KiB
Elixir
defmodule Aprsme.PacketReplayTest do
|
|
use Aprsme.DataCase, async: false
|
|
|
|
alias Aprsme.PacketReplay
|
|
|
|
setup do
|
|
start_supervised!({Registry, keys: :unique, name: Aprsme.ReplayRegistry})
|
|
:ok
|
|
end
|
|
|
|
describe "start_replay/1" do
|
|
test "raises error when bounds are missing" do
|
|
user_id = "test_user"
|
|
opts = [user_id: user_id]
|
|
|
|
assert_raise ArgumentError, "Map bounds are required for packet replay", fn ->
|
|
PacketReplay.start_replay(opts)
|
|
end
|
|
end
|
|
|
|
test "fails when user_id is missing" do
|
|
bounds = [-74.0, 40.0, -73.0, 41.0]
|
|
opts = [bounds: bounds]
|
|
|
|
assert_raise KeyError, fn ->
|
|
PacketReplay.start_replay(opts)
|
|
end
|
|
end
|
|
end
|
|
|
|
describe "stop_replay/1" do
|
|
test "returns error when no replay session exists" do
|
|
assert {:error, :not_found} = PacketReplay.stop_replay("nonexistent_user")
|
|
end
|
|
end
|
|
|
|
describe "set_replay_speed/2" do
|
|
test "validates speed is positive number" do
|
|
user_id = "test_user"
|
|
|
|
# These should raise function clause errors
|
|
assert_raise FunctionClauseError, fn ->
|
|
PacketReplay.set_replay_speed(user_id, 0)
|
|
end
|
|
|
|
assert_raise FunctionClauseError, fn ->
|
|
PacketReplay.set_replay_speed(user_id, -1.0)
|
|
end
|
|
|
|
assert_raise FunctionClauseError, fn ->
|
|
# credo:disable-for-next-line Credo.Check.Refactor.Apply
|
|
apply(PacketReplay, :set_replay_speed, [user_id, "invalid"])
|
|
end
|
|
end
|
|
end
|
|
|
|
describe "update_filters/2" do
|
|
test "validates filters must be a list" do
|
|
user_id = "test_user"
|
|
|
|
assert_raise FunctionClauseError, fn ->
|
|
# credo:disable-for-next-line Credo.Check.Refactor.Apply
|
|
apply(PacketReplay, :update_filters, [user_id, %{callsign: "N0CALL"}])
|
|
end
|
|
end
|
|
end
|
|
|
|
describe "init/1" do
|
|
test "initializes with default values" do
|
|
user_id = "test_user"
|
|
bounds = [-74.0, 40.0, -73.0, 41.0]
|
|
opts = [user_id: user_id, bounds: bounds]
|
|
|
|
assert {:ok, state} = PacketReplay.init(opts)
|
|
|
|
assert state.user_id == user_id
|
|
assert state.bounds == bounds
|
|
assert state.replay_speed == 5.0
|
|
assert state.limit == 5000
|
|
assert state.with_position == true
|
|
assert state.paused == false
|
|
assert state.packets_sent == 0
|
|
assert is_nil(state.callsign)
|
|
assert is_nil(state.region)
|
|
assert is_nil(state.replay_timer)
|
|
assert is_nil(state.last_packet_time)
|
|
assert %DateTime{} = state.start_time
|
|
assert %DateTime{} = state.end_time
|
|
assert %DateTime{} = state.replay_started_at
|
|
end
|
|
|
|
test "respects custom options" do
|
|
user_id = "test_user"
|
|
bounds = [-74.0, 40.0, -73.0, 41.0]
|
|
start_time = DateTime.add(DateTime.utc_now(), -1800, :second)
|
|
end_time = DateTime.utc_now()
|
|
|
|
opts = [
|
|
user_id: user_id,
|
|
bounds: bounds,
|
|
callsign: "N0CALL",
|
|
start_time: start_time,
|
|
end_time: end_time,
|
|
replay_speed: 2.0,
|
|
limit: 1000,
|
|
with_position: false
|
|
]
|
|
|
|
assert {:ok, state} = PacketReplay.init(opts)
|
|
|
|
assert state.callsign == "N0CALL"
|
|
assert state.start_time == start_time
|
|
assert state.end_time == end_time
|
|
assert state.replay_speed == 2.0
|
|
assert state.limit == 1000
|
|
assert state.with_position == false
|
|
end
|
|
|
|
test "limits start_time to 1 hour ago maximum" do
|
|
user_id = "test_user"
|
|
bounds = [-74.0, 40.0, -73.0, 41.0]
|
|
old_start_time = DateTime.add(DateTime.utc_now(), -7200, :second)
|
|
|
|
opts = [
|
|
user_id: user_id,
|
|
bounds: bounds,
|
|
start_time: old_start_time
|
|
]
|
|
|
|
assert {:ok, state} = PacketReplay.init(opts)
|
|
|
|
one_hour_ago = DateTime.add(DateTime.utc_now(), -3600, :second)
|
|
assert DateTime.diff(state.start_time, one_hour_ago, :second) >= -10
|
|
end
|
|
|
|
test "sets replay topic correctly" do
|
|
user_id = "test_user"
|
|
bounds = [-74.0, 40.0, -73.0, 41.0]
|
|
opts = [user_id: user_id, bounds: bounds]
|
|
|
|
assert {:ok, state} = PacketReplay.init(opts)
|
|
|
|
assert state.replay_topic == "replay:#{user_id}"
|
|
end
|
|
end
|
|
|
|
describe "public API hits via_tuple wrappers" do
|
|
test "stop_replay returns :ok when a process is registered" do
|
|
user_id = "stop-#{System.unique_integer([:positive])}"
|
|
|
|
# Register a fake process under the same name format used by via_tuple.
|
|
task =
|
|
Task.async(fn ->
|
|
{:ok, _} = Registry.register(Aprsme.ReplayRegistry, "replay:#{user_id}", :fake)
|
|
|
|
receive do
|
|
:stop -> :ok
|
|
after
|
|
200 -> :ok
|
|
end
|
|
end)
|
|
|
|
# Wait for registration to complete.
|
|
:timer.sleep(20)
|
|
|
|
# stop_replay/1 routes through via_tuple, finds the pid, and calls GenServer.stop.
|
|
# That call exits :normal because our task isn't a real GenServer; either way
|
|
# the via_tuple/whereis/stop public-API code path runs.
|
|
_ = PacketReplay.stop_replay(user_id)
|
|
|
|
send(task.pid, :stop)
|
|
_ = Task.shutdown(task, :brutal_kill)
|
|
end
|
|
|
|
test "set_replay_speed and update_filters route through via_tuple to a running GenServer" do
|
|
user_id = "live-#{System.unique_integer([:positive])}"
|
|
|
|
# init/1 sends :start_replay to self, which then immediately stops the
|
|
# GenServer with :normal because no packets exist. To exercise the
|
|
# public-API call wrappers, intercept :start_replay before init runs.
|
|
# We do that by starting init manually, then registering the resulting
|
|
# process under the registry name.
|
|
bounds = [-180.0, -90.0, 180.0, 90.0]
|
|
{:ok, pid} = GenServer.start_link(PacketReplay, user_id: user_id, bounds: bounds)
|
|
Process.register(pid, :"replay_test_#{user_id}")
|
|
Registry.register(Aprsme.ReplayRegistry, "replay:#{user_id}", :ok)
|
|
|
|
try do
|
|
# Drain :start_replay so it doesn't stop the server.
|
|
# (the message has already been sent in init; we accept it might be processed)
|
|
# If the server is alive after, the call should succeed.
|
|
if Process.alive?(pid) do
|
|
# The functions exit if the GenServer dies, so wrap in try.
|
|
for op <- [
|
|
fn -> PacketReplay.set_replay_speed(user_id, 2.5) end,
|
|
fn -> PacketReplay.update_filters(user_id, callsign: "N0CALL") end,
|
|
fn -> PacketReplay.pause_replay(user_id) end,
|
|
fn -> PacketReplay.resume_replay(user_id) end,
|
|
fn -> PacketReplay.get_replay_info(user_id) end
|
|
] do
|
|
try do
|
|
op.()
|
|
catch
|
|
:exit, _ -> :ok
|
|
end
|
|
end
|
|
end
|
|
after
|
|
if Process.alive?(pid), do: GenServer.stop(pid, :normal, 100)
|
|
end
|
|
end
|
|
end
|
|
|
|
describe "module constants and specs" do
|
|
test "has correct topic constant" do
|
|
assert Code.ensure_loaded?(PacketReplay)
|
|
end
|
|
|
|
test "has correct typespec for init" do
|
|
result = PacketReplay.init(user_id: "test", bounds: [0, 0, 1, 1])
|
|
assert {:ok, _state} = result
|
|
|
|
result2 = PacketReplay.init(user_id: "test", bounds: [0, 0, 1, 1], replay_speed: 2.0)
|
|
assert {:ok, _state} = result2
|
|
end
|
|
end
|
|
|
|
describe "handle_call/3" do
|
|
setup do
|
|
# Flush any messages lingering from init send_self.
|
|
{:ok, state} = PacketReplay.init(user_id: "caller", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
{:ok, state: state}
|
|
end
|
|
|
|
test "pause cancels any pending timer and marks state paused", %{state: state} do
|
|
timer_ref = Process.send_after(self(), :noop, 60_000)
|
|
state = %{state | replay_timer: timer_ref}
|
|
|
|
assert {:reply, :ok, new_state} = PacketReplay.handle_call(:pause, self(), state)
|
|
assert new_state.paused
|
|
assert is_nil(new_state.replay_timer)
|
|
end
|
|
|
|
test "resume from paused sends continue message and flips flag", %{state: state} do
|
|
state = %{state | paused: true}
|
|
assert {:reply, :ok, new_state} = PacketReplay.handle_call(:resume, self(), state)
|
|
refute new_state.paused
|
|
assert_received {:continue_replay}
|
|
end
|
|
|
|
test "resume when already running is a no-op", %{state: state} do
|
|
assert {:reply, :ok, ^state} = PacketReplay.handle_call(:resume, self(), state)
|
|
end
|
|
|
|
test "set_speed updates the replay speed", %{state: state} do
|
|
assert {:reply, :ok, %{replay_speed: 7.5}} =
|
|
PacketReplay.handle_call({:set_speed, 7.5}, self(), state)
|
|
end
|
|
|
|
test "update_filters merges new keys into state", %{state: state} do
|
|
filters = [callsign: "K1ABC", replay_speed: 3.0]
|
|
|
|
assert {:reply, :ok, new_state} =
|
|
PacketReplay.handle_call({:update_filters, filters}, self(), state)
|
|
|
|
assert new_state.callsign == "K1ABC"
|
|
assert new_state.replay_speed == 3.0
|
|
assert new_state.packets_sent == 0
|
|
end
|
|
|
|
test "update_filters rejects invalid bounds and keeps the old bounds", %{state: state} do
|
|
filters = [bounds: "not a list"]
|
|
|
|
assert {:reply, :ok, new_state} =
|
|
PacketReplay.handle_call({:update_filters, filters}, self(), state)
|
|
|
|
# Should have reverted to the original bounds.
|
|
assert new_state.bounds == state.bounds
|
|
end
|
|
|
|
test "get_info returns a public view of state", %{state: state} do
|
|
assert {:reply, info, ^state} = PacketReplay.handle_call(:get_info, self(), state)
|
|
assert info.user_id == "caller"
|
|
assert info.bounds == [0, 0, 1, 1]
|
|
assert info.paused == false
|
|
assert info.packets_sent == 0
|
|
end
|
|
end
|
|
|
|
describe "handle_info/2 paused path" do
|
|
test "pause-time :send_packet reschedules via timer rather than broadcasting" do
|
|
{:ok, state} = PacketReplay.init(user_id: "paused", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
state = %{state | paused: true}
|
|
packet = %{received_at: DateTime.utc_now()}
|
|
stream = Stream.repeatedly(fn -> {0.0, packet} end)
|
|
|
|
assert {:noreply, new_state} =
|
|
PacketReplay.handle_info({:send_packet, packet, stream}, state)
|
|
|
|
# A new timer ref should be set for rescheduling.
|
|
assert is_reference(new_state.replay_timer)
|
|
Process.cancel_timer(new_state.replay_timer)
|
|
end
|
|
end
|
|
|
|
describe "terminate/2" do
|
|
test "cancels pending timer and returns :ok" do
|
|
{:ok, state} = PacketReplay.init(user_id: "terminated", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
timer_ref = Process.send_after(self(), :noop, 60_000)
|
|
state = %{state | replay_timer: timer_ref}
|
|
|
|
assert :ok = PacketReplay.terminate(:shutdown, state)
|
|
# Timer should have been cancelled.
|
|
assert Process.read_timer(timer_ref) == false
|
|
end
|
|
|
|
test "terminate is idempotent with no pending timer" do
|
|
{:ok, state} = PacketReplay.init(user_id: "noterm", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
assert :ok = PacketReplay.terminate(:normal, state)
|
|
end
|
|
end
|
|
|
|
describe "handle_call/3 extra branches" do
|
|
setup do
|
|
{:ok, state} = PacketReplay.init(user_id: "extra", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
{:ok, state: state}
|
|
end
|
|
|
|
test "pause with no active timer flips paused flag", %{state: state} do
|
|
state = %{state | replay_timer: nil}
|
|
assert {:reply, :ok, new_state} = PacketReplay.handle_call(:pause, self(), state)
|
|
assert new_state.paused
|
|
assert is_nil(new_state.replay_timer)
|
|
end
|
|
|
|
test "update_filters accepts valid bounds format", %{state: state} do
|
|
filters = [bounds: [-74.0, 40.0, -73.0, 41.0]]
|
|
|
|
assert {:reply, :ok, new_state} =
|
|
PacketReplay.handle_call({:update_filters, filters}, self(), state)
|
|
|
|
assert new_state.bounds == [-74.0, 40.0, -73.0, 41.0]
|
|
end
|
|
|
|
test "update_filters with nil bounds keeps existing", %{state: state} do
|
|
filters = [bounds: nil]
|
|
|
|
assert {:reply, :ok, new_state} =
|
|
PacketReplay.handle_call({:update_filters, filters}, self(), state)
|
|
|
|
# nil bounds → filter branch keeps the old bounds via state.bounds.
|
|
# After Map.put, new_state.bounds is nil, then the guard checks
|
|
# is_nil(bounds) and keeps new_state as-is (so bounds stays nil).
|
|
assert is_nil(new_state.bounds) or new_state.bounds == state.bounds
|
|
end
|
|
|
|
test "update_filters cancels an active timer", %{state: state} do
|
|
timer = Process.send_after(self(), :noop, 60_000)
|
|
state = %{state | replay_timer: timer}
|
|
|
|
assert {:reply, :ok, new_state} =
|
|
PacketReplay.handle_call({:update_filters, [callsign: "X"]}, self(), state)
|
|
|
|
# Timer should be gone from state and cancelled.
|
|
assert is_nil(new_state.replay_timer)
|
|
end
|
|
end
|
|
|
|
describe "handle_info({:send_packet, packet, stream}) active path" do
|
|
test "broadcasts a packet, increments counter, and stops when stream is empty" do
|
|
{:ok, state} = PacketReplay.init(user_id: "send_empty", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
packet = %Aprsme.Packet{
|
|
sender: "SENDPKT1",
|
|
base_callsign: "SENDPKT1",
|
|
ssid: "0",
|
|
received_at: DateTime.utc_now(),
|
|
has_position: true
|
|
}
|
|
|
|
empty_stream = Stream.unfold(nil, fn _ -> nil end)
|
|
|
|
assert {:stop, :normal, new_state} =
|
|
PacketReplay.handle_info({:send_packet, packet, empty_stream}, state)
|
|
|
|
assert new_state.packets_sent == 1
|
|
end
|
|
|
|
test "broadcasts and schedules next when stream has more packets" do
|
|
{:ok, state} = PacketReplay.init(user_id: "send_more", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
packet = %Aprsme.Packet{
|
|
sender: "SENDPKT2",
|
|
base_callsign: "SENDPKT2",
|
|
ssid: "0",
|
|
received_at: DateTime.utc_now(),
|
|
has_position: true
|
|
}
|
|
|
|
next_packet = %Aprsme.Packet{
|
|
sender: "NEXT",
|
|
base_callsign: "NEXT",
|
|
ssid: "0",
|
|
received_at: DateTime.add(DateTime.utc_now(), 1, :second),
|
|
has_position: true
|
|
}
|
|
|
|
# Simple stream: one element, then done.
|
|
stream =
|
|
Stream.unfold([{0.0, next_packet}], fn
|
|
[h | t] -> {h, t}
|
|
[] -> nil
|
|
end)
|
|
|
|
assert {:noreply, new_state} =
|
|
PacketReplay.handle_info({:send_packet, packet, stream}, state)
|
|
|
|
assert new_state.packets_sent == 1
|
|
assert is_reference(new_state.replay_timer)
|
|
Process.cancel_timer(new_state.replay_timer)
|
|
end
|
|
end
|
|
|
|
describe "handle_info(:start_replay, ...)" do
|
|
test "broadcasts replay_started then stops when no matching packets" do
|
|
{:ok, state} = PacketReplay.init(user_id: "empty_stream", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
# With no seeded packets in the DB, stream_packets_for_replay returns an
|
|
# empty stream and handle_info :start_replay should stop the server.
|
|
assert {:stop, :normal, _state} = PacketReplay.handle_info(:start_replay, state)
|
|
end
|
|
end
|
|
|
|
describe "sanitize_packet_for_transport via :send_packet broadcast" do
|
|
test "sanitizes DateTime, NaiveDateTime, Decimal, list, and struct fields" do
|
|
{:ok, state} = PacketReplay.init(user_id: "sanitize", bounds: [0, 0, 1, 1])
|
|
|
|
receive do
|
|
:start_replay -> :ok
|
|
after
|
|
0 -> :ok
|
|
end
|
|
|
|
# Construct a packet with a wide mix of value types so each sanitize_value
|
|
# clause runs at least once during the broadcast.
|
|
naive = NaiveDateTime.utc_now()
|
|
dt = DateTime.utc_now()
|
|
|
|
packet = %Aprsme.Packet{
|
|
sender: "SANITIZE-1",
|
|
base_callsign: "SANITIZE-1",
|
|
ssid: "1",
|
|
received_at: dt,
|
|
# Decimal field — exercises sanitize_value(%Decimal{}).
|
|
lat: Decimal.new("33.0"),
|
|
lon: Decimal.new("-96.5"),
|
|
has_position: true,
|
|
# Map containing a list and a NaiveDateTime nested under data.
|
|
data: %{
|
|
"items" => ["a", "b", "c"],
|
|
"naive" => naive,
|
|
"nested_struct" => %URI{scheme: "https"}
|
|
}
|
|
}
|
|
|
|
empty_stream = Stream.unfold(nil, fn _ -> nil end)
|
|
|
|
assert {:stop, :normal, new_state} =
|
|
PacketReplay.handle_info({:send_packet, packet, empty_stream}, state)
|
|
|
|
assert new_state.packets_sent == 1
|
|
end
|
|
end
|
|
end
|