Graham McIntire
52cde2cbc5
Sonar, Splynx, VISP, NetBox, and Gaiia sync workers all followed the same
"cron runs, then Enum.map syncs every integration in the same job" shape.
A single slow tenant — anything from an upstream rate limit to a stuck
GraphQL call — would hold the maintenance queue slot for ten minutes
while every other integration waited in line.
Each worker now uses the dispatcher/per-integration pattern that
UispSyncWorker already had:
- the cron `perform(%Oban.Job{args: %{}})` fans out one job per eligible
integration via `PollingOffset.calculate_offset/2`, staggered across a
5- to 10-minute window with `unique: [period: window_seconds]`
- `perform(%Oban.Job{args: %{"integration_id" => id}})` does the actual
sync, classifying failures via `Towerops.Workers.SyncErrors.transient?/1`
so permanent errors (auth, 4xx) don't get retried forever
Worker-specific defaults preserved:
- gaiia / netbox: 30/600s window, longer interval
- sonar / splynx / visp: 10/300s window
Tests updated to match the two-stage dispatcher pattern — `perform_job/2`
with empty args now asserts the per-integration jobs got enqueued and
the actual sync is invoked via a separate `perform_job/2` call with the
integration_id arg.
109 lines
3.3 KiB
Elixir
109 lines
3.3 KiB
Elixir
defmodule Towerops.Workers.GaiiaSyncWorkerTest do
|
|
use Towerops.DataCase, async: false
|
|
use Oban.Testing, repo: Towerops.Repo
|
|
|
|
import Towerops.AccountsFixtures
|
|
import Towerops.IntegrationsFixtures
|
|
import Towerops.OrganizationsFixtures
|
|
|
|
alias Towerops.Gaiia.Client
|
|
alias Towerops.Workers.GaiiaSyncWorker
|
|
|
|
setup do
|
|
user = user_fixture()
|
|
org = organization_fixture(user.id)
|
|
integration = integration_fixture(org.id, %{provider: "gaiia"})
|
|
%{org: org, integration: integration}
|
|
end
|
|
|
|
describe "perform/1 dispatcher (empty args)" do
|
|
test "enqueues a per-integration sync job for each eligible integration", %{integration: integration} do
|
|
assert :ok = perform_job(GaiiaSyncWorker, %{})
|
|
|
|
jobs = all_enqueued(worker: GaiiaSyncWorker)
|
|
assert length(jobs) == 1
|
|
[job] = jobs
|
|
assert job.args == %{"integration_id" => integration.id}
|
|
end
|
|
|
|
test "skips recently synced integrations", %{integration: integration} do
|
|
# Mark as recently synced — the dispatcher should now enqueue nothing.
|
|
Towerops.Integrations.update_sync_status(integration, "success")
|
|
|
|
assert :ok = perform_job(GaiiaSyncWorker, %{})
|
|
assert [] = all_enqueued(worker: GaiiaSyncWorker)
|
|
end
|
|
end
|
|
|
|
describe "perform/1 (with integration_id)" do
|
|
test "syncs a single enabled gaiia integration", %{org: org, integration: integration} do
|
|
stub_all_endpoints()
|
|
|
|
assert :ok = perform_job(GaiiaSyncWorker, %{"integration_id" => integration.id})
|
|
|
|
# Verify data was synced
|
|
assert length(Towerops.Gaiia.list_accounts(org.id)) == 1
|
|
end
|
|
|
|
test "returns {:error, :not_found} for a missing integration" do
|
|
missing_id = Ecto.UUID.generate()
|
|
|
|
ExUnit.CaptureLog.capture_log(fn ->
|
|
assert {:error, :not_found} =
|
|
perform_job(GaiiaSyncWorker, %{"integration_id" => missing_id})
|
|
end)
|
|
end
|
|
end
|
|
|
|
defp stub_all_endpoints do
|
|
Req.Test.stub(Client, fn conn ->
|
|
{:ok, body, _conn} = Plug.Conn.read_body(conn)
|
|
decoded = Jason.decode!(body)
|
|
query = decoded["query"]
|
|
|
|
response =
|
|
cond do
|
|
String.contains?(query, "ListAccounts") -> accounts_response()
|
|
String.contains?(query, "ListNetworkSites") -> empty_response("networkSites")
|
|
String.contains?(query, "ListInventoryItems") -> empty_response("inventoryItems")
|
|
String.contains?(query, "ListBillingSubscriptions") -> empty_response("billingSubscriptions")
|
|
true -> %{"data" => %{}}
|
|
end
|
|
|
|
Req.Test.json(conn, response)
|
|
end)
|
|
end
|
|
|
|
defp accounts_response do
|
|
%{
|
|
"data" => %{
|
|
"accounts" => %{
|
|
"edges" => [
|
|
%{
|
|
"node" => %{
|
|
"id" => "acct-1",
|
|
"name" => "Test Account",
|
|
"status" => %{"id" => "status-1", "name" => "Active", "type" => "ACTIVE"},
|
|
"type" => %{"id" => "type-1", "name" => "Residential"},
|
|
"billingSubscriptions" => %{"edges" => []},
|
|
"readableId" => "1001"
|
|
}
|
|
}
|
|
],
|
|
"pageInfo" => %{"hasNextPage" => false, "endCursor" => nil}
|
|
}
|
|
}
|
|
}
|
|
end
|
|
|
|
defp empty_response(root_key) do
|
|
%{
|
|
"data" => %{
|
|
root_key => %{
|
|
"edges" => [],
|
|
"pageInfo" => %{"hasNextPage" => false, "endCursor" => nil}
|
|
}
|
|
}
|
|
}
|
|
end
|
|
end
|