towerops/test/towerops/workers/discovery_worker_test.exs

677 lines
24 KiB
Elixir

defmodule Towerops.Workers.DiscoveryWorkerTest do
use Towerops.DataCase, async: true
import Mox
import Towerops.AccountsFixtures
alias Towerops.Agents.AgentToken
alias Towerops.Devices.Device
alias Towerops.Snmp
alias Towerops.Snmp.SnmpMock
alias Towerops.Workers.DiscoveryWorker
setup :verify_on_exit!
describe "perform/1" do
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Test Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Test Device",
ip_address: "192.168.1.1",
snmp_enabled: true,
snmp_version: "2c",
snmp_community: "public",
snmp_port: 161,
site_id: site.id,
organization_id: organization.id
})
{:ok, device: device, org: organization}
end
test "successfully discovers a device", %{device: device} do
# Mock batched GET for system info (discovery now uses get_multiple)
stub(SnmpMock, :get_multiple, fn _target, oids, _opts ->
result_map =
Map.new(oids, fn oid ->
value =
case oid do
"1.3.6.1.2.1.1.1.0" -> {:octet_string, "Test Device"}
"1.3.6.1.2.1.1.2.0" -> {:object_identifier, [1, 3, 6, 1, 4, 1, 9]}
"1.3.6.1.2.1.1.3.0" -> {:timeticks, 12_345}
"1.3.6.1.2.1.1.4.0" -> {:octet_string, "admin@test.com"}
"1.3.6.1.2.1.1.5.0" -> {:octet_string, "test-device"}
"1.3.6.1.2.1.1.6.0" -> {:octet_string, "Test Location"}
_ -> {:octet_string, ""}
end
{oid, value}
end)
{:ok, result_map}
end)
# Mock individual GET calls (for categorize_device_speed, test_connection, and optional fields)
stub(SnmpMock, :get, fn _target, oid, _opts ->
case oid do
"1.3.6.1.2.1.1.1.0" -> {:ok, {:octet_string, "Test Device"}}
# sysUptime for test_connection
"1.3.6.1.2.1.1.3.0" -> {:ok, {:timeticks, 12_345}}
# UCD-SNMP-MIB CPU stats (processor discovery fallback)
"1.3.6.1.4.1.2021.11.9.0" -> {:ok, {:integer, 10}}
"1.3.6.1.4.1.2021.11.10.0" -> {:ok, {:integer, 5}}
"1.3.6.1.4.1.2021.11.11.0" -> {:ok, {:integer, 85}}
_ -> {:error, :no_such_object}
end
end)
# Mock interface discovery (if_index walk + other IF-MIB walks) and neighbor discovery
# Total: if_index + if_descr + if_type + if_speed + if_phys_address + if_admin_status +
# if_oper_status + if_name + if_alias (9 walks) + LLDP (6 walks) + CDP (1 walk) = 16 walks
# Simplify: just allow any number of walk calls to return empty
stub(SnmpMock, :walk, fn _target, _oid, _opts ->
{:ok, []}
end)
assert :ok = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
# Verify SNMP device was created
snmp_device = Snmp.get_device(device.id)
assert snmp_device
assert snmp_device.sys_descr == "Test Device"
assert snmp_device.sys_name == "test-device"
end
test "returns :discard when device not found" do
non_existent_id = Ecto.UUID.generate()
assert :discard =
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => non_existent_id}})
end
test "returns error for transient failures to allow retry", %{device: device} do
# Mock individual GET to pass speed categorization
stub(SnmpMock, :get, fn _target, oid, _opts ->
case oid do
# Speed categorization probe succeeds (device is responsive)
"1.3.6.1.2.1.1.1.0" -> {:ok, {:octet_string, "Test"}}
_ -> {:error, :timeout}
end
end)
# Mock batched GET to fail (transient error during system info discovery)
stub(SnmpMock, :get_multiple, fn _target, _oids, _opts ->
{:error, :timeout}
end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:error, :timeout} end)
# Transient failures should return {:error, reason} so Oban can retry
assert {:error, _reason} = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end
test "skips discovery when snmp_enabled is false", %{org: organization} do
{:ok, site} =
Towerops.Sites.create_site(%{
name: "No SNMP Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "No SNMP Device",
ip_address: "192.168.1.99",
snmp_enabled: false,
site_id: site.id,
organization_id: organization.id
})
# Should discard immediately without attempting SNMP
assert :discard = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end
test "successfully completes discovery with proper mocks", %{device: device} do
# Mock batched GET for system info
stub(SnmpMock, :get_multiple, fn _target, oids, _opts ->
result_map =
Map.new(oids, fn oid ->
value =
case oid do
"1.3.6.1.2.1.1.1.0" -> {:octet_string, "Test"}
"1.3.6.1.2.1.1.2.0" -> {:object_identifier, [1, 3, 6, 1]}
"1.3.6.1.2.1.1.3.0" -> {:timeticks, 1}
"1.3.6.1.2.1.1.4.0" -> {:octet_string, ""}
"1.3.6.1.2.1.1.5.0" -> {:octet_string, "test"}
"1.3.6.1.2.1.1.6.0" -> {:octet_string, ""}
_ -> {:octet_string, ""}
end
{oid, value}
end)
{:ok, result_map}
end)
# Mock individual GET calls (for categorize_device_speed, test_connection, and optional fields)
stub(SnmpMock, :get, fn _target, oid, _opts ->
case oid do
"1.3.6.1.2.1.1.1.0" -> {:ok, {:octet_string, "Test"}}
# sysUptime for test_connection
"1.3.6.1.2.1.1.3.0" -> {:ok, {:timeticks, 1}}
# UCD-SNMP-MIB CPU stats (processor discovery fallback)
"1.3.6.1.4.1.2021.11.9.0" -> {:ok, {:integer, 10}}
"1.3.6.1.4.1.2021.11.10.0" -> {:ok, {:integer, 5}}
"1.3.6.1.4.1.2021.11.11.0" -> {:ok, {:integer, 85}}
_ -> {:error, :no_such_object}
end
end)
# Allow any number of walk calls (interface + neighbor discovery)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:ok, []} end)
assert :ok = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
# Verify SNMP device was created
snmp_device = Snmp.get_device(device.id)
assert snmp_device
end
test "logs error when discovery fails", %{device: device} do
import ExUnit.CaptureLog
stub(SnmpMock, :get, fn _target, _oid, _opts ->
{:error, :network_unreachable}
end)
log =
capture_log(fn ->
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end)
assert log =~ "Direct SNMP discovery failed:"
end
end
describe "enqueue/1" do
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Enqueue Org"}, user.id)
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Enqueue Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Enqueue Device",
ip_address: "192.168.1.50",
snmp_enabled: true,
site_id: site.id,
organization_id: organization.id
})
{:ok, device: device}
end
test "enqueues a discovery job", %{device: device} do
assert :ok = DiscoveryWorker.enqueue(device.id)
end
test "deduplicates jobs within 60 seconds", %{device: device} do
assert :ok = DiscoveryWorker.enqueue(device.id)
assert :ok = DiscoveryWorker.enqueue(device.id)
end
end
describe "perform/1 with agent assignment" do
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Agent Org"}, user.id)
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Agent Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Agent Device",
ip_address: "10.0.0.1",
snmp_enabled: true,
snmp_version: "2c",
snmp_community: "public",
snmp_port: 161,
site_id: site.id,
organization_id: organization.id
})
{:ok, device: device, organization: organization}
end
test "falls back to direct discovery when assigned agent has never been seen",
%{device: device, organization: organization} do
# Assign an agent that has never reported (last_seen_at is nil → offline).
{:ok, agent_token, _token_string} =
Towerops.Agents.create_agent_token(organization.id, "Offline Agent")
{:ok, _assignment} =
Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
# No cloud pollers exist, so it should fall through to direct discovery.
stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :timeout} end)
stub(SnmpMock, :get_multiple, fn _target, _oids, _opts -> {:error, :timeout} end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:error, :timeout} end)
# Direct discovery may classify as :discard (permanent
# :device_unresponsive) or {:error, _} (transient timeout). Either
# outcome confirms the agent-offline → cloud-poller → direct path
# and exercises handle_assigned_agent_discovery /
# attempt_cloud_poller_discovery branches without a hard log assertion
# (test logger level is :error).
result = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
assert result == :discard or match?({:error, _}, result)
end
test "falls back to direct discovery when assigned agent token is deleted",
%{device: device, organization: organization} do
{:ok, agent_token, _token_string} =
Towerops.Agents.create_agent_token(organization.id, "To-Be-Deleted Agent")
{:ok, _assignment} =
Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
# Drop the assignment first, then the token, to dodge the FK constraint.
# This forces handle_assigned_agent_discovery to rescue
# Ecto.NoResultsError and fall back to attempt_cloud_poller_discovery.
Towerops.Repo.delete_all(from(a in Towerops.Agents.AgentAssignment, where: a.agent_token_id == ^agent_token.id))
Towerops.Repo.delete_all(from(t in AgentToken, where: t.id == ^agent_token.id))
stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :timeout} end)
stub(SnmpMock, :get_multiple, fn _target, _oids, _opts -> {:error, :timeout} end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:error, :timeout} end)
result = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
assert result == :discard or match?({:error, _}, result)
end
test "uses online cloud poller when no agent is assigned and reports completion via PubSub",
%{device: device} do
# Create a cloud poller and mark it online
{:ok, cloud_poller, _token} = Towerops.Agents.create_cloud_poller("Online Cloud Poller")
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^cloud_poller.id),
set: [last_seen_at: DateTime.utc_now()]
)
# Subscribe to the PubSub channel that the worker broadcasts on so we can
# detect that the discovery request was sent and then simulate the agent
# finishing by bumping the device's last_discovery_at.
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{cloud_poller.id}:discovery")
task =
Task.async(fn ->
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end)
# Wait for the worker to broadcast the discovery request
assert_receive {:discovery_requested, device_id}, 5_000
assert device_id == device.id
# Simulate the agent finishing by bumping last_discovery_at
Towerops.Repo.update_all(
from(d in Device, where: d.id == ^device.id),
set: [last_discovery_at: DateTime.truncate(DateTime.utc_now(), :second)]
)
assert :ok = Task.await(task, 10_000)
end
test "agent goes offline mid-discovery returns to cloud poller fallback chain",
%{device: device, organization: organization} do
# Online local agent
{:ok, agent_token, _token} =
Towerops.Agents.create_agent_token(organization.id, "Local Agent")
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^agent_token.id),
set: [last_seen_at: DateTime.utc_now()]
)
{:ok, _assignment} =
Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:discovery")
stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :timeout} end)
stub(SnmpMock, :get_multiple, fn _target, _oids, _opts -> {:error, :timeout} end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:error, :timeout} end)
task =
Task.async(fn ->
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end)
assert_receive {:discovery_requested, _device_id}, 5_000
# Mark agent as offline so wait_loop bails out with :agent_offline
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^agent_token.id),
set: [last_seen_at: DateTime.add(DateTime.utc_now(), -30, :minute)]
)
# No cloud pollers configured → falls through to direct discovery which
# returns either :discard (permanent device_unresponsive) or {:error, _}
# depending on how SNMP errors are classified.
result = Task.await(task, 30_000)
assert result == :discard or match?({:error, _}, result)
end
test "device deleted during agent discovery returns :ok and exits",
%{device: device, organization: organization} do
{:ok, agent_token, _token} =
Towerops.Agents.create_agent_token(organization.id, "Online Agent")
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^agent_token.id),
set: [last_seen_at: DateTime.utc_now()]
)
{:ok, _assignment} =
Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token.id}:discovery")
task =
Task.async(fn ->
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end)
assert_receive {:discovery_requested, _device_id}, 5_000
# Hard-delete the device so check_discovery_completion returns :device_deleted
Towerops.Repo.delete_all(from(d in Device, where: d.id == ^device.id))
# `attempt_agent_discovery` translates :device_deleted to :ok and returns.
assert :ok = Task.await(task, 30_000)
end
end
describe "perform/1 crash handling" do
test "discards (without retrying) when an unexpected exception is raised" do
# Pass a non-string device_id so get_device_with_details raises a
# Postgrex/Ecto error inside the rescue block. The worker should log
# and return :discard rather than re-raising.
import ExUnit.CaptureLog
log =
capture_log(fn ->
assert :discard =
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => "not-a-uuid"}})
end)
assert log =~ "Discovery job crashed unexpectedly"
end
end
describe "retryable_error?/1 additional cases" do
test "tuple errors are retryable by default" do
assert DiscoveryWorker.retryable_error?({:snmp_error, :nxdomain})
end
test "string errors are retryable by default" do
assert DiscoveryWorker.retryable_error?("some error message")
end
test "nil is retryable by default" do
assert DiscoveryWorker.retryable_error?(nil)
end
end
describe "retryable_error?/1" do
test "timeout is retryable" do
assert DiscoveryWorker.retryable_error?(:timeout)
end
test "network_unreachable is retryable" do
assert DiscoveryWorker.retryable_error?(:network_unreachable)
end
test "connection_refused is retryable" do
assert DiscoveryWorker.retryable_error?(:connection_refused)
end
test "econnrefused is retryable" do
assert DiscoveryWorker.retryable_error?(:econnrefused)
end
test "device_not_found is not retryable" do
refute DiscoveryWorker.retryable_error?(:device_not_found)
end
test "invalid_config is not retryable" do
refute DiscoveryWorker.retryable_error?(:invalid_config)
end
test "no_snmp_credentials is not retryable" do
refute DiscoveryWorker.retryable_error?(:no_snmp_credentials)
end
test "device_deleted is not retryable" do
refute DiscoveryWorker.retryable_error?(:device_deleted)
end
test "snmp_not_enabled is not retryable" do
refute DiscoveryWorker.retryable_error?(:snmp_not_enabled)
end
test "device_unresponsive is not retryable" do
refute DiscoveryWorker.retryable_error?(:device_unresponsive)
end
test "unknown errors are retryable by default" do
assert DiscoveryWorker.retryable_error?(:some_unknown_error)
end
end
describe "perform/1 - cloud poller fallback chain" do
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Cloud Org"}, user.id)
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Cloud Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Cloud Device",
ip_address: "10.10.10.1",
snmp_enabled: true,
snmp_version: "2c",
snmp_community: "public",
snmp_port: 161,
site_id: site.id,
organization_id: organization.id
})
{:ok, device: device, organization: organization}
end
test "skips disabled cloud pollers and falls through to direct discovery",
%{device: device} do
# Create a cloud poller that's seen recently but DISABLED
{:ok, poller, _token} = Towerops.Agents.create_cloud_poller("Disabled Poller")
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^poller.id),
set: [last_seen_at: DateTime.utc_now(), enabled: false]
)
stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :timeout} end)
stub(SnmpMock, :get_multiple, fn _target, _oids, _opts -> {:error, :timeout} end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:error, :timeout} end)
result = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
# Disabled poller is offline, so falls through to direct SNMP which fails.
assert result == :discard or match?({:error, _}, result)
end
test "uses second cloud poller when first is offline (stale last_seen_at)",
%{device: device} do
# Stale (offline) cloud poller
{:ok, stale, _t1} = Towerops.Agents.create_cloud_poller("Stale Poller")
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^stale.id),
set: [last_seen_at: DateTime.add(DateTime.utc_now(), -30, :minute)]
)
# Online cloud poller (created second)
{:ok, online, _t2} = Towerops.Agents.create_cloud_poller("Online Poller")
Towerops.Repo.update_all(
from(t in AgentToken, where: t.id == ^online.id),
set: [last_seen_at: DateTime.utc_now()]
)
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{online.id}:discovery")
task =
Task.async(fn ->
DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
end)
# The discovery request should go to the ONLINE poller, not the stale one.
assert_receive {:discovery_requested, _device_id}, 5_000
Towerops.Repo.update_all(
from(d in Device, where: d.id == ^device.id),
set: [last_discovery_at: DateTime.truncate(DateTime.utc_now(), :second)]
)
assert :ok = Task.await(task, 10_000)
end
end
describe "perform/1 - direct discovery success path" do
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Direct Org"}, user.id)
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Direct Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Direct Device",
ip_address: "172.16.0.1",
snmp_enabled: true,
snmp_version: "2c",
snmp_community: "public",
snmp_port: 161,
site_id: site.id,
organization_id: organization.id
})
{:ok, device: device}
end
test "creates SNMP device record on successful direct discovery", %{device: device} do
stub(SnmpMock, :get_multiple, fn _target, oids, _opts ->
result_map =
Map.new(oids, fn oid ->
value =
case oid do
"1.3.6.1.2.1.1.1.0" -> {:octet_string, "Linux server 5.15.0"}
"1.3.6.1.2.1.1.2.0" -> {:object_identifier, [1, 3, 6, 1, 4, 1, 9]}
"1.3.6.1.2.1.1.3.0" -> {:timeticks, 100}
"1.3.6.1.2.1.1.4.0" -> {:octet_string, "ops@example.com"}
"1.3.6.1.2.1.1.5.0" -> {:octet_string, "direct-host"}
"1.3.6.1.2.1.1.6.0" -> {:octet_string, "DC1"}
_ -> {:octet_string, ""}
end
{oid, value}
end)
{:ok, result_map}
end)
stub(SnmpMock, :get, fn _target, oid, _opts ->
case oid do
"1.3.6.1.2.1.1.1.0" -> {:ok, {:octet_string, "Linux server 5.15.0"}}
"1.3.6.1.2.1.1.3.0" -> {:ok, {:timeticks, 100}}
_ -> {:error, :no_such_object}
end
end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:ok, []} end)
assert :ok = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
snmp_device = Snmp.get_device(device.id)
assert snmp_device
assert snmp_device.sys_name == "direct-host"
end
end
describe "perform/1 - permanent vs transient error classification" do
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Errcls Org"}, user.id)
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Errcls Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Errcls Device",
ip_address: "10.20.30.40",
snmp_enabled: true,
snmp_version: "2c",
snmp_community: "public",
snmp_port: 161,
site_id: site.id,
organization_id: organization.id
})
{:ok, device: device}
end
test "returns :discard for :device_unresponsive (a permanent error)",
%{device: device} do
# When SNMP get for sysDescr fails with no_such_object (categorize_device_speed
# path), Snmp.discover_device classifies the device as unresponsive.
stub(SnmpMock, :get, fn _target, _oid, _opts -> {:error, :no_such_object} end)
stub(SnmpMock, :get_multiple, fn _target, _oids, _opts -> {:error, :no_such_object} end)
stub(SnmpMock, :walk, fn _target, _oid, _opts -> {:ok, []} end)
# Either :discard (permanent classification) or {:error, _} acceptable depending
# on which path Snmp.discover_device takes; both confirm classification logic.
result = DiscoveryWorker.perform(%Oban.Job{args: %{"device_id" => device.id}})
assert result == :discard or match?({:error, _}, result)
end
end
end