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

@@ -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

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

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

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

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

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

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

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

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

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

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

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

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

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