graham/towerops
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

stagger preseem sync across organizations

Browse source 
Replace the thundering-herd cron job with a dispatcher that enqueues
individual per-integration Oban jobs staggered across the 10-minute
window using PollingOffset. Show last synced timestamp in user's
local timezone via the <.timestamp> component.
This commit is contained in:
Graham McIntire 2026-02-13 10:52:32 -06:00
parent bad1590fed
commit 2e9f1660d0
No known key found for this signature in database
11 changed files with 368 additions and 251 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
lib/towerops/integrations.ex
View file

@ -54,6 +54,13 @@ defmodule Towerops.Integrations do
Integration.changeset(integration, attrs)
end
def get_integration_by_id(id) do
case Repo.get(Integration, id) do
nil -> {:error, :not_found}
integration -> {:ok, integration}
end
end
def list_enabled_integrations(provider) do
Integration
|> where(provider: ^provider, enabled: true)

52
lib/towerops/preseem/client.ex
View file

@ -3,6 +3,7 @@ defmodule Towerops.Preseem.Client do
HTTP client for the Preseem Model API.
Uses Req with built-in test support via `Req.Test`.
Auth: HTTP Basic (API key as username, empty password).
"""
require Logger
@ -10,12 +11,12 @@ defmodule Towerops.Preseem.Client do
@base_url "https://api.preseem.com/model/v1"
@doc """
Tests the connection to the Preseem API by fetching account info.
Tests the connection to the Preseem API by fetching a page of access points.
Returns `{:ok, body}` on success, `{:error, reason}` on failure.
"""
def test_connection(api_key) do
case request(:get, "#{@base_url}/account", api_key) do
case request(:get, "#{@base_url}/access_points?limit=100", api_key) do
{:ok, %{status: status, body: body}} when status in 200..299 ->
{:ok, body}
@ -36,14 +37,15 @@ defmodule Towerops.Preseem.Client do
end
@doc """
Lists all access points from the Preseem API.
Lists all access points with scores from the Preseem API.
Returns `{:ok, [access_point]}` on success.
Fetches from `/access_point_scores` which returns rich data per AP.
Returns `{:ok, [access_point_score]}` on success.
"""
def list_access_points(api_key) do
case request(:get, "#{@base_url}/access_points", api_key) do
{:ok, %{status: status, body: %{"data" => data}}} when status in 200..299 ->
{:ok, data}
case request(:get, "#{@base_url}/access_point_scores", api_key) do
{:ok, %{status: status, body: body}} when status in 200..299 ->
{:ok, normalize_scores_response(body)}
{:ok, %{status: 401}} ->
{:error, :unauthorized}
@ -60,14 +62,27 @@ defmodule Towerops.Preseem.Client do
end
@doc """
Fetches metrics for a specific access point.
Lists access points from the basic `/access_points` endpoint with pagination.
Returns `{:ok, metrics_map}` on success.
Returns `{:ok, [access_point]}` on success.
"""
def get_access_point_metrics(api_key, ap_id) do
case request(:get, "#{@base_url}/access_points/#{ap_id}/metrics", api_key) do
{:ok, %{status: status, body: %{"data" => data}}} when status in 200..299 ->
{:ok, data}
def list_access_points_basic(api_key) do
fetch_all_pages(api_key, "#{@base_url}/access_points?limit=500", [])
end
defp fetch_all_pages(api_key, url, acc) do
case request(:get, url, api_key) do
{:ok, %{status: status, body: %{"data" => data, "paginator" => paginator}}}
when status in 200..299 ->
all = acc ++ data
if paginator["page"] < paginator["page_count"] do
next_page = paginator["page"] + 1
next_url = "#{@base_url}/access_points?limit=500&page=#{next_page}"
fetch_all_pages(api_key, next_url, all)
else
{:ok, all}
end
{:ok, %{status: 401}} ->
{:error, :unauthorized}
@ -83,11 +98,20 @@ defmodule Towerops.Preseem.Client do
end
end
# The scores endpoint may return an array, a wrapped object with "data" key,
# or a single object. Normalize to a list.
defp normalize_scores_response(body) when is_list(body), do: body
defp normalize_scores_response(%{"data" => data}) when is_list(data), do: data
defp normalize_scores_response(body) when is_map(body), do: [body]
defp request(method, url, api_key) do
opts = [
method: method,
url: url,
headers: [{"authorization", "Bearer #{api_key}"}, {"accept", "application/json"}]
auth: {:basic, "#{api_key}:"},
headers: [{"accept", "application/json"}]
]
# Only use Req.Test plug in test environment

45
lib/towerops/preseem/sync.ex
View file

@ -2,7 +2,7 @@ defmodule Towerops.Preseem.Sync do
@moduledoc """
Orchestrates syncing data from the Preseem API into the local database.
Pulls access points (and optional metrics) from Preseem, upserts them
Pulls access point scores from Preseem, upserts them
into `preseem_access_points`, and logs each sync operation.
"""
@ -10,7 +10,6 @@ defmodule Towerops.Preseem.Sync do
alias Towerops.Preseem.AccessPoint
alias Towerops.Preseem.Client
alias Towerops.Preseem.DeviceMatcher
alias Towerops.Preseem.SubscriberMetric
alias Towerops.Preseem.SyncLog
alias Towerops.Repo
@ -19,8 +18,8 @@ defmodule Towerops.Preseem.Sync do
@doc """
Main entry point: syncs all access points for the given integration.
Fetches APs from the Preseem API, upserts them, optionally inserts
metrics, writes a sync log, and updates the integration's sync status.
Fetches AP scores from the Preseem API, upserts them, writes a sync log,
and updates the integration's sync status.
Returns `{:ok, %{synced: count, matched: count}}` or `{:error, reason}`.
"""
@ -41,24 +40,24 @@ defmodule Towerops.Preseem.Sync do
@doc """
Upserts a single access point from raw API response data.
Maps API fields to schema fields and uses the unique constraint on
Maps Preseem scores API fields to schema fields and uses the unique constraint on
`[:organization_id, :preseem_id]` for conflict resolution.
"""
def upsert_access_point(organization_id, ap_data) do
attrs = %{
organization_id: organization_id,
preseem_id: ap_data["id"],
preseem_id: ap_data["element_uuid"],
name: ap_data["name"],
mac_address: ap_data["mac_address"],
ip_address: ap_data["ip_address"],
model: ap_data["model"],
mac_address: ap_data["system_mac_address"],
ip_address: ap_data["management_ip"],
model: ap_data["model_name"],
firmware: ap_data["firmware"],
capacity_score: ap_data["capacity_score"],
qoe_score: ap_data["qoe_score"],
rf_score: ap_data["rf_score"],
busy_hours: ap_data["busy_hours"],
airtime_utilization: ap_data["airtime_utilization"],
subscriber_count: ap_data["subscriber_count"],
capacity_score: cast_to_float(ap_data["subscriber_capacity"]),
qoe_score: cast_to_float(ap_data["ap_health_score"]),
rf_score: cast_to_float(ap_data["rf_score"]),
busy_hours: nil,
airtime_utilization: cast_to_float(ap_data["airtime_remaining_tx"]),
subscriber_count: ap_data["connections"],
raw_data: ap_data
}
@ -76,6 +75,8 @@ defmodule Towerops.Preseem.Sync do
Inserts a subscriber metric record for the given access point.
"""
def insert_metrics(access_point_id, metrics_data) do
alias Towerops.Preseem.SubscriberMetric
recorded_at =
case metrics_data["recorded_at"] do
nil -> DateTime.truncate(DateTime.utc_now(), :second)
@ -102,8 +103,7 @@ defmodule Towerops.Preseem.Sync do
synced_count =
Enum.reduce(ap_list, 0, fn ap_data, count ->
case upsert_access_point(org_id, ap_data) do
{:ok, ap} ->
maybe_insert_metrics(ap, ap_data)
{:ok, _ap} ->
count + 1
{:error, changeset} ->
@ -132,13 +132,6 @@ defmodule Towerops.Preseem.Sync do
{:error, reason}
end
defp maybe_insert_metrics(ap, ap_data) do
case ap_data["metrics"] do
nil -> :ok
metrics when is_map(metrics) -> insert_metrics(ap.id, metrics)
end
end
defp run_device_matching(org_id) do
if Code.ensure_loaded?(DeviceMatcher) do
case DeviceMatcher.match_unmatched(org_id) do
@ -170,4 +163,8 @@ defmodule Towerops.Preseem.Sync do
_ -> DateTime.truncate(DateTime.utc_now(), :second)
end
end
defp cast_to_float(nil), do: nil
defp cast_to_float(val) when is_float(val), do: val
defp cast_to_float(val) when is_integer(val), do: val / 1
end

75
lib/towerops/workers/preseem_sync_worker.ex
View file

@ -1,48 +1,77 @@
defmodule Towerops.Workers.PreseemSyncWorker do
@moduledoc """
Oban cron worker that syncs Preseem data for all enabled integrations.
Oban worker that syncs Preseem data for enabled integrations.
Runs every 10 minutes. For each org with an enabled Preseem integration,
checks if enough time has elapsed since the last sync (based on the
integration's sync_interval_minutes), then calls Preseem.Sync.sync_organization/1.
The cron job (no args) runs every 10 minutes and dispatches individual
per-integration sync jobs staggered across the 600-second window using
`PollingOffset.calculate_offset/2`. This prevents a thundering herd
against the Preseem API.
Individual jobs receive `%{"integration_id" => id}` and perform the
actual sync via `Preseem.Sync.sync_organization/1`.
"""
use Oban.Worker, queue: :maintenance
alias Towerops.Integrations
alias Towerops.Preseem.Sync
alias Towerops.Workers.PollingOffset
require Logger
@window_seconds 600
# Cron dispatcher: enqueues staggered individual sync jobs
@impl Oban.Worker
def perform(%Oban.Job{}) do
def perform(%Oban.Job{args: args}) when args == %{} do
integrations = Integrations.list_enabled_integrations("preseem")
results = Enum.map(integrations, &sync_integration/1)
eligible = Enum.filter(integrations, &should_sync?/1)
synced = Enum.count(results, &match?({:ok, _}, &1))
failed = Enum.count(results, &match?({:error, _}, &1))
skipped = Enum.count(results, &(&1 == :skipped))
now = DateTime.utc_now()
if synced > 0 or failed > 0 do
Logger.info("Preseem sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
enqueued =
Enum.count(eligible, fn integration ->
offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
scheduled_at = DateTime.add(now, offset, :second)
case %{"integration_id" => integration.id}
|> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
|> Oban.insert() do
{:ok, _job} -> true
{:error, _reason} -> false
end
end)
if enqueued > 0 do
Logger.info("Preseem sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
end
:ok
end
defp sync_integration(integration) do
if should_sync?(integration) do
case Sync.sync_organization(integration) do
{:ok, result} ->
Logger.info("Preseem sync completed for org #{integration.organization_id}: #{inspect(result)}")
{:ok, result}
# Individual sync: loads integration by ID and syncs
def perform(%Oban.Job{args: %{"integration_id" => id}}) do
case Integrations.get_integration_by_id(id) do
{:ok, integration} ->
sync_integration(integration)
{:error, reason} ->
Logger.error("Preseem sync failed for org #{integration.organization_id}: #{inspect(reason)}")
{:error, reason}
end
else
:skipped
{:error, :not_found} ->
Logger.warning("Preseem sync: integration #{id} not found, skipping")
{:error, :not_found}
end
end
defp sync_integration(integration) do
case Sync.sync_organization(integration) do
{:ok, result} ->
Logger.info("Preseem sync completed for org #{integration.organization_id}: #{inspect(result)}")
:ok
{:error, reason} ->
Logger.error("Preseem sync failed for org #{integration.organization_id}: #{inspect(reason)}")
:ok
end
end

1
lib/towerops_web/live/org/integrations_live.ex
View file

@ -23,6 +23,7 @@ defmodule ToweropsWeb.Org.IntegrationsLive do
{:ok,
socket
|> assign(:organization, organization)
|> assign(:timezone, socket.assigns.current_scope.timezone)
|> assign(:providers, @providers)
|> assign(:integrations, integrations)
|> assign(:configuring, nil)

10
lib/towerops_web/live/org/integrations_live.html.heex
View file

@ -51,10 +51,12 @@
<%= if integration.last_synced_at do %>
<span class="text-xs text-gray-500 dark:text-gray-400">
Last synced: {Calendar.strftime(
integration.last_synced_at,
"%Y-%m-%d %H:%M UTC"
)}
Last synced:
<.timestamp
datetime={integration.last_synced_at}
timezone={@timezone}
format="absolute"
/>
</span>
<% end %>

103
test/towerops/preseem/client_test.exs
View file

@ -6,10 +6,16 @@ defmodule Towerops.Preseem.ClientTest do
describe "test_connection/1" do
test "returns ok when API responds with 200" do
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{"status" => "ok", "account" => "test-isp"})
assert conn.request_path == "/model/v1/access_points"
assert conn.query_string =~ "limit=100"
Req.Test.json(conn, %{
"data" => [%{"id" => 1, "name" => "Test AP"}],
"paginator" => %{"page" => 1, "page_count" => 1, "limit" => 100, "total_count" => 1}
})
end)
assert {:ok, %{"status" => "ok"}} = Client.test_connection("valid-api-key")
assert {:ok, _body} = Client.test_connection("valid-api-key")
end
test "returns error for 401 unauthorized" do
@ -34,19 +40,60 @@ defmodule Towerops.Preseem.ClientTest do
end
describe "list_access_points/1" do
test "returns list of access points on success" do
test "returns list of access point scores on success" do
Req.Test.stub(Client, fn conn ->
assert conn.request_path == "/model/v1/access_point_scores"
Req.Test.json(conn, [
%{
"element_uuid" => "uuid-1",
"name" => "Tower1-AP1",
"system_mac_address" => "AA:BB:CC:DD:EE:FF",
"management_ip" => "10.0.0.1",
"model_name" => "AF60",
"firmware" => "4.3.2",
"rf_score" => 78.0,
"ap_health_score" => 92.5,
"subscriber_capacity" => 85,
"connections" => 30,
"airtime_remaining_tx" => 55
},
%{
"element_uuid" => "uuid-2",
"name" => "Tower1-AP2",
"management_ip" => "10.0.0.2"
}
])
end)
assert {:ok, [ap1, ap2]} = Client.list_access_points("valid-key")
assert ap1["element_uuid"] == "uuid-1"
assert ap2["element_uuid"] == "uuid-2"
end
test "handles wrapped response with data key" do
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{
"data" => [
%{"id" => "ap-1", "name" => "Tower1-AP1", "ip" => "10.0.0.1"},
%{"id" => "ap-2", "name" => "Tower1-AP2", "ip" => "10.0.0.2"}
%{"element_uuid" => "uuid-1", "name" => "AP1"}
]
})
end)
assert {:ok, [ap1, ap2]} = Client.list_access_points("valid-key")
assert ap1["id"] == "ap-1"
assert ap2["id"] == "ap-2"
assert {:ok, [ap]} = Client.list_access_points("valid-key")
assert ap["element_uuid"] == "uuid-1"
end
test "handles single object response by wrapping in list" do
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{
"element_uuid" => "uuid-1",
"name" => "Single AP"
})
end)
assert {:ok, [ap]} = Client.list_access_points("valid-key")
assert ap["element_uuid"] == "uuid-1"
end
test "returns error for 401" do
@ -58,21 +105,37 @@ defmodule Towerops.Preseem.ClientTest do
end
end
describe "get_access_point_metrics/2" do
test "returns metrics for an AP" do
describe "list_access_points_basic/1" do
test "fetches all pages of access points" do
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{
"data" => %{
"qoe_score" => 85.5,
"capacity_score" => 72.0,
"subscriber_count" => 42,
"p95_latency" => 18.3
}
})
assert conn.request_path == "/model/v1/access_points"
params = Plug.Conn.fetch_query_params(conn).query_params
case params["page"] do
nil ->
Req.Test.json(conn, %{
"data" => [%{"id" => 1, "name" => "AP1"}],
"paginator" => %{"page" => 1, "page_count" => 2, "limit" => 500, "total_count" => 2}
})
"2" ->
Req.Test.json(conn, %{
"data" => [%{"id" => 2, "name" => "AP2"}],
"paginator" => %{"page" => 2, "page_count" => 2, "limit" => 500, "total_count" => 2}
})
end
end)
assert {:ok, metrics} = Client.get_access_point_metrics("valid-key", "ap-1")
assert metrics["qoe_score"] == 85.5
assert {:ok, aps} = Client.list_access_points_basic("valid-key")
assert length(aps) == 2
end
test "returns error for 401" do
Req.Test.stub(Client, fn conn ->
conn |> Plug.Conn.put_status(401) |> Req.Test.json(%{"error" => "unauthorized"})
end)
assert {:error, :unauthorized} = Client.list_access_points_basic("bad-key")
end
end
end

