From 094a50d1839bbcaf0f09ea1883245c1f67d2d98a Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 9 Feb 2026 17:19:50 -0600 Subject: [PATCH] Add tests for 14 modules with 0% coverage Adds test files for: - AprsIsConnection (83% coverage) - Cluster.PacketDistributor (100%) - Cluster.PacketReceiver (100%) - Cluster.Topology (100%) - ConnectionMonitor (78%) - DbOptimizer (90%) - DynamicSupervisor (100%) - ErrorHandler (89%) - MigrationLock (90%) - MockHelpers (100%) - PacketConsumerPool (90%) - PacketCounter (100%) - PacketPipelineSetup (75%) - PacketPipelineSupervisor (100%) Uses stub repo modules to test MigrationLock lock contention paths without real multi-session PostgreSQL advisory lock contention. --- test/aprsme/aprs_is_connection_test.exs | 204 +++++++++++++ .../cluster/packet_distributor_test.exs | 91 ++++++ test/aprsme/cluster/packet_receiver_test.exs | 108 +++++++ test/aprsme/cluster/topology_test.exs | 51 ++++ test/aprsme/connection_monitor_test.exs | 145 +++++++++ test/aprsme/db_optimizer_test.exs | 214 +++++++++++++ test/aprsme/dynamic_supervisor_test.exs | 88 ++++++ test/aprsme/error_handler_test.exs | 289 ++++++++++++++++++ test/aprsme/migration_lock_test.exs | 179 +++++++++++ test/aprsme/mock_helpers_test.exs | 22 ++ test/aprsme/packet_consumer_pool_test.exs | 84 +++++ test/aprsme/packet_counter_test.exs | 40 +++ test/aprsme/packet_pipeline_setup_test.exs | 99 ++++++ .../packet_pipeline_supervisor_test.exs | 125 ++++++++ 14 files changed, 1739 insertions(+) create mode 100644 test/aprsme/aprs_is_connection_test.exs create mode 100644 test/aprsme/cluster/packet_distributor_test.exs create mode 100644 test/aprsme/cluster/packet_receiver_test.exs create mode 100644 test/aprsme/cluster/topology_test.exs create mode 100644 test/aprsme/connection_monitor_test.exs create mode 100644 test/aprsme/db_optimizer_test.exs create mode 100644 test/aprsme/dynamic_supervisor_test.exs create mode 100644 test/aprsme/error_handler_test.exs create mode 100644 test/aprsme/migration_lock_test.exs create mode 100644 test/aprsme/mock_helpers_test.exs create mode 100644 test/aprsme/packet_consumer_pool_test.exs create mode 100644 test/aprsme/packet_counter_test.exs create mode 100644 test/aprsme/packet_pipeline_setup_test.exs create mode 100644 test/aprsme/packet_pipeline_supervisor_test.exs diff --git a/test/aprsme/aprs_is_connection_test.exs b/test/aprsme/aprs_is_connection_test.exs new file mode 100644 index 0000000..26e7054 --- /dev/null +++ b/test/aprsme/aprs_is_connection_test.exs @@ -0,0 +1,204 @@ +defmodule Aprsme.AprsIsConnectionTest do + use ExUnit.Case, async: false + + alias Aprsme.AprsIsConnection + + setup do + # Stop any existing AprsIsConnection process + case Process.whereis(AprsIsConnection) do + nil -> :ok + pid -> GenServer.stop(pid, :normal, 5000) + end + + # Ensure disable_aprs_connection is true so we don't try real connections + original_env = Application.get_env(:aprsme, :disable_aprs_connection) + Application.put_env(:aprsme, :disable_aprs_connection, true) + + on_exit(fn -> + # Stop the process if it's still running + case Process.whereis(AprsIsConnection) do + nil -> + :ok + + pid -> + try do + GenServer.stop(pid, :normal, 5000) + catch + :exit, _ -> :ok + end + end + + # Reset env + if original_env do + Application.put_env(:aprsme, :disable_aprs_connection, original_env) + else + Application.delete_env(:aprsme, :disable_aprs_connection) + end + end) + + :ok + end + + describe "init/1 when disabled" do + test "starts with socket: nil" do + {:ok, pid} = AprsIsConnection.start_link([]) + state = :sys.get_state(pid) + + assert state.socket == nil + end + end + + describe "handle_info {:tcp, socket, data}" do + test "broadcasts received data via PubSub" do + {:ok, pid} = AprsIsConnection.start_link([]) + Phoenix.PubSub.subscribe(Aprsme.PubSub, "aprs_is:raw") + + send(pid, {:tcp, make_ref(), "test data"}) + + assert_receive {:aprsme_is_line, "test data"} + end + end + + describe "handle_info {:tcp_closed, socket}" do + test "sets socket to nil in state" do + {:ok, pid} = AprsIsConnection.start_link([]) + + send(pid, {:tcp_closed, make_ref()}) + + # Allow the message to be processed + :sys.get_state(pid) + state = :sys.get_state(pid) + + assert state.socket == nil + end + end + + describe "handle_info {:tcp_error, socket, reason}" do + test "sets socket to nil in state" do + {:ok, pid} = AprsIsConnection.start_link([]) + + send(pid, {:tcp_error, make_ref(), :etimedout}) + + # Allow the message to be processed + :sys.get_state(pid) + state = :sys.get_state(pid) + + assert state.socket == nil + end + end + + describe "handle_call {:send, packet} without socket" do + test "returns {:error, :not_connected}" do + {:ok, pid} = AprsIsConnection.start_link([]) + + assert GenServer.call(pid, {:send, "test"}) == {:error, :not_connected} + end + end + + describe "handle_call {:send, packet} with socket" do + test "sends data over the socket and returns :ok" do + {:ok, pid} = AprsIsConnection.start_link([]) + + {:ok, listen} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) + {:ok, port} = :inet.port(listen) + {:ok, client} = :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false]) + {:ok, server} = :gen_tcp.accept(listen) + + :sys.replace_state(pid, fn state -> %{state | socket: client} end) + + assert GenServer.call(pid, {:send, "test packet"}) == :ok + + {:ok, data} = :gen_tcp.recv(server, 0, 1000) + assert data =~ "test packet" + + :gen_tcp.close(server) + :gen_tcp.close(client) + :gen_tcp.close(listen) + end + end + + describe "terminate/2 with socket" do + test "closes the socket without crashing" do + {:ok, pid} = AprsIsConnection.start_link([]) + + {:ok, listen} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) + {:ok, port} = :inet.port(listen) + {:ok, client} = :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false]) + {:ok, _server} = :gen_tcp.accept(listen) + + :sys.replace_state(pid, fn state -> %{state | socket: client} end) + + assert GenServer.stop(pid) == :ok + + :gen_tcp.close(listen) + end + end + + describe "send_packet/1" do + test "delegates to GenServer.call with {:send, packet}" do + {:ok, _pid} = AprsIsConnection.start_link([]) + + # Without a socket it returns not_connected + assert AprsIsConnection.send_packet("test") == {:error, :not_connected} + end + end + + describe "handle_call {:send, packet} with closed socket" do + test "returns error when send fails on a closed socket" do + {:ok, pid} = AprsIsConnection.start_link([]) + + # Create a socket then close it to simulate a broken connection + {:ok, listen} = :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true]) + {:ok, port} = :inet.port(listen) + {:ok, client} = :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false]) + {:ok, _server} = :gen_tcp.accept(listen) + + :gen_tcp.close(client) + + :sys.replace_state(pid, fn state -> %{state | socket: client} end) + + assert {:error, _reason} = GenServer.call(pid, {:send, "test packet"}) + + # State should have socket set to nil after send failure + state = :sys.get_state(pid) + assert state.socket == nil + + :gen_tcp.close(listen) + end + end + + describe "terminate/2 without socket" do + test "returns :ok when socket is nil" do + {:ok, pid} = AprsIsConnection.start_link([]) + + # Socket is nil by default in disabled mode + assert GenServer.stop(pid) == :ok + end + end + + describe "handle_info :connect" do + test "handles connection attempt with circuit breaker" do + {:ok, pid} = AprsIsConnection.start_link([]) + + # Send :connect message - it will fail since we can't connect to APRS-IS in tests + # but the process should handle the error gracefully + send(pid, :connect) + + # Allow the message to be processed + Process.sleep(200) + + # Process should still be alive after failed connection attempt + assert Process.alive?(pid) + state = :sys.get_state(pid) + assert state.socket == nil + end + end + + describe "code_change/3" do + test "returns {:ok, state}" do + state = %{socket: nil, backoff: 2000} + + assert AprsIsConnection.code_change(:old, state, :extra) == {:ok, state} + end + end +end diff --git a/test/aprsme/cluster/packet_distributor_test.exs b/test/aprsme/cluster/packet_distributor_test.exs new file mode 100644 index 0000000..67dd1b6 --- /dev/null +++ b/test/aprsme/cluster/packet_distributor_test.exs @@ -0,0 +1,91 @@ +defmodule Aprsme.Cluster.PacketDistributorTest do + use ExUnit.Case, async: false + + alias Aprsme.Cluster.LeaderElection + alias Aprsme.Cluster.PacketDistributor + + @test_packet %{raw: "TEST>APRS:test packet", sender: "TEST"} + + setup do + # Ensure cluster_enabled is false initially + Application.put_env(:aprsme, :cluster_enabled, false) + + # Stop any running LeaderElection process + if Process.whereis(LeaderElection) do + try do + GenServer.stop(LeaderElection, :normal, 100) + catch + :exit, _ -> :ok + end + end + + # Start LeaderElection in non-clustered mode + {:ok, _} = LeaderElection.start_link([]) + # Wait for election to complete (100ms timer + processing) + Process.sleep(300) + + on_exit(fn -> + try do + if Process.whereis(LeaderElection) do + GenServer.stop(LeaderElection, :normal, 100) + end + catch + :exit, _ -> :ok + end + + # Reset cluster_enabled to false + Application.put_env(:aprsme, :cluster_enabled, false) + end) + + :ok + end + + describe "distribute_packet/1" do + test "does not distribute when cluster is disabled" do + # cluster_enabled is false from setup + PacketDistributor.subscribe() + + PacketDistributor.distribute_packet(@test_packet) + + refute_receive {:distributed_packet, _}, 200 + end + + test "distributes packet when cluster is enabled and node is leader" do + Application.put_env(:aprsme, :cluster_enabled, true) + # LeaderElection started in non-clustered mode becomes leader quickly + Process.sleep(300) + + PacketDistributor.subscribe() + + PacketDistributor.distribute_packet(@test_packet) + + assert_receive {:distributed_packet, packet}, 500 + assert packet.raw == "TEST>APRS:test packet" + assert packet.sender == "TEST" + end + end + + describe "subscribe/0" do + test "subscribes to cluster:packets topic" do + PacketDistributor.subscribe() + + Phoenix.PubSub.broadcast( + Aprsme.PubSub, + "cluster:packets", + {:distributed_packet, @test_packet} + ) + + assert_receive {:distributed_packet, _}, 500 + end + end + + describe "handle_distributed_packet/1" do + test "processes a distributed packet without crashing" do + # StreamingPacketsPubSub and PacketStore are already running in test + result = PacketDistributor.handle_distributed_packet({:distributed_packet, @test_packet}) + + # The function logs and returns :ok from Logger.debug + assert result == :ok + end + end +end diff --git a/test/aprsme/cluster/packet_receiver_test.exs b/test/aprsme/cluster/packet_receiver_test.exs new file mode 100644 index 0000000..e79dfdd --- /dev/null +++ b/test/aprsme/cluster/packet_receiver_test.exs @@ -0,0 +1,108 @@ +defmodule Aprsme.Cluster.PacketReceiverTest do + use ExUnit.Case, async: false + + alias Aprsme.Cluster.LeaderElection + alias Aprsme.Cluster.PacketReceiver + + @election_key {:aprs_is_leader, LeaderElection} + + setup do + # Clean up any existing global registrations + :global.unregister_name(@election_key) + + # Ensure clustering is disabled so LeaderElection becomes leader + Application.put_env(:aprsme, :cluster_enabled, false) + + # Stop any existing PacketReceiver + if Process.whereis(PacketReceiver) do + GenServer.stop(PacketReceiver) + end + + # Stop any existing LeaderElection + if Process.whereis(LeaderElection) do + GenServer.stop(LeaderElection) + end + + # Start LeaderElection and wait for election + {:ok, leader_pid} = LeaderElection.start_link([]) + Process.sleep(300) + + on_exit(fn -> + try do + if Process.whereis(PacketReceiver), do: GenServer.stop(PacketReceiver, :normal, 100) + catch + :exit, _ -> :ok + end + + try do + if Process.whereis(LeaderElection), do: GenServer.stop(LeaderElection, :normal, 100) + catch + :exit, _ -> :ok + end + + :global.unregister_name(@election_key) + Application.put_env(:aprsme, :cluster_enabled, false) + end) + + {:ok, leader_pid: leader_pid} + end + + describe "start_link/1" do + test "starts and registers the process" do + assert {:ok, pid} = PacketReceiver.start_link([]) + assert Process.alive?(pid) + assert Process.whereis(PacketReceiver) == pid + end + end + + describe "handle_info/2 when leader" do + test "does not forward distributed packets when node is leader" do + {:ok, pid} = PacketReceiver.start_link([]) + + # Confirm we are the leader + assert LeaderElection.leader?() == true + + packet = %{raw: "TEST>APRS:test", sender: "TEST"} + send(pid, {:distributed_packet, packet}) + + # Give it time to process + Process.sleep(100) + + # Process should still be alive (no crash) + assert Process.alive?(pid) + end + end + + describe "handle_info/2 when not leader" do + test "forwards distributed packets when node is not leader" do + {:ok, pid} = PacketReceiver.start_link([]) + + # Force non-leader state + :sys.replace_state(LeaderElection, fn state -> %{state | is_leader: false} end) + + assert LeaderElection.leader?() == false + + packet = %{raw: "TEST>APRS:test", sender: "TEST"} + send(pid, {:distributed_packet, packet}) + + # Give it time to process + Process.sleep(100) + + # Process should still be alive (no crash) + assert Process.alive?(pid) + end + end + + describe "handle_info/2 with unknown messages" do + test "ignores unknown messages without crashing" do + {:ok, pid} = PacketReceiver.start_link([]) + + send(pid, :some_unknown_message) + send(pid, {:unexpected, "data"}) + + Process.sleep(100) + + assert Process.alive?(pid) + end + end +end diff --git a/test/aprsme/cluster/topology_test.exs b/test/aprsme/cluster/topology_test.exs new file mode 100644 index 0000000..8674f32 --- /dev/null +++ b/test/aprsme/cluster/topology_test.exs @@ -0,0 +1,51 @@ +defmodule Aprsme.Cluster.TopologyTest do + use ExUnit.Case, async: false + + alias Aprsme.Cluster.Topology + alias Cluster.Strategy.Gossip + + setup do + on_exit(fn -> + Application.put_env(:aprsme, :cluster_enabled, false) + Application.put_env(:libcluster, :topologies, []) + end) + + :ok + end + + describe "start_link/1" do + test "returns :ignore" do + assert :ignore = Topology.start_link([]) + end + end + + describe "child_spec/1" do + test "returns no-op spec when cluster is disabled" do + Application.put_env(:aprsme, :cluster_enabled, false) + + spec = Topology.child_spec([]) + + assert %{id: Topology, start: {Topology, :start_link, [[]]}, type: :worker} = spec + end + + test "returns no-op spec when cluster is enabled but no topologies configured" do + Application.put_env(:aprsme, :cluster_enabled, true) + Application.put_env(:libcluster, :topologies, []) + + spec = Topology.child_spec([]) + + assert %{id: Topology, start: {Topology, :start_link, [[]]}, type: :worker} = spec + end + + test "returns Cluster.Supervisor tuple when cluster is enabled with valid topologies" do + Application.put_env(:aprsme, :cluster_enabled, true) + Application.put_env(:libcluster, :topologies, gossip: [strategy: Gossip]) + + spec = Topology.child_spec([]) + + assert {Cluster.Supervisor, [topologies, supervisor_opts]} = spec + assert topologies == [gossip: [strategy: Gossip]] + assert Keyword.get(supervisor_opts, :name) == Aprsme.ClusterSupervisor + end + end +end diff --git a/test/aprsme/connection_monitor_test.exs b/test/aprsme/connection_monitor_test.exs new file mode 100644 index 0000000..c532543 --- /dev/null +++ b/test/aprsme/connection_monitor_test.exs @@ -0,0 +1,145 @@ +defmodule Aprsme.ConnectionMonitorTest do + use ExUnit.Case, async: false + + alias Aprsme.ConnectionMonitor + + describe "cluster disabled (default test config)" do + test "init/1 returns :ignore when cluster disabled" do + assert ConnectionMonitor.init([]) == :ignore + end + + test "accepting_connections?/0 returns true when cluster disabled" do + assert ConnectionMonitor.accepting_connections?() == true + end + + test "get_stats/0 returns default stats when cluster disabled" do + assert ConnectionMonitor.get_stats() == %{connections: 0, cpu: 0.0, memory: 0.0} + end + end + + describe "cluster enabled" do + setup do + Application.put_env(:aprsme, :cluster_enabled, true) + {:ok, pid} = ConnectionMonitor.start_link([]) + + on_exit(fn -> + if Process.alive?(pid), do: GenServer.stop(pid) + Application.put_env(:aprsme, :cluster_enabled, false) + end) + + %{pid: pid} + end + + test "register_connection/0 increments connections" do + ConnectionMonitor.register_connection() + # Allow the cast to be processed + stats = ConnectionMonitor.get_stats() + assert stats.connections == 1 + end + + test "unregister_connection/0 decrements connections" do + ConnectionMonitor.register_connection() + ConnectionMonitor.register_connection() + ConnectionMonitor.unregister_connection() + stats = ConnectionMonitor.get_stats() + assert stats.connections == 1 + end + + test "unregister_connection/0 does not go below 0" do + ConnectionMonitor.unregister_connection() + stats = ConnectionMonitor.get_stats() + assert stats.connections == 0 + end + + test "accepting_connections?/0 returns true when not draining" do + assert ConnectionMonitor.accepting_connections?() == true + end + + test "accepting_connections?/0 returns false when draining", %{pid: pid} do + :sys.replace_state(pid, fn state -> %{state | draining: true} end) + assert ConnectionMonitor.accepting_connections?() == false + end + + test ":check_load handler keeps process alive", %{pid: pid} do + send(pid, :check_load) + # Give it time to process + _ = ConnectionMonitor.get_stats() + assert Process.alive?(pid) + end + + test "get_stats/0 returns cpu and memory fields", %{pid: _pid} do + stats = ConnectionMonitor.get_stats() + + assert Map.has_key?(stats, :cpu) + assert Map.has_key?(stats, :memory) + assert is_number(stats.cpu) + assert is_number(stats.memory) + end + + test "draining triggers when connections exceed imbalance ratio", %{pid: pid} do + # Simulate having many connections to trigger draining analysis + for _ <- 1..20 do + ConnectionMonitor.register_connection() + end + + # Force a load check which will gather stats and analyze + send(pid, :check_load) + Process.sleep(100) + + assert Process.alive?(pid) + stats = ConnectionMonitor.get_stats() + assert stats.connections == 20 + end + + test "register_connection/0 is a no-op when cluster disabled" do + Application.put_env(:aprsme, :cluster_enabled, false) + + # Should not crash even though the GenServer is running + ConnectionMonitor.register_connection() + + Application.put_env(:aprsme, :cluster_enabled, true) + end + + test "unregister_connection/0 is a no-op when cluster disabled" do + Application.put_env(:aprsme, :cluster_enabled, false) + + ConnectionMonitor.unregister_connection() + + Application.put_env(:aprsme, :cluster_enabled, true) + end + + test "draining is set to true when state has high connections", %{pid: pid} do + # Set up state with draining true and connections > 0 to trigger drain broadcast + Phoenix.PubSub.subscribe(Aprsme.PubSub, "connection:drain:#{Node.self()}") + + :sys.replace_state(pid, fn state -> + %{state | draining: true, local_connections: 50} + end) + + # Trigger check_load to run the full analysis pipeline + send(pid, :check_load) + Process.sleep(200) + + assert Process.alive?(pid) + end + + test "get_stats returns draining field", %{pid: pid} do + :sys.replace_state(pid, fn state -> %{state | draining: true} end) + stats = ConnectionMonitor.get_stats() + assert stats.draining == true + end + + test "gather_cluster_stats handles local node stats", %{pid: pid} do + ConnectionMonitor.register_connection() + ConnectionMonitor.register_connection() + + # Trigger check_load which calls gather_cluster_stats internally + send(pid, :check_load) + Process.sleep(100) + + state = :sys.get_state(pid) + # node_stats should have been populated + assert is_map(state.node_stats) + end + end +end diff --git a/test/aprsme/db_optimizer_test.exs b/test/aprsme/db_optimizer_test.exs new file mode 100644 index 0000000..4faae90 --- /dev/null +++ b/test/aprsme/db_optimizer_test.exs @@ -0,0 +1,214 @@ +defmodule Aprsme.DbOptimizerTest do + use Aprsme.DataCase, async: false + + alias Aprsme.DbOptimizer + + describe "calculate_optimal_batch_size/1" do + test "returns max cap (2000) for small maps" do + entries = [%{name: "a", value: 1}, %{name: "b", value: 2}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "uses default 1024 size when first entry is nil" do + # With nil as first entry, estimate_entry_size returns 1024 + # available_memory = 14 * 1024 * 1024 = 14_680_064 + # max_batch = div(14_680_064, 1024) = 14_336 + # capped at min(14_336, 2000) = 2000 + entries = [nil, %{name: "a"}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "returns 2000 for single-element list with small map" do + entries = [%{x: 1}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "returns minimum of 100 for extremely large entries" do + # Create a map with very large values to push the estimated size way up + # We need estimated_size > 14 * 1024 * 1024 / 100 = 146_800 + # Each binary value contributes its byte_size + 100 overhead + big_value = String.duplicate("x", 200_000) + entries = [%{data: big_value}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 100 + end + end + + describe "copy_insert/3" do + test "inserts rows when count is <= 1000 using regular_batch_insert" do + now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second) + + rows = [ + ["test_copy_small_1", 10, now, now], + ["test_copy_small_2", 20, now, now] + ] + + count = + DbOptimizer.copy_insert( + "packet_counters", + ["counter_type", "count", "inserted_at", "updated_at"], + rows + ) + + assert count == 2 + end + + test "falls back to regular_batch_insert when > 1000 rows (COPY fails in sandbox)" do + now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second) + + rows = + for i <- 1..1001 do + ["test_copy_large_#{i}", i, now, now] + end + + count = + DbOptimizer.copy_insert( + "packet_counters", + ["counter_type", "count", "inserted_at", "updated_at"], + rows + ) + + # COPY will fail in sandbox, falls back to regular_batch_insert + # Some rows may conflict on unique index but on_conflict: :nothing handles that + assert is_integer(count) + assert count > 0 + end + end + + describe "optimized_batch_insert/3" do + test "inserts valid entries and returns {count, 0}" do + now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second) + + entries = [ + %{ + "counter_type" => "test_optimized_1", + "count" => 100, + "inserted_at" => now, + "updated_at" => now + }, + %{ + "counter_type" => "test_optimized_2", + "count" => 200, + "inserted_at" => now, + "updated_at" => now + } + ] + + assert {2, 0} = DbOptimizer.optimized_batch_insert("packet_counters", entries) + end + + test "returns {0, 0} for empty list" do + assert {0, 0} = DbOptimizer.optimized_batch_insert("packet_counters", []) + end + end + + describe "analyze_table/1" do + test "returns :ok for packets table" do + assert :ok = DbOptimizer.analyze_table("packets") + end + end + + describe "vacuum_table/1" do + test "returns :error inside sandbox (VACUUM cannot run in transactions)" do + assert :error = DbOptimizer.vacuum_table("packets") + end + end + + describe "get_connection_stats/0" do + test "returns map with expected keys and integer values" do + stats = DbOptimizer.get_connection_stats() + + assert is_map(stats) + assert Map.has_key?(stats, :total) + assert Map.has_key?(stats, :active) + assert Map.has_key?(stats, :idle) + assert Map.has_key?(stats, :idle_in_transaction) + assert Map.has_key?(stats, :waiting) + + assert is_integer(stats.total) + assert is_integer(stats.active) + assert is_integer(stats.idle) + assert is_integer(stats.idle_in_transaction) + assert is_integer(stats.waiting) + end + end + + describe "calculate_optimal_batch_size/1 with various value types" do + test "handles maps with float values" do + entries = [%{temp: 98.6, pressure: 1013.25}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "handles maps with boolean values" do + entries = [%{active: true, verified: false}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "handles maps with DateTime values" do + entries = [%{created_at: DateTime.utc_now(), name: "test"}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "handles maps with Date values" do + entries = [%{date: Date.utc_today(), label: "today"}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "handles maps with nil values" do + entries = [%{name: nil, value: nil}] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "handles maps with mixed value types" do + entries = [ + %{ + name: "test", + count: 42, + ratio: 3.14, + active: true, + created: DateTime.utc_now(), + date: Date.utc_today(), + extra: nil + } + ] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + + test "handles non-map first entry as unknown type" do + entries = ["just a string", "another"] + + assert DbOptimizer.calculate_optimal_batch_size(entries) == 2000 + end + end + + describe "optimized_batch_insert/3 with invalid data" do + test "handles insert errors and returns error count" do + # Insert entries into a nonexistent table - should error + entries = [%{"bad_col" => "value"}] + + {success, errors} = DbOptimizer.optimized_batch_insert("nonexistent_table_xyz", entries) + + assert success == 0 + assert errors == 1 + end + end + + describe "vacuum_table/1 with options" do + test "with full: true returns :error in sandbox" do + assert :error = DbOptimizer.vacuum_table("packets", full: true) + end + + test "with analyze: false returns :error in sandbox" do + assert :error = DbOptimizer.vacuum_table("packets", analyze: false) + end + end +end diff --git a/test/aprsme/dynamic_supervisor_test.exs b/test/aprsme/dynamic_supervisor_test.exs new file mode 100644 index 0000000..2b4d70b --- /dev/null +++ b/test/aprsme/dynamic_supervisor_test.exs @@ -0,0 +1,88 @@ +defmodule Aprsme.DynamicSupervisorTest do + use ExUnit.Case, async: false + + alias Aprsme.DynamicSupervisor, as: DynSup + + setup do + # Stop existing DynamicSupervisor if it's running + case Process.whereis(DynSup) do + nil -> :ok + pid -> GenServer.stop(pid) + end + + on_exit(fn -> + case Process.whereis(DynSup) do + nil -> + :ok + + pid -> + ref = Process.monitor(pid) + Process.exit(pid, :shutdown) + + receive do + {:DOWN, ^ref, :process, ^pid, _} -> :ok + after + 5000 -> :ok + end + end + end) + + :ok + end + + describe "start_link/1" do + test "registers process with module name" do + assert {:ok, pid} = DynSup.start_link([]) + assert is_pid(pid) + assert Process.whereis(DynSup) == pid + assert Process.alive?(pid) + end + end + + describe "starting a child" do + test "starts a child process that is alive" do + {:ok, _sup} = DynSup.start_link([]) + + child_spec = %{ + id: :test_agent, + start: {Agent, :start_link, [fn -> :ok end]}, + restart: :temporary + } + + assert {:ok, child_pid} = DynamicSupervisor.start_child(DynSup, child_spec) + assert Process.alive?(child_pid) + end + end + + describe "terminating a child" do + test "terminates a child and it is no longer alive" do + {:ok, _sup} = DynSup.start_link([]) + + child_spec = %{ + id: :test_agent, + start: {Agent, :start_link, [fn -> :ok end]}, + restart: :temporary + } + + {:ok, child_pid} = DynamicSupervisor.start_child(DynSup, child_spec) + assert Process.alive?(child_pid) + + assert :ok = DynamicSupervisor.terminate_child(DynSup, child_pid) + refute Process.alive?(child_pid) + end + end + + describe "init/1" do + test "returns expected supervisor flags" do + assert {:ok, flags} = DynSup.init([]) + + assert %{ + intensity: 3, + period: 5, + strategy: :one_for_one, + max_children: :infinity, + extra_arguments: [] + } = flags + end + end +end diff --git a/test/aprsme/error_handler_test.exs b/test/aprsme/error_handler_test.exs new file mode 100644 index 0000000..e2df046 --- /dev/null +++ b/test/aprsme/error_handler_test.exs @@ -0,0 +1,289 @@ +defmodule Aprsme.ErrorHandlerTest do + use ExUnit.Case, async: true + + alias Aprsme.ErrorHandler + + # ── helpers to build error structs without hitting complex exception/1 ── + + defp cast_error(message \\ "cannot cast value") do + %Ecto.Query.CastError{type: :string, value: 42, message: message} + end + + defp constraint_error(constraint \\ "users_email_index") do + %Ecto.ConstraintError{type: :unique, constraint: constraint, message: "constraint error"} + end + + defp postgrex_unique_violation do + %Postgrex.Error{postgres: %{code: :unique_violation}, message: nil, connection_id: nil} + end + + defp db_connection_error do + %DBConnection.ConnectionError{message: "connection refused", severity: :error, reason: :error} + end + + defp query_error(message \\ "invalid query") do + %Ecto.QueryError{message: message} + end + + defp transport_error(reason) do + %Req.TransportError{reason: reason} + end + + # ── handle_database_error/2 ── + + describe "handle_database_error/2" do + test "CastError returns {:error, :invalid_data}" do + assert {:error, :invalid_data} = ErrorHandler.handle_database_error(cast_error()) + end + + test "ConstraintError returns {:error, :constraint_violation}" do + assert {:error, :constraint_violation} = + ErrorHandler.handle_database_error(constraint_error()) + end + + test "Postgrex.Error with unique_violation returns {:error, :duplicate_record}" do + assert {:error, :duplicate_record} = + ErrorHandler.handle_database_error(postgrex_unique_violation()) + end + + test "DBConnection.ConnectionError returns {:error, :database_unavailable}" do + assert {:error, :database_unavailable} = + ErrorHandler.handle_database_error(db_connection_error()) + end + + test "Ecto.QueryError returns {:error, :query_failed}" do + assert {:error, :query_failed} = ErrorHandler.handle_database_error(query_error()) + end + + test "catch-all returns {:error, :database_error}" do + assert {:error, :database_error} = + ErrorHandler.handle_database_error(%RuntimeError{message: "boom"}) + end + + test "passes context through" do + assert {:error, :invalid_data} = + ErrorHandler.handle_database_error(cast_error(), %{table: "users"}) + end + end + + # ── handle_network_error/3 ── + + describe "handle_network_error/3" do + test "Req.TransportError with :timeout returns {:error, :timeout}" do + assert {:error, :timeout} = + ErrorHandler.handle_network_error(transport_error(:timeout), :aprs_is) + end + + test "Req.TransportError with :econnrefused returns {:error, :service_unavailable}" do + assert {:error, :service_unavailable} = + ErrorHandler.handle_network_error(transport_error(:econnrefused), :aprs_is) + end + + test "Req.TransportError with other reason returns {:error, :network_error}" do + assert {:error, :network_error} = + ErrorHandler.handle_network_error(transport_error(:nxdomain), :geocoder) + end + + test "{:error, :circuit_open} returns {:error, :service_unavailable}" do + assert {:error, :service_unavailable} = + ErrorHandler.handle_network_error({:error, :circuit_open}, :weather_api) + end + + test "catch-all returns {:error, :network_error}" do + assert {:error, :network_error} = + ErrorHandler.handle_network_error(:some_unknown_error, :some_service) + end + + test "passes context through" do + assert {:error, :timeout} = + ErrorHandler.handle_network_error( + transport_error(:timeout), + :aprs_is, + %{attempt: 1} + ) + end + end + + # ── handle_packet_error/3 ── + + describe "handle_packet_error/3" do + test "Jason.DecodeError returns {:error, :invalid_json}" do + error = %Jason.DecodeError{position: 0, token: nil, data: "bad"} + + assert {:error, :invalid_json} = + ErrorHandler.handle_packet_error(error, %{sender: "N0CALL"}) + end + + test "ArgumentError with 'invalid' in message returns {:error, :invalid_format}" do + error = %ArgumentError{message: "invalid byte sequence"} + + assert {:error, :invalid_format} = + ErrorHandler.handle_packet_error(error, %{sender: "N0CALL"}) + end + + test "ArgumentError without 'invalid' in message returns {:error, :processing_failed}" do + error = %ArgumentError{message: "bad argument"} + + assert {:error, :processing_failed} = + ErrorHandler.handle_packet_error(error, %{sender: "N0CALL"}) + end + + test "catch-all returns {:error, :processing_failed}" do + error = %RuntimeError{message: "something went wrong"} + + assert {:error, :processing_failed} = + ErrorHandler.handle_packet_error(error, %{sender: "N0CALL"}) + end + + test "uses 'unknown' when packet_data has no :sender key" do + error = %Jason.DecodeError{position: 0, token: nil, data: "bad"} + + assert {:error, :invalid_json} = ErrorHandler.handle_packet_error(error, %{}) + end + + test "passes context through" do + error = %Jason.DecodeError{position: 0, token: nil, data: "bad"} + + assert {:error, :invalid_json} = + ErrorHandler.handle_packet_error(error, %{sender: "N0CALL"}, %{source: "is"}) + end + end + + # ── handle_process_error/3 ── + + describe "handle_process_error/3" do + test ":timeout returns {:error, :timeout}" do + assert {:error, :timeout} = ErrorHandler.handle_process_error(:timeout, :my_genserver) + end + + test "{:noproc, _} returns {:error, :process_unavailable}" do + assert {:error, :process_unavailable} = + ErrorHandler.handle_process_error({:noproc, {GenServer, :call, [:pid]}}, :worker) + end + + test "{:badrpc, reason} returns {:error, :rpc_failed}" do + assert {:error, :rpc_failed} = + ErrorHandler.handle_process_error({:badrpc, :nodedown}, :remote_node) + end + + test "catch-all returns {:error, :process_error}" do + assert {:error, :process_error} = + ErrorHandler.handle_process_error(:something_else, :my_process) + end + + test "passes context through" do + assert {:error, :timeout} = + ErrorHandler.handle_process_error(:timeout, :my_genserver, %{call: :get_state}) + end + end + + # ── with_error_handling/2 ── + + describe "with_error_handling/2" do + test "success path returns {:ok, value}" do + assert {:ok, 42} = ErrorHandler.with_error_handling(fn -> 42 end) + end + + test "success path returns {:ok, value} for complex results" do + assert {:ok, %{key: "value"}} = ErrorHandler.with_error_handling(fn -> %{key: "value"} end) + end + + test "raise with no retries goes to categorize_and_handle_error" do + result = + ErrorHandler.with_error_handling(fn -> + raise RuntimeError, "boom" + end) + + assert {:error, :unknown_error} = result + end + + test "raise with retries retries before failing" do + {:ok, counter} = Agent.start_link(fn -> 0 end) + + result = + ErrorHandler.with_error_handling( + fn -> + count = Agent.get_and_update(counter, fn n -> {n, n + 1} end) + + if count < 2 do + raise RuntimeError, "boom" + end + + :success + end, + max_retries: 2, + retry_delay: 1 + ) + + assert {:ok, :success} = result + + final_count = Agent.get(counter, & &1) + assert final_count == 3 + + Agent.stop(counter) + end + + test "raise with retries exhausted calls categorize_and_handle_error" do + {:ok, counter} = Agent.start_link(fn -> 0 end) + + result = + ErrorHandler.with_error_handling( + fn -> + Agent.update(counter, &(&1 + 1)) + raise RuntimeError, "always fails" + end, + max_retries: 2, + retry_delay: 1 + ) + + assert {:error, :unknown_error} = result + + # Initial attempt + 2 retries = 3 total calls + final_count = Agent.get(counter, & &1) + assert final_count == 3 + + Agent.stop(counter) + end + + test "throw path returns {:error, :unexpected_error}" do + result = + ErrorHandler.with_error_handling(fn -> + throw(:something_bad) + end) + + assert {:error, :unexpected_error} = result + end + + test "database error routes through handle_database_error" do + result = + ErrorHandler.with_error_handling(fn -> + raise DBConnection.ConnectionError, "connection refused" + end) + + assert {:error, :database_unavailable} = result + end + + test "network error routes through handle_network_error" do + result = + ErrorHandler.with_error_handling(fn -> + raise Req.TransportError, reason: :timeout + end) + + assert {:error, :timeout} = result + end + + test "uncategorized error returns {:error, :unknown_error}" do + result = + ErrorHandler.with_error_handling(fn -> + raise ArgumentError, "nope" + end) + + assert {:error, :unknown_error} = result + end + + test "accepts context option" do + assert {:ok, :fine} = + ErrorHandler.with_error_handling(fn -> :fine end, context: %{source: "test"}) + end + end +end diff --git a/test/aprsme/migration_lock_test.exs b/test/aprsme/migration_lock_test.exs new file mode 100644 index 0000000..e4c0c7a --- /dev/null +++ b/test/aprsme/migration_lock_test.exs @@ -0,0 +1,179 @@ +defmodule Aprsme.MigrationLockTest do + use Aprsme.DataCase, async: false + + alias Aprsme.MigrationLock + + @lock_id 8_764_293_847_291 + + setup do + # Ensure the advisory lock is released before each test. + # Advisory locks are session-level, so we need to make sure none + # are left over from a previous test. + Repo.query("SELECT pg_advisory_unlock_all()") + :ok + end + + describe "with_lock/2" do + test "acquires lock, runs function, and returns result" do + result = MigrationLock.with_lock(Repo, fn -> {:ok, :migrations_ran} end) + + assert result == {:ok, :migrations_ran} + + # Verify lock was released by acquiring it again successfully + {:ok, %{rows: [[locked]]}} = + Repo.query("SELECT pg_try_advisory_lock($1)", [@lock_id]) + + assert locked == true + + # Clean up: release the lock we just acquired + Repo.query("SELECT pg_advisory_unlock($1)", [@lock_id]) + end + + test "releases lock even when function raises" do + assert_raise RuntimeError, "boom", fn -> + MigrationLock.with_lock(Repo, fn -> raise "boom" end) + end + + # Verify lock was released by acquiring it again successfully + {:ok, %{rows: [[locked]]}} = + Repo.query("SELECT pg_try_advisory_lock($1)", [@lock_id]) + + assert locked == true + + # Clean up: release the lock we just acquired + Repo.query("SELECT pg_advisory_unlock($1)", [@lock_id]) + end + + test "returns :skipped when lock is already held" do + # Acquire the advisory lock manually in this session + {:ok, %{rows: [[true]]}} = + Repo.query("SELECT pg_try_advisory_lock($1)", [@lock_id]) + + # Call with_lock - since the lock is already held by the same session, + # pg_try_advisory_lock will return true (PostgreSQL advisory locks are + # reentrant within the same session). In the sandbox, all queries go + # through the same DB connection/session, so we can't simulate true + # contention. Instead, we verify the function runs and returns its result + # (showing the lock was successfully acquired and released). + result = MigrationLock.with_lock(Repo, fn -> :ran end) + assert result == :ran + + # Clean up: release all advisory locks + Repo.query("SELECT pg_advisory_unlock_all()") + end + + test "with_lock returns the result of the passed function" do + result = MigrationLock.with_lock(Repo, fn -> {:ok, :done} end) + + assert result == {:ok, :done} + end + + test "with_lock passes through various return values" do + assert MigrationLock.with_lock(Repo, fn -> :ok end) == :ok + assert MigrationLock.with_lock(Repo, fn -> 42 end) == 42 + assert MigrationLock.with_lock(Repo, fn -> [1, 2, 3] end) == [1, 2, 3] + end + end + + describe "with_lock/2 using stub repo" do + defmodule LockedThenUnlockedRepo do + @moduledoc false + @lock_id 8_764_293_847_291 + + # First call: acquire_lock returns false (locked by another node) + # Second call: wait_for_migrations finds lock available (other node finished) + def query("SELECT pg_try_advisory_lock($1)", [@lock_id]) do + case Process.get(:lock_call_count, 0) do + 0 -> + Process.put(:lock_call_count, 1) + {:ok, %{rows: [[false]]}} + + _ -> + {:ok, %{rows: [[true]]}} + end + end + + def query("SELECT pg_advisory_unlock($1)", [@lock_id]) do + {:ok, %{rows: [[true]]}} + end + end + + defmodule ErrorRepo do + @moduledoc false + @lock_id 8_764_293_847_291 + + def query("SELECT pg_try_advisory_lock($1)", [@lock_id]) do + {:error, :database_error} + end + + def query("SELECT pg_advisory_unlock($1)", [@lock_id]) do + {:error, :database_error} + end + end + + defmodule WaitErrorRepo do + @moduledoc false + @lock_id 8_764_293_847_291 + + # First call: acquire_lock returns false (locked) + # Second call: wait_for_migrations gets an error + def query("SELECT pg_try_advisory_lock($1)", [@lock_id]) do + case Process.get(:wait_err_count, 0) do + 0 -> + Process.put(:wait_err_count, 1) + {:ok, %{rows: [[false]]}} + + _ -> + {:error, :connection_lost} + end + end + + def query("SELECT pg_advisory_unlock($1)", [@lock_id]) do + {:ok, %{rows: [[true]]}} + end + end + + test "returns :skipped when lock held, then other node finishes" do + Process.put(:lock_call_count, 0) + + result = MigrationLock.with_lock(LockedThenUnlockedRepo, fn -> :should_not_run end) + + assert result == :skipped + end + + test "handles error from acquire_lock" do + result = MigrationLock.with_lock(ErrorRepo, fn -> :should_not_run end) + + assert result == :skipped + end + + test "handles error during wait_for_migrations" do + Process.put(:wait_err_count, 0) + + result = MigrationLock.with_lock(WaitErrorRepo, fn -> :should_not_run end) + + assert result == :skipped + end + end + + describe "with_lock/2 release error" do + defmodule AcquireButFailRelease do + @moduledoc false + @lock_id 8_764_293_847_291 + + def query("SELECT pg_try_advisory_lock($1)", [@lock_id]) do + {:ok, %{rows: [[true]]}} + end + + def query("SELECT pg_advisory_unlock($1)", [@lock_id]) do + {:error, :release_failed} + end + end + + test "still returns result even when release_lock fails" do + result = MigrationLock.with_lock(AcquireButFailRelease, fn -> {:ok, :done} end) + + assert result == {:ok, :done} + end + end +end diff --git a/test/aprsme/mock_helpers_test.exs b/test/aprsme/mock_helpers_test.exs new file mode 100644 index 0000000..c4bfc39 --- /dev/null +++ b/test/aprsme/mock_helpers_test.exs @@ -0,0 +1,22 @@ +defmodule Aprsme.MockHelpersTest do + use ExUnit.Case, async: false + + import Aprsme.MockHelpers + + setup do + Mox.set_mox_private() + :ok + end + + describe "stub_packets_mock/0" do + test "stubs get_recent_packets/1 to return []" do + stub_packets_mock() + assert [] = Aprsme.PacketsMock.get_recent_packets(%{}) + end + + test "stubs get_nearby_stations/4 to return []" do + stub_packets_mock() + assert [] = Aprsme.PacketsMock.get_nearby_stations(39.0, -104.0, nil, %{}) + end + end +end diff --git a/test/aprsme/packet_consumer_pool_test.exs b/test/aprsme/packet_consumer_pool_test.exs new file mode 100644 index 0000000..30d4663 --- /dev/null +++ b/test/aprsme/packet_consumer_pool_test.exs @@ -0,0 +1,84 @@ +defmodule Aprsme.PacketConsumerPoolTest do + use ExUnit.Case, async: true + + alias Aprsme.PacketConsumerPool + + describe "child_spec/1" do + test "returns expected spec" do + spec = PacketConsumerPool.child_spec([]) + + assert spec.id == PacketConsumerPool + assert spec.type == :supervisor + end + end + + describe "init/1" do + test "returns supervisor flags with one_for_one strategy and 3 children by default" do + {:ok, {sup_flags, children}} = PacketConsumerPool.init([]) + + assert sup_flags.strategy == :one_for_one + assert length(children) == 3 + end + + test "each child has a unique id of {Aprsme.PacketConsumer, index}" do + {:ok, {_sup_flags, children}} = PacketConsumerPool.init([]) + + ids = Enum.map(children, & &1.id) + + assert {Aprsme.PacketConsumer, 1} in ids + assert {Aprsme.PacketConsumer, 2} in ids + assert {Aprsme.PacketConsumer, 3} in ids + assert length(ids) == length(Enum.uniq(ids)) + end + + test "creates 5 children when num_consumers: 5 is passed in opts" do + {:ok, {_sup_flags, children}} = PacketConsumerPool.init(num_consumers: 5) + + assert length(children) == 5 + + ids = Enum.map(children, & &1.id) + + for index <- 1..5 do + assert {Aprsme.PacketConsumer, index} in ids + end + end + + test "max_demand is divided evenly among consumers" do + original_config = Application.get_env(:aprsme, :packet_pipeline, []) + + on_exit(fn -> + Application.put_env(:aprsme, :packet_pipeline, original_config) + end) + + Application.put_env(:aprsme, :packet_pipeline, max_demand: 500) + + {:ok, {_sup_flags, children}} = PacketConsumerPool.init(num_consumers: 5) + + expected_demand = div(500, 5) + + for child <- children do + {Aprsme.PacketConsumer, :start_link, [opts]} = child.start + assert opts[:max_demand] == expected_demand + assert opts[:subscribe_to] == [{Aprsme.PacketProducer, max_demand: expected_demand}] + end + end + + test "uses default max_demand of 250 when not configured" do + original_config = Application.get_env(:aprsme, :packet_pipeline, []) + + on_exit(fn -> + Application.put_env(:aprsme, :packet_pipeline, original_config) + end) + + Application.delete_env(:aprsme, :packet_pipeline) + + {:ok, {_sup_flags, children}} = PacketConsumerPool.init([]) + + expected_demand = div(250, 3) + + child = List.first(children) + {Aprsme.PacketConsumer, :start_link, [opts]} = child.start + assert opts[:max_demand] == expected_demand + end + end +end diff --git a/test/aprsme/packet_counter_test.exs b/test/aprsme/packet_counter_test.exs new file mode 100644 index 0000000..385beae --- /dev/null +++ b/test/aprsme/packet_counter_test.exs @@ -0,0 +1,40 @@ +defmodule Aprsme.PacketCounterTest do + use Aprsme.DataCase, async: true + + import Ecto.Query + + alias Aprsme.PacketCounter + alias Aprsme.Repo + + describe "get_count/0" do + test "returns 0 when no records exist" do + Repo.delete_all(PacketCounter) + + assert PacketCounter.get_count() == 0 + end + end + + describe "get_count/1" do + test "returns the count after inserting a record" do + # Update the existing seeded row rather than inserting a duplicate + Repo.update_all(from(pc in PacketCounter, where: pc.counter_type == "total_packets"), set: [count: 42]) + assert PacketCounter.get_count("total_packets") == 42 + end + + test "returns 0 for a counter_type that does not exist" do + assert PacketCounter.get_count("nonexistent_type") == 0 + end + + test "returns the correct count for a custom counter_type" do + Repo.insert!(%PacketCounter{counter_type: "daily_packets", count: 100}) + + assert PacketCounter.get_count("daily_packets") == 100 + end + end + + describe "subscribe_to_changes/0" do + test "returns :ok" do + assert PacketCounter.subscribe_to_changes() == :ok + end + end +end diff --git a/test/aprsme/packet_pipeline_setup_test.exs b/test/aprsme/packet_pipeline_setup_test.exs new file mode 100644 index 0000000..5ad10b7 --- /dev/null +++ b/test/aprsme/packet_pipeline_setup_test.exs @@ -0,0 +1,99 @@ +defmodule Aprsme.PacketPipelineSetupTest do + use ExUnit.Case, async: false + + import ExUnit.CaptureLog + + alias Aprsme.PacketPipelineSetup + + setup do + # Stop any existing PacketPipelineSetup process + case GenServer.whereis(PacketPipelineSetup) do + nil -> :ok + pid -> GenServer.stop(pid, :normal, 5000) + end + + on_exit(fn -> + case GenServer.whereis(PacketPipelineSetup) do + nil -> + :ok + + pid -> + try do + GenServer.stop(pid, :normal, 5000) + catch + :exit, _ -> :ok + end + end + end) + + :ok + end + + describe "start_link/0" do + test "starts the GenServer with default opts" do + capture_log(fn -> + {:ok, pid} = PacketPipelineSetup.start_link() + + assert Process.alive?(pid) + assert GenServer.whereis(PacketPipelineSetup) == pid + end) + end + end + + describe "start_link/1" do + test "starts the GenServer and registers it under its module name" do + capture_log(fn -> + {:ok, pid} = PacketPipelineSetup.start_link([]) + + assert Process.alive?(pid) + assert GenServer.whereis(PacketPipelineSetup) == pid + end) + end + end + + describe "handle_info(:setup_subscription, state)" do + test "GenServer survives when PacketProducer/PacketConsumer are not running and schedules retry" do + log = + capture_log(fn -> + {:ok, pid} = PacketPipelineSetup.start_link([]) + + # Wait for the initial :setup_subscription to fire (~100ms) and fail + Process.sleep(200) + + # The GenServer should still be alive after the failure + assert Process.alive?(pid) + end) + + # The error path should have logged the failure + assert log =~ "Failed to establish GenStage subscription" + end + + test "retries subscription after failure" do + log = + capture_log(fn -> + {:ok, pid} = PacketPipelineSetup.start_link([]) + + # Wait for initial attempt (~100ms) and retry (~1000ms) + Process.sleep(1300) + + assert Process.alive?(pid) + end) + + # Should see multiple failure logs from retry + assert log =~ "Failed to establish GenStage subscription" + end + end + + describe "init/1" do + test "schedules :setup_subscription message" do + capture_log(fn -> + {:ok, pid} = PacketPipelineSetup.start_link([]) + + # The init should have scheduled a message + # Verify state is initialized as empty map + state = :sys.get_state(pid) + assert state == %{} + end) + end + end +end diff --git a/test/aprsme/packet_pipeline_supervisor_test.exs b/test/aprsme/packet_pipeline_supervisor_test.exs new file mode 100644 index 0000000..e9937cf --- /dev/null +++ b/test/aprsme/packet_pipeline_supervisor_test.exs @@ -0,0 +1,125 @@ +defmodule Aprsme.PacketPipelineSupervisorTest do + use ExUnit.Case, async: true + + alias Aprsme.PacketPipelineSupervisor + + describe "init/1 with default config" do + setup do + original = Application.get_env(:aprsme, :packet_pipeline) + Application.delete_env(:aprsme, :packet_pipeline) + + on_exit(fn -> + if original do + Application.put_env(:aprsme, :packet_pipeline, original) + else + Application.delete_env(:aprsme, :packet_pipeline) + end + end) + + :ok + end + + test "returns :one_for_one strategy" do + {:ok, {sup_flags, _children}} = PacketPipelineSupervisor.init([]) + + assert %{strategy: :one_for_one} = sup_flags + end + + test "returns two child specs" do + {:ok, {_sup_flags, children}} = PacketPipelineSupervisor.init([]) + + assert length(children) == 2 + end + + test "includes PacketProducer with default max_buffer_size of 1000" do + {:ok, {_sup_flags, children}} = PacketPipelineSupervisor.init([]) + + producer_spec = Enum.find(children, fn spec -> spec.id == Aprsme.PacketProducer end) + + assert producer_spec + assert {Aprsme.PacketProducer, [max_buffer_size: 1000]} = extract_mfa(producer_spec.start) + end + + test "includes PacketConsumerPool with default num_consumers of 3" do + {:ok, {_sup_flags, children}} = PacketPipelineSupervisor.init([]) + + pool_spec = Enum.find(children, fn spec -> spec.id == Aprsme.PacketConsumerPool end) + + assert pool_spec + assert {Aprsme.PacketConsumerPool, [num_consumers: 3]} = extract_mfa(pool_spec.start) + end + end + + describe "init/1 with custom config" do + setup do + original = Application.get_env(:aprsme, :packet_pipeline) + + Application.put_env(:aprsme, :packet_pipeline, + max_buffer_size: 5000, + num_consumers: 10 + ) + + on_exit(fn -> + if original do + Application.put_env(:aprsme, :packet_pipeline, original) + else + Application.delete_env(:aprsme, :packet_pipeline) + end + end) + + :ok + end + + test "uses custom max_buffer_size from config" do + {:ok, {_sup_flags, children}} = PacketPipelineSupervisor.init([]) + + producer_spec = Enum.find(children, fn spec -> spec.id == Aprsme.PacketProducer end) + + assert {Aprsme.PacketProducer, [max_buffer_size: 5000]} = extract_mfa(producer_spec.start) + end + + test "uses custom num_consumers from config" do + {:ok, {_sup_flags, children}} = PacketPipelineSupervisor.init([]) + + pool_spec = Enum.find(children, fn spec -> spec.id == Aprsme.PacketConsumerPool end) + + assert {Aprsme.PacketConsumerPool, [num_consumers: 10]} = extract_mfa(pool_spec.start) + end + end + + describe "start_link/0" do + test "is exported as a function" do + assert function_exported?(PacketPipelineSupervisor, :start_link, 0) + end + + test "starts the supervisor process" do + # Stop existing instance if running + case Process.whereis(PacketPipelineSupervisor) do + nil -> :ok + pid -> Supervisor.stop(pid, :normal, 5000) + end + + case PacketPipelineSupervisor.start_link() do + {:ok, pid} -> + assert Process.alive?(pid) + Supervisor.stop(pid, :normal, 5000) + + {:error, _reason} -> + # Children may fail to start if dependencies aren't running + :ok + end + end + + test "child_spec/1 returns expected spec" do + spec = PacketPipelineSupervisor.child_spec([]) + + assert spec.id == PacketPipelineSupervisor + assert spec.type == :supervisor + end + end + + # Extracts {module, args} from a child spec's start MFA tuple + defp extract_mfa({module, :start_link, [args]}) do + {module, args} + end +end