diff --git a/test/aprs/packet_replay_test.exs b/test/aprs/packet_replay_test.exs new file mode 100644 index 0000000..3311852 --- /dev/null +++ b/test/aprs/packet_replay_test.exs @@ -0,0 +1,725 @@ +defmodule Aprs.PacketReplayTest do + use Aprs.DataCase, async: true + + import Mox + + alias Aprs.PacketReplay + alias AprsWeb.Endpoint + alias Phoenix.Socket.Broadcast + + setup :verify_on_exit! + + setup do + # Mock the Packets module functions + Mox.defmock(MockPackets, for: Aprs.PacketsBehaviour) + Application.put_env(:aprs, :packets_module, MockPackets) + + # Start a test registry for the packet replay processes + start_supervised!({Registry, keys: :unique, name: Aprs.ReplayRegistry}) + + # Example test data + user_id = "test_user_#{:rand.uniform(10_000)}" + # New York area + bounds = [-74.0, 40.0, -73.0, 41.0] + + mock_packet = %{ + id: "test_packet_1", + received_at: DateTime.utc_now(), + lat: 40.5, + lon: -73.5, + sender: "N0CALL", + data_type: "position" + } + + %{ + user_id: user_id, + bounds: bounds, + mock_packet: mock_packet + } + end + + describe "start_replay/1" do + test "starts replay with required options", %{user_id: user_id, bounds: bounds} do + opts = [user_id: user_id, bounds: bounds] + + assert {:ok, pid} = PacketReplay.start_replay(opts) + assert is_pid(pid) + assert Process.alive?(pid) + + # Clean up + GenServer.stop(pid) + end + + test "raises error when bounds are missing", %{user_id: user_id} do + opts = [user_id: user_id] + + assert_raise ArgumentError, "Map bounds are required for packet replay", fn -> + PacketReplay.start_replay(opts) + end + end + + test "starts replay with all optional parameters", %{user_id: user_id, bounds: bounds} do + opts = [ + user_id: user_id, + bounds: bounds, + callsign: "N0CALL", + start_time: DateTime.add(DateTime.utc_now(), -3600, :second), + end_time: DateTime.utc_now(), + replay_speed: 2.0, + limit: 1000, + with_position: true + ] + + assert {:ok, pid} = PacketReplay.start_replay(opts) + assert Process.alive?(pid) + + # Clean up + GenServer.stop(pid) + end + + test "fails when user_id is missing", %{bounds: bounds} do + opts = [bounds: bounds] + + assert_raise KeyError, fn -> + PacketReplay.start_replay(opts) + end + end + + test "registers process with unique name per user", %{bounds: bounds} do + user_id1 = "user1" + user_id2 = "user2" + + {:ok, pid1} = PacketReplay.start_replay(user_id: user_id1, bounds: bounds) + {:ok, pid2} = PacketReplay.start_replay(user_id: user_id2, bounds: bounds) + + assert pid1 != pid2 + assert Process.alive?(pid1) + assert Process.alive?(pid2) + + # Clean up + GenServer.stop(pid1) + GenServer.stop(pid2) + end + end + + describe "stop_replay/1" do + test "stops an active replay session", %{user_id: user_id, bounds: bounds} do + {:ok, pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + assert Process.alive?(pid) + + assert :ok = PacketReplay.stop_replay(user_id) + + # Give it a moment to stop + Process.sleep(10) + refute Process.alive?(pid) + end + + test "returns error when no replay session exists" do + assert {:error, :not_found} = PacketReplay.stop_replay("nonexistent_user") + end + end + + describe "pause_replay/1" do + test "pauses an active replay session", %{user_id: user_id, bounds: bounds} do + {:ok, _pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + assert :ok = PacketReplay.pause_replay(user_id) + + # Verify state + info = PacketReplay.get_replay_info(user_id) + assert info.paused == true + + # Clean up + PacketReplay.stop_replay(user_id) + end + + test "fails when replay session doesn't exist" do + assert_raise RuntimeError, fn -> + PacketReplay.pause_replay("nonexistent_user") + end + end + end + + describe "resume_replay/1" do + test "resumes a paused replay session", %{user_id: user_id, bounds: bounds} do + {:ok, _pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + # First pause it + :ok = PacketReplay.pause_replay(user_id) + info = PacketReplay.get_replay_info(user_id) + assert info.paused == true + + # Then resume it + assert :ok = PacketReplay.resume_replay(user_id) + + info = PacketReplay.get_replay_info(user_id) + assert info.paused == false + + # Clean up + PacketReplay.stop_replay(user_id) + end + + test "acknowledges resume on already running session", %{user_id: user_id, bounds: bounds} do + {:ok, _pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + # Resume without pausing first + assert :ok = PacketReplay.resume_replay(user_id) + + info = PacketReplay.get_replay_info(user_id) + assert info.paused == false + + # Clean up + PacketReplay.stop_replay(user_id) + end + end + + describe "set_replay_speed/2" do + test "sets replay speed", %{user_id: user_id, bounds: bounds} do + {:ok, _pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + assert :ok = PacketReplay.set_replay_speed(user_id, 10.0) + + info = PacketReplay.get_replay_info(user_id) + assert info.replay_speed == 10.0 + + # Clean up + PacketReplay.stop_replay(user_id) + end + + 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 -> + PacketReplay.set_replay_speed(user_id, "invalid") + end + end + end + + describe "update_filters/2" do + test "updates filters and restarts replay", %{user_id: user_id, bounds: bounds} do + {:ok, _pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + new_filters = [callsign: "N1CALL", limit: 2000] + assert :ok = PacketReplay.update_filters(user_id, new_filters) + + info = PacketReplay.get_replay_info(user_id) + assert info.callsign == "N1CALL" + + # Clean up + PacketReplay.stop_replay(user_id) + end + + test "validates bounds format in filters", %{user_id: user_id, bounds: bounds} do + {:ok, _pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + # Valid bounds update + new_bounds = [-75.0, 39.0, -72.0, 42.0] + assert :ok = PacketReplay.update_filters(user_id, bounds: new_bounds) + + info = PacketReplay.get_replay_info(user_id) + assert info.bounds == new_bounds + + # Invalid bounds should keep old bounds + assert :ok = PacketReplay.update_filters(user_id, bounds: "invalid") + + info = PacketReplay.get_replay_info(user_id) + # Should keep the valid bounds + assert info.bounds == new_bounds + + # Clean up + PacketReplay.stop_replay(user_id) + end + + test "validates filters must be a list" do + user_id = "test_user" + + assert_raise FunctionClauseError, fn -> + PacketReplay.update_filters(user_id, %{callsign: "N0CALL"}) + end + end + end + + describe "get_replay_info/1" do + test "returns replay session information", %{user_id: user_id, bounds: bounds} do + 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: 3.0, + limit: 2500, + with_position: true + ] + + {:ok, _pid} = PacketReplay.start_replay(opts) + + info = PacketReplay.get_replay_info(user_id) + + assert info.user_id == user_id + assert info.bounds == bounds + assert info.callsign == "N0CALL" + assert info.start_time == start_time + assert info.end_time == end_time + assert info.replay_speed == 3.0 + assert info.with_position == true + assert info.packets_sent == 0 + assert info.paused == false + assert %DateTime{} = info.replay_started_at + + # Clean up + PacketReplay.stop_replay(user_id) + end + end + + describe "init/1" do + test "initializes with default values", %{user_id: user_id, bounds: bounds} do + opts = [user_id: user_id, bounds: bounds] + + assert {:ok, state} = PacketReplay.init(opts) + + assert state.user_id == user_id + assert state.bounds == bounds + # default + assert state.replay_speed == 5.0 + # default + assert state.limit == 5000 + # default + 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", %{user_id: user_id, bounds: bounds} do + 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", %{user_id: user_id, bounds: bounds} do + # Try to set start time to 2 hours ago + 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) + + # Should be limited to 1 hour ago (3600 seconds) + 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", %{user_id: user_id, bounds: bounds} do + opts = [user_id: user_id, bounds: bounds] + + assert {:ok, state} = PacketReplay.init(opts) + + assert state.replay_topic == "replay:#{user_id}" + end + end + + describe "handle_info(:start_replay)" do + test "handles start_replay message with mock packets", %{mock_packet: mock_packet} do + # Set up mock expectations + MockPackets + |> expect(:get_historical_packet_count, fn _opts -> 1 end) + |> expect(:stream_packets_for_replay, fn _opts -> + Stream.unfold([{0.1, mock_packet}], fn + [packet | rest] -> {packet, rest} + [] -> nil + end) + end) + + user_id = "test_user" + bounds = [-74.0, 40.0, -73.0, 41.0] + + state = %{ + user_id: user_id, + replay_topic: "replay:#{user_id}", + replay_speed: 5.0, + start_time: DateTime.add(DateTime.utc_now(), -3600, :second), + end_time: DateTime.utc_now(), + region: nil, + bounds: bounds, + callsign: nil, + with_position: true, + limit: 5000, + paused: false, + packets_sent: 0, + replay_started_at: DateTime.utc_now(), + replay_timer: nil, + last_packet_time: nil + } + + # Subscribe to the broadcast topic to verify messages + Endpoint.subscribe("replay:#{user_id}") + + result = PacketReplay.handle_info(:start_replay, state) + + assert {:noreply, new_state} = result + assert new_state.last_packet_time == mock_packet.received_at + assert is_reference(new_state.replay_timer) + + # Should have received broadcast + assert_receive %Broadcast{ + topic: "replay:" <> _, + event: "replay_started" + } + end + + test "handles empty packet stream", %{user_id: user_id, bounds: bounds} do + # Set up mock expectations for empty stream + MockPackets + |> expect(:get_historical_packet_count, fn _opts -> 0 end) + |> expect(:stream_packets_for_replay, fn _opts -> + Stream.unfold([], fn + [] -> nil + end) + end) + + state = %{ + user_id: user_id, + replay_topic: "replay:#{user_id}", + replay_speed: 5.0, + start_time: DateTime.add(DateTime.utc_now(), -3600, :second), + end_time: DateTime.utc_now(), + region: nil, + bounds: bounds, + callsign: nil, + with_position: true, + limit: 5000, + paused: false, + packets_sent: 0, + replay_started_at: DateTime.utc_now(), + replay_timer: nil, + last_packet_time: nil + } + + # Subscribe to the broadcast topic + Endpoint.subscribe("replay:#{user_id}") + + result = PacketReplay.handle_info(:start_replay, state) + + assert {:stop, :normal, ^state} = result + + # Should have received completion broadcast + assert_receive %Broadcast{ + topic: "replay:" <> _, + event: "replay_complete", + payload: %{packets_sent: 0, message: "No matching packets found for replay"} + } + end + end + + describe "handle_info({:send_packet, packet, stream})" do + test "reschedules packet when paused", %{mock_packet: mock_packet} do + stream = Stream.unfold([], fn _ -> nil end) + + state = %{ + user_id: "test_user", + replay_topic: "replay:test_user", + paused: true, + replay_timer: nil, + packets_sent: 0, + last_packet_time: nil + } + + result = PacketReplay.handle_info({:send_packet, mock_packet, stream}, state) + + assert {:noreply, new_state} = result + assert is_reference(new_state.replay_timer) + end + + test "broadcasts packet and schedules next when not paused", %{mock_packet: mock_packet} do + next_packet = %{mock_packet | id: "test_packet_2"} + + stream = + Stream.unfold([{0.1, next_packet}], fn + [packet | rest] -> {packet, rest} + [] -> nil + end) + + state = %{ + user_id: "test_user", + replay_topic: "replay:test_user", + paused: false, + replay_timer: nil, + packets_sent: 0, + last_packet_time: nil + } + + # Subscribe to broadcasts + Endpoint.subscribe("replay:test_user") + + result = PacketReplay.handle_info({:send_packet, mock_packet, stream}, state) + + assert {:noreply, new_state} = result + assert new_state.packets_sent == 1 + assert new_state.last_packet_time == next_packet.received_at + assert is_reference(new_state.replay_timer) + + # Should have received packet broadcast + assert_receive %Broadcast{ + event: "historical_packet", + payload: %{is_historical: true} + } + end + + test "completes replay when no more packets", %{mock_packet: mock_packet} do + # Empty stream - no more packets + stream = Stream.unfold([], fn _ -> nil end) + + state = %{ + user_id: "test_user", + replay_topic: "replay:test_user", + paused: false, + replay_timer: nil, + packets_sent: 5, + last_packet_time: nil + } + + # Subscribe to broadcasts + Endpoint.subscribe("replay:test_user") + + result = PacketReplay.handle_info({:send_packet, mock_packet, stream}, state) + + assert {:stop, :normal, new_state} = result + assert new_state.packets_sent == 6 + + # Should have received completion broadcast + assert_receive %Broadcast{ + event: "replay_complete", + payload: %{packets_sent: 6, message: "Replay complete"} + } + end + end + + describe "sanitize_packet_for_transport/1" do + test "converts packet to JSON-safe format" do + packet = %{ + __meta__: %{}, + __struct__: SomeStruct, + id: "test", + received_at: ~U[2023-01-01 12:00:00Z], + lat: Decimal.new("40.123456"), + nested: %{ + timestamp: ~N[2023-01-01 12:00:00], + value: Decimal.new("1.23") + }, + list: [~U[2023-01-01 12:00:00Z], Decimal.new("4.56")] + } + + result = PacketReplay.sanitize_packet_for_transport(packet) + + # Should drop meta fields + refute Map.has_key?(result, :__meta__) + refute Map.has_key?(result, :__struct__) + + # Should preserve normal fields + assert result.id == "test" + + # Should convert DateTime to ISO string + assert result.received_at == "2023-01-01T12:00:00Z" + + # Should convert Decimal to float + assert result.lat == 40.123456 + + # Should handle nested structures + assert result.nested.timestamp == "2023-01-01T12:00:00" + assert result.nested.value == 1.23 + + # Should handle lists + assert result.list == ["2023-01-01T12:00:00Z", 4.56] + end + + test "handles packets without special types" do + packet = %{ + id: "simple", + name: "test", + count: 42, + active: true + } + + result = PacketReplay.sanitize_packet_for_transport(packet) + + assert result == %{ + id: "simple", + name: "test", + count: 42, + active: true + } + end + end + + describe "via_tuple/1" do + test "generates correct registry tuple" do + user_id = "test_user_123" + + result = PacketReplay.via_tuple(user_id) + + assert result == {:via, Registry, {Aprs.ReplayRegistry, "replay:test_user_123"}} + end + end + + describe "terminate/2" do + test "cleans up timers and broadcasts stop message" do + timer_ref = make_ref() + + state = %{ + user_id: "test_user", + replay_topic: "replay:test_user", + replay_timer: timer_ref, + packets_sent: 10 + } + + # Subscribe to broadcasts + Endpoint.subscribe("replay:test_user") + + result = PacketReplay.terminate(:normal, state) + + assert result == :ok + + # Should have received stop broadcast + assert_receive %Broadcast{ + event: "replay_stopped", + payload: %{packets_sent: 10, message: "Replay stopped"} + } + end + + test "handles state without timer" do + state = %{ + user_id: "test_user", + replay_topic: "replay:test_user", + replay_timer: nil, + packets_sent: 0 + } + + # Subscribe to broadcasts + Endpoint.subscribe("replay:test_user") + + result = PacketReplay.terminate(:shutdown, state) + + assert result == :ok + + # Should still broadcast stop message + assert_receive %Broadcast{ + event: "replay_stopped" + } + end + end + + describe "integration tests" do + test "full replay lifecycle", %{user_id: user_id, bounds: bounds, mock_packet: mock_packet} do + # Set up mock for full lifecycle + MockPackets + |> expect(:get_historical_packet_count, fn _opts -> 1 end) + |> expect(:stream_packets_for_replay, fn _opts -> + # Single packet stream + Stream.unfold([{0.001, mock_packet}], fn + [packet | rest] -> {packet, rest} + [] -> nil + end) + end) + + # Start replay + {:ok, pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + # Subscribe to events + Endpoint.subscribe("replay:#{user_id}") + + # Should receive start event + assert_receive %Broadcast{ + event: "replay_started" + }, + 1000 + + # Should receive packet event quickly (1ms delay) + assert_receive %Broadcast{ + event: "historical_packet" + }, + 1000 + + # Should receive completion event + assert_receive %Broadcast{ + event: "replay_complete" + }, + 1000 + + # Process should terminate + Process.sleep(50) + refute Process.alive?(pid) + end + + test "pause and resume functionality", %{user_id: user_id, bounds: bounds} do + {:ok, pid} = PacketReplay.start_replay(user_id: user_id, bounds: bounds) + + # Subscribe to events + Endpoint.subscribe("replay:#{user_id}") + + # Pause + :ok = PacketReplay.pause_replay(user_id) + + assert_receive %Broadcast{ + event: "replay_paused" + } + + info = PacketReplay.get_replay_info(user_id) + assert info.paused == true + + # Resume + :ok = PacketReplay.resume_replay(user_id) + + assert_receive %Broadcast{ + event: "replay_resumed" + } + + info = PacketReplay.get_replay_info(user_id) + assert info.paused == false + + # Clean up + GenServer.stop(pid) + end + end +end