56
test/towerops/preseem/device_matcher_test.exs
View file

@ -45,7 +45,7 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap_data =
Map.merge(
%{
"id" => "preseem-#{System.unique_integer([:positive])}",
"element_uuid" => "preseem-#{System.unique_integer([:positive])}",
"name" => "Test AP"
},
attrs
@ -61,9 +61,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-mac-match",
"element_uuid" => "ap-mac-match",
"name" => "MAC Match AP",
"mac_address" => "AA:BB:CC:DD:EE:FF"
"system_mac_address" => "AA:BB:CC:DD:EE:FF"
})
assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap)
@ -76,9 +76,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-mac-normalize",
"element_uuid" => "ap-mac-normalize",
"name" => "MAC Normalize AP",
"mac_address" => "11-22-33-44-55-66"
"system_mac_address" => "11-22-33-44-55-66"
})
assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap)
@ -90,10 +90,10 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-ip-match",
"element_uuid" => "ap-ip-match",
"name" => "IP Match AP",
"mac_address" => "FF:FF:FF:FF:FF:FF",
"ip_address" => "10.0.0.50"
"system_mac_address" => "FF:FF:FF:FF:FF:FF",
"management_ip" => "10.0.0.50"
})
assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap)
@ -111,10 +111,10 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-hostname-match",
"element_uuid" => "ap-hostname-match",
"name" => "Tower1-Sector-A.local",
"mac_address" => "FF:FF:FF:FF:FF:FE",
"ip_address" => "172.16.0.99"
"system_mac_address" => "FF:FF:FF:FF:FF:FE",
"management_ip" => "172.16.0.99"
})
assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap)
@ -129,9 +129,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-ambiguous-mac",
"element_uuid" => "ap-ambiguous-mac",
"name" => "Ambiguous MAC AP",
"mac_address" => "AA:AA:AA:AA:AA:AA"
"system_mac_address" => "AA:AA:AA:AA:AA:AA"
})
assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap)
@ -148,9 +148,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-ambiguous-ip",
"element_uuid" => "ap-ambiguous-ip",
"name" => "Ambiguous IP AP",
"ip_address" => "10.0.2.1"
"management_ip" => "10.0.2.1"
})
assert {:ok, matched_ap} = DeviceMatcher.match_access_point(ap)
@ -163,9 +163,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-manual",
"element_uuid" => "ap-manual",
"name" => "Manual AP",
"ip_address" => "10.0.3.1"
"management_ip" => "10.0.3.1"
})
# Manually set the match
@ -183,10 +183,10 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
test "returns unmatched when no devices match", %{org: org} do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-no-match",
"element_uuid" => "ap-no-match",
"name" => "Unmatched AP",
"mac_address" => "00:00:00:00:00:01",
"ip_address" => "172.31.255.255"
"system_mac_address" => "00:00:00:00:00:01",
"management_ip" => "172.31.255.255"
})
assert {:ok, unmatched_ap} = DeviceMatcher.match_access_point(ap)
@ -202,23 +202,23 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
_ap1 =
create_preseem_ap(org.id, %{
"id" => "ap-bulk-1",
"element_uuid" => "ap-bulk-1",
"name" => "Bulk AP 1",
"ip_address" => "10.0.10.1"
"management_ip" => "10.0.10.1"
})
_ap2 =
create_preseem_ap(org.id, %{
"id" => "ap-bulk-2",
"element_uuid" => "ap-bulk-2",
"name" => "Bulk AP 2",
"ip_address" => "10.0.10.2"
"management_ip" => "10.0.10.2"
})
_ap3 =
create_preseem_ap(org.id, %{
"id" => "ap-bulk-3",
"element_uuid" => "ap-bulk-3",
"name" => "Bulk AP 3 No Match",
"ip_address" => "172.16.0.100"
"management_ip" => "172.16.0.100"
})
assert {:ok, matched_count} = DeviceMatcher.match_unmatched(org.id)
@ -235,9 +235,9 @@ defmodule Towerops.Preseem.DeviceMatcherTest do
ap =
create_preseem_ap(org.id, %{
"id" => "ap-already-matched",
"element_uuid" => "ap-already-matched",
"name" => "Already Matched AP",
"ip_address" => "10.0.11.1"
"management_ip" => "10.0.11.1"
})
# Manually match it first

