273 lines
8.3 KiB
Elixir
273 lines
8.3 KiB
Elixir
defmodule Towerops.JobMonitoringTest do
|
|
use Towerops.DataCase, async: true
|
|
|
|
import Towerops.DevicesFixtures
|
|
import Towerops.JobsFixtures
|
|
|
|
alias Towerops.JobMonitoring
|
|
|
|
describe "list_active_jobs/0" do
|
|
test "returns jobs in executing state for polling and discovery workers" do
|
|
device = device_fixture()
|
|
|
|
# Create jobs with different attempted_at times
|
|
older_job =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
# 60 seconds ago
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -60, :second)
|
|
})
|
|
|
|
newer_job =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
# Now
|
|
attempted_at: DateTime.utc_now()
|
|
})
|
|
|
|
# Create completed job (should not be returned)
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id}
|
|
})
|
|
|
|
jobs = JobMonitoring.list_active_jobs()
|
|
|
|
assert length(jobs) == 2
|
|
assert Enum.all?(jobs, fn j -> j.state == "executing" end)
|
|
|
|
# Verify ascending order by attempted_at
|
|
assert hd(jobs).id == older_job.id
|
|
assert List.last(jobs).id == newer_job.id
|
|
end
|
|
|
|
test "excludes jobs from other workers" do
|
|
device = device_fixture()
|
|
|
|
# Create job from non-monitored worker
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.SomeOtherWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.utc_now()
|
|
})
|
|
|
|
# Create job from monitored worker
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.utc_now()
|
|
})
|
|
|
|
jobs = JobMonitoring.list_active_jobs()
|
|
|
|
# Should only return the DevicePollerWorker job
|
|
assert length(jobs) == 1
|
|
assert hd(jobs).worker == "Towerops.Workers.DevicePollerWorker"
|
|
end
|
|
end
|
|
|
|
describe "list_stuck_jobs/0" do
|
|
test "returns polling jobs executing longer than 2 minutes" do
|
|
device = device_fixture()
|
|
|
|
# Stuck polling job (3 minutes ago)
|
|
stuck_job =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
|
|
})
|
|
|
|
# Recent polling job (30 seconds ago) - not stuck
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -30, :second)
|
|
})
|
|
|
|
stuck = JobMonitoring.list_stuck_jobs()
|
|
|
|
assert length(stuck) == 1
|
|
assert hd(stuck).id == stuck_job.id
|
|
end
|
|
|
|
test "returns discovery jobs executing longer than 5 minutes" do
|
|
device = device_fixture()
|
|
|
|
# Stuck discovery job (6 minutes ago)
|
|
stuck_job =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -360, :second)
|
|
})
|
|
|
|
# Recent discovery job (3 minutes ago) - not stuck
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
|
|
})
|
|
|
|
stuck = JobMonitoring.list_stuck_jobs()
|
|
|
|
assert length(stuck) == 1
|
|
assert hd(stuck).id == stuck_job.id
|
|
end
|
|
|
|
test "returns stuck jobs from both worker types ordered by attempted_at" do
|
|
device = device_fixture()
|
|
|
|
# Stuck polling job (3 minutes ago)
|
|
stuck_poller =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -180, :second)
|
|
})
|
|
|
|
# Stuck discovery job (6 minutes ago - older, should be first)
|
|
stuck_discovery =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -360, :second)
|
|
})
|
|
|
|
# Recent jobs that should not be included
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -30, :second)
|
|
})
|
|
|
|
stuck = JobMonitoring.list_stuck_jobs()
|
|
|
|
assert length(stuck) == 2
|
|
# Verify oldest first (discovery job at -360s)
|
|
assert Enum.at(stuck, 0).id == stuck_discovery.id
|
|
assert Enum.at(stuck, 1).id == stuck_poller.id
|
|
end
|
|
|
|
test "excludes jobs just under the threshold boundary" do
|
|
device = device_fixture()
|
|
|
|
# Polling job at 119 seconds - should NOT be stuck (under 120s threshold)
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -119, :second)
|
|
})
|
|
|
|
# Discovery job at 299 seconds - should NOT be stuck (under 300s threshold)
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -299, :second)
|
|
})
|
|
|
|
assert JobMonitoring.list_stuck_jobs() == []
|
|
end
|
|
|
|
test "includes jobs just over the threshold" do
|
|
device = device_fixture()
|
|
|
|
# Polling job at 125 seconds - should be stuck (over 120s threshold)
|
|
stuck_poller =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -125, :second)
|
|
})
|
|
|
|
# Discovery job at 305 seconds - should be stuck (over 300s threshold)
|
|
stuck_discovery =
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id},
|
|
attempted_at: DateTime.add(DateTime.utc_now(), -305, :second)
|
|
})
|
|
|
|
stuck = JobMonitoring.list_stuck_jobs()
|
|
|
|
assert length(stuck) == 2
|
|
assert stuck |> Enum.map(& &1.id) |> Enum.sort() == Enum.sort([stuck_discovery.id, stuck_poller.id])
|
|
end
|
|
end
|
|
|
|
describe "list_failed_jobs/0" do
|
|
test "returns retryable and cancelled jobs" do
|
|
device = device_fixture()
|
|
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "retryable",
|
|
args: %{"device_id" => device.id}
|
|
})
|
|
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DiscoveryWorker",
|
|
state: "cancelled",
|
|
args: %{"device_id" => device.id}
|
|
})
|
|
|
|
# Should not include completed
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id}
|
|
})
|
|
|
|
failed = JobMonitoring.list_failed_jobs()
|
|
|
|
assert length(failed) == 2
|
|
assert Enum.all?(failed, fn j -> j.state in ["retryable", "cancelled"] end)
|
|
end
|
|
end
|
|
|
|
describe "list_recent_completions/1" do
|
|
test "returns completed, cancelled, and discarded jobs up to limit" do
|
|
device = device_fixture()
|
|
|
|
# Create 5 completed jobs
|
|
for _ <- 1..5 do
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "completed",
|
|
args: %{"device_id" => device.id},
|
|
completed_at: DateTime.utc_now()
|
|
})
|
|
end
|
|
|
|
# Should not include executing
|
|
oban_job_fixture(%{
|
|
worker: "Towerops.Workers.DevicePollerWorker",
|
|
state: "executing",
|
|
args: %{"device_id" => device.id}
|
|
})
|
|
|
|
recent = JobMonitoring.list_recent_completions(3)
|
|
|
|
assert length(recent) == 3
|
|
assert Enum.all?(recent, fn j -> j.state == "completed" end)
|
|
end
|
|
end
|
|
end
|