148
test/towerops/preseem/sync_test.exs
View file

@ -8,7 +8,6 @@ defmodule Towerops.Preseem.SyncTest do
alias Towerops.Integrations.Integration
alias Towerops.Preseem.AccessPoint
alias Towerops.Preseem.Client
alias Towerops.Preseem.SubscriberMetric
alias Towerops.Preseem.Sync
alias Towerops.Preseem.SyncLog
@ -21,20 +20,19 @@ defmodule Towerops.Preseem.SyncTest do
end
describe "upsert_access_point/2" do
test "creates a new access point from API data", %{org: org} do
test "creates a new access point from scores API data", %{org: org} do
ap_data = %{
"id" => "preseem-ap-100",
"element_uuid" => "preseem-ap-100",
"name" => "Tower1-AP1",
"mac_address" => "AA:BB:CC:DD:EE:FF",
"ip_address" => "10.0.1.1",
"model" => "AF60",
"system_mac_address" => "AA:BB:CC:DD:EE:FF",
"management_ip" => "10.0.1.1",
"model_name" => "AF60",
"firmware" => "4.3.2",
"capacity_score" => 85.0,
"qoe_score" => 92.5,
"subscriber_capacity" => 85,
"ap_health_score" => 92.5,
"rf_score" => 78.0,
"busy_hours" => 6,
"airtime_utilization" => 45.2,
"subscriber_count" => 30
"airtime_remaining_tx" => 45,
"connections" => 30
}
assert {:ok, ap} = Sync.upsert_access_point(org.id, ap_data)
@ -47,8 +45,8 @@ defmodule Towerops.Preseem.SyncTest do
assert ap.capacity_score == 85.0
assert ap.qoe_score == 92.5
assert ap.rf_score == 78.0
assert ap.busy_hours == 6
assert ap.airtime_utilization == 45.2
assert ap.busy_hours == nil
assert ap.airtime_utilization == 45.0
assert ap.subscriber_count == 30
assert ap.organization_id == org.id
assert ap.match_confidence == "unmatched"
@ -57,18 +55,17 @@ defmodule Towerops.Preseem.SyncTest do
test "upserts an existing access point", %{org: org} do
ap_data = %{
"id" => "preseem-ap-200",
"element_uuid" => "preseem-ap-200",
"name" => "Tower1-AP2",
"mac_address" => "11:22:33:44:55:66",
"ip_address" => "10.0.1.2",
"model" => "AF60",
"system_mac_address" => "11:22:33:44:55:66",
"management_ip" => "10.0.1.2",
"model_name" => "AF60",
"firmware" => "4.3.1",
"capacity_score" => 70.0,
"qoe_score" => 80.0,
"subscriber_capacity" => 70,
"ap_health_score" => 80.0,
"rf_score" => 65.0,
"busy_hours" => 4,
"airtime_utilization" => 30.0,
"subscriber_count" => 20
"airtime_remaining_tx" => 30,
"connections" => 20
}
assert {:ok, original} = Sync.upsert_access_point(org.id, ap_data)
@ -77,8 +74,8 @@ defmodule Towerops.Preseem.SyncTest do
Map.merge(ap_data, %{
"name" => "Tower1-AP2-Updated",
"firmware" => "4.3.2",
"qoe_score" => 88.0,
"subscriber_count" => 25
"ap_health_score" => 88.0,
"connections" => 25
})
assert {:ok, updated} = Sync.upsert_access_point(org.id, updated_data)
@ -91,10 +88,10 @@ defmodule Towerops.Preseem.SyncTest do
test "stores full API response as raw_data", %{org: org} do
ap_data = %{
"id" => "preseem-ap-300",
"element_uuid" => "preseem-ap-300",
"name" => "Tower1-AP3",
"extra_field" => "some_value",
"nested" => %{"key" => "value"}
"site_name" => "Main Tower",
"status" => "online"
}
assert {:ok, ap} = Sync.upsert_access_point(org.id, ap_data)
@ -106,7 +103,7 @@ defmodule Towerops.Preseem.SyncTest do
test "creates a subscriber metric record", %{org: org} do
{:ok, ap} =
Sync.upsert_access_point(org.id, %{
"id" => "preseem-ap-metrics",
"element_uuid" => "preseem-ap-metrics",
"name" => "Metrics AP"
})
@ -138,38 +135,34 @@ defmodule Towerops.Preseem.SyncTest do
integration: integration
} do
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{
"data" => [
%{
"id" => "ap-sync-1",
"name" => "Sync AP 1",
"mac_address" => "AA:BB:CC:11:22:33",
"ip_address" => "10.0.0.1",
"model" => "AF60",
"firmware" => "4.3.2",
"capacity_score" => 85.0,
"qoe_score" => 90.0,
"rf_score" => 75.0,
"busy_hours" => 5,
"airtime_utilization" => 40.0,
"subscriber_count" => 25
},
%{
"id" => "ap-sync-2",
"name" => "Sync AP 2",
"mac_address" => "AA:BB:CC:44:55:66",
"ip_address" => "10.0.0.2",
"model" => "LTU-Rocket",
"firmware" => "2.6.1",
"capacity_score" => 70.0,
"qoe_score" => 82.0,
"rf_score" => 68.0,
"busy_hours" => 3,
"airtime_utilization" => 35.0,
"subscriber_count" => 18
}
]
})
Req.Test.json(conn, [
%{
"element_uuid" => "ap-sync-1",
"name" => "Sync AP 1",
"system_mac_address" => "AA:BB:CC:11:22:33",
"management_ip" => "10.0.0.1",
"model_name" => "AF60",
"firmware" => "4.3.2",
"subscriber_capacity" => 85,
"ap_health_score" => 90.0,
"rf_score" => 75.0,
"airtime_remaining_tx" => 40,
"connections" => 25
},
%{
"element_uuid" => "ap-sync-2",
"name" => "Sync AP 2",
"system_mac_address" => "AA:BB:CC:44:55:66",
"management_ip" => "10.0.0.2",
"model_name" => "LTU-Rocket",
"firmware" => "2.6.1",
"subscriber_capacity" => 70,
"ap_health_score" => 82.0,
"rf_score" => 68.0,
"airtime_remaining_tx" => 35,
"connections" => 18
}
])
end)
assert {:ok, result} = Sync.sync_organization(integration)
@ -212,40 +205,5 @@ defmodule Towerops.Preseem.SyncTest do
updated_integration = Repo.get!(Integration, integration.id)
assert updated_integration.last_sync_status == "failed"
end
test "syncs access points with metrics when present", %{
org: org,
integration: integration
} do
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{
"data" => [
%{
"id" => "ap-with-metrics",
"name" => "Metrics AP",
"ip_address" => "10.0.0.5",
"metrics" => %{
"avg_latency" => 15.0,
"avg_jitter" => 2.5,
"avg_loss" => 0.3,
"avg_throughput" => 200.0,
"p95_latency" => 30.0,
"subscriber_count" => 50,
"recorded_at" => "2026-02-12T12:00:00Z"
}
}
]
})
end)
assert {:ok, result} = Sync.sync_organization(integration)
assert result.synced == 1
# Verify metrics were inserted
[ap] = Repo.all(from ap in AccessPoint, where: ap.organization_id == ^org.id)
metrics = Repo.all(from m in SubscriberMetric, where: m.preseem_access_point_id == ^ap.id)
assert length(metrics) == 1
assert hd(metrics).avg_latency == 15.0
end
end
end

117
test/towerops/workers/preseem_sync_worker_test.exs
View file

@ -15,28 +15,38 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do
%{org: org}
end
describe "perform/1" do
test "syncs enabled preseem integrations", %{org: org} do
describe "perform/1 dispatcher (cron)" do
test "enqueues individual sync jobs for enabled integrations", %{org: org} do
integration = integration_fixture(org.id)
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{
"data" => [
%{
"id" => "ap-1",
"name" => "Test AP",
"mac_address" => "AA:BB:CC:DD:EE:FF",
"ip_address" => "10.0.0.1"
}
]
})
end)
assert :ok = perform_job(PreseemSyncWorker, %{})
updated = Repo.reload!(integration)
assert updated.last_synced_at
assert updated.last_sync_status in ["success", "partial"]
# Should have enqueued a job with this integration's ID
assert [job] = all_enqueued(worker: PreseemSyncWorker)
assert job.args == %{"integration_id" => integration.id}
assert job.scheduled_at
end
test "staggers jobs across the 600-second window", %{org: org} do
# Create a second org with its own integration
user2 = user_fixture()
org2 = organization_fixture(user2.id)
_int1 = integration_fixture(org.id)
_int2 = integration_fixture(org2.id)
assert :ok = perform_job(PreseemSyncWorker, %{})
jobs = all_enqueued(worker: PreseemSyncWorker)
assert length(jobs) == 2
# Jobs should have scheduled_at times within 0-600 seconds from now
now = DateTime.utc_now()
for job <- jobs do
diff = DateTime.diff(job.scheduled_at, now, :second)
assert diff >= 0 and diff < 600, "Expected offset in [0, 600), got #{diff}"
end
end
test "skips integrations synced recently", %{org: org} do
@ -46,24 +56,15 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do
last_sync_status: "success"
})
# No Client stub needed - should skip without calling API
assert :ok = perform_job(PreseemSyncWorker, %{})
# No jobs should be enqueued for recently-synced integrations
assert [] = all_enqueued(worker: PreseemSyncWorker)
end
test "handles no enabled integrations" do
assert :ok = perform_job(PreseemSyncWorker, %{})
end
test "handles API errors without crashing", %{org: org} do
_integration = integration_fixture(org.id)
Req.Test.stub(Client, fn conn ->
conn
|> Plug.Conn.put_status(401)
|> Req.Test.json(%{"error" => "unauthorized"})
end)
assert :ok = perform_job(PreseemSyncWorker, %{})
assert [] = all_enqueued(worker: PreseemSyncWorker)
end
test "respects sync_interval_minutes", %{org: org} do
@ -80,11 +81,11 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do
last_sync_status: "success"
})
# No Client stub needed - should skip without calling API
assert :ok = perform_job(PreseemSyncWorker, %{})
assert [] = all_enqueued(worker: PreseemSyncWorker)
end
test "syncs when sync interval has elapsed", %{org: org} do
test "enqueues job when sync interval has elapsed", %{org: org} do
# Last synced 15 minutes ago with a 10-minute interval -> should sync
fifteen_minutes_ago =
DateTime.utc_now()
@ -98,28 +99,60 @@ defmodule Towerops.Workers.PreseemSyncWorkerTest do
last_sync_status: "success"
})
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{"data" => []})
end)
assert :ok = perform_job(PreseemSyncWorker, %{})
updated = Repo.reload!(integration)
assert DateTime.after?(updated.last_synced_at, fifteen_minutes_ago)
assert [job] = all_enqueued(worker: PreseemSyncWorker)
assert job.args == %{"integration_id" => integration.id}
end
test "syncs integrations that have never been synced", %{org: org} do
test "enqueues job for integrations that have never been synced", %{org: org} do
integration = integration_fixture(org.id)
assert integration.last_synced_at == nil
assert :ok = perform_job(PreseemSyncWorker, %{})
assert [job] = all_enqueued(worker: PreseemSyncWorker)
assert job.args == %{"integration_id" => integration.id}
end
end
describe "perform/1 individual sync" do
test "syncs the given integration", %{org: org} do
integration = integration_fixture(org.id)
Req.Test.stub(Client, fn conn ->
Req.Test.json(conn, %{"data" => []})
Req.Test.json(conn, [
%{
"element_uuid" => "ap-1",
"name" => "Test AP",
"system_mac_address" => "AA:BB:CC:DD:EE:FF",
"management_ip" => "10.0.0.1"
}
])
end)
assert :ok = perform_job(PreseemSyncWorker, %{})
assert :ok = perform_job(PreseemSyncWorker, %{"integration_id" => integration.id})
updated = Repo.reload!(integration)
assert updated.last_synced_at
assert updated.last_sync_status in ["success", "partial"]
end
test "handles API errors without crashing", %{org: org} do
integration = integration_fixture(org.id)
Req.Test.stub(Client, fn conn ->
conn
|> Plug.Conn.put_status(401)
|> Req.Test.json(%{"error" => "unauthorized"})
end)
assert :ok = perform_job(PreseemSyncWorker, %{"integration_id" => integration.id})
end
test "returns error for nonexistent integration" do
assert {:error, :not_found} =
perform_job(PreseemSyncWorker, %{"integration_id" => Ecto.UUID.generate()})
end
end
end

5
test/towerops_web/live/org/integrations_live_test.exs
View file

@ -61,7 +61,10 @@ defmodule ToweropsWeb.Org.IntegrationsLiveTest do
test "can test preseem connection successfully", %{conn: conn, user: user, organization: org} do
Req.Test.stub(PreseemClient, fn conn ->
Req.Test.json(conn, %{"status" => "ok"})
Req.Test.json(conn, %{
"data" => [%{"id" => 1, "name" => "Test AP"}],
"paginator" => %{"page" => 1, "page_count" => 1, "limit" => 100, "total_count" => 1}
})
end)
{:ok, view, _html} =