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

test: cloud poller AgentChannel + SNMP sensor v3/no-snmp + firmware month parser

Browse source
This commit is contained in:
Graham McIntire 2026-05-09 09:20:27 -05:00
parent 23a21a674d
commit b51759dfdf
3 changed files with 175 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
test/towerops/monitoring/executors/snmp_sensor_executor_test.exs
View file

@ -286,5 +286,40 @@ defmodule Towerops.Monitoring.Executors.SnmpSensorExecutorTest do
assert {:error, :timeout} = SnmpSensorExecutor.execute(check)
end
test "returns error when device has no SNMP device record", %{organization: organization} do
# Create a device with snmp_enabled but no Snmp.Device association inserted.
{: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: "10.99.99.99",
snmp_enabled: true,
snmp_version: "2c",
snmp_community: "public",
snmp_port: 161,
site_id: site.id,
organization_id: organization.id
})
{:ok, check} =
Monitoring.create_check(%{
organization_id: organization.id,
device_id: device.id,
name: "Bad device check",
check_type: "snmp_sensor",
source_type: "manual",
interval_seconds: 60,
enabled: true,
config: %{
"sensor_id" => Ecto.UUID.generate(),
"sensor_oid" => "1.2.3.4",
"sensor_type" => "temperature"
}
})
assert {:error, _reason} = SnmpSensorExecutor.execute(check)
end
end
end

58
test/towerops/workers/firmware_version_fetcher_worker_test.exs
View file

@ -175,6 +175,64 @@ defmodule Towerops.Workers.FirmwareVersionFetcherWorkerTest do
assert data.release_date == ~D[2025-02-01]
end
test "parses every month abbreviation" do
months = [
{"Jan", 1},
{"Feb", 2},
{"Mar", 3},
{"Apr", 4},
{"May", 5},
{"Jun", 6},
{"Jul", 7},
{"Aug", 8},
{"Sep", 9},
{"Oct", 10},
{"Nov", 11},
{"Dec", 12}
]
for {abbrev, month_num} <- months do
rss = """
<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<title>RouterOS latest stable version</title>
<item>
<title>7.14.1</title>
<link>https://mikrotik.com/download</link>
<pubDate>Mon, 15 #{abbrev} 2025 12:00:00 +0000</pubDate>
<description>Test</description>
</item>
</channel>
</rss>
"""
assert {:ok, data} = FirmwareVersionFetcherWorker.parse_rss_feed(rss)
assert data.release_date.month == month_num
assert data.release_date.year == 2025
end
end
test "returns nil release_date for unrecognized month" do
rss = """
<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<title>RouterOS latest stable version</title>
<item>
<title>7.14.1</title>
<link>https://mikrotik.com/download</link>
<pubDate>Mon, 15 Foo 2025 12:00:00 +0000</pubDate>
<description>Test</description>
</item>
</channel>
</rss>
"""
assert {:ok, data} = FirmwareVersionFetcherWorker.parse_rss_feed(rss)
assert is_nil(data.release_date)
end
test "handles invalid RFC822 date format gracefully" do
invalid_date = """
<?xml version="1.0" encoding="UTF-8"?>

82
test/towerops_web/channels/agent_channel_cloud_poller_test.exs Normal file
View file

@ -0,0 +1,82 @@
defmodule ToweropsWeb.AgentChannelCloudPollerTest do
@moduledoc """
Drives the `is_cloud_poller` join branch + latency_probe_jobs subscribe
side-effect. The default AgentChannel test connects with a regular
(non-cloud) agent token; this file specifically tests the cloud-poller
path so the `Phoenix.PubSub.subscribe(... :latency_probe)` line + downstream
handler executes with cloud_poller=true.
"""
use Towerops.DataCase, async: false
import Phoenix.ChannelTest
alias Towerops.AccountsFixtures
alias Towerops.Agent.AgentJobList
alias Towerops.Agents
alias Towerops.OrganizationsFixtures
alias ToweropsWeb.AgentSocket
@endpoint ToweropsWeb.Endpoint
setup do
user = AccountsFixtures.user_fixture()
organization = OrganizationsFixtures.organization_fixture(user.id)
{:ok, agent_token, token_string} = Agents.create_cloud_poller("Test Cloud Poller")
{:ok, socket} = connect(AgentSocket, %{})
{:ok, _, socket} =
subscribe_and_join(socket, "agent:#{agent_token.id}", %{"token" => token_string})
assert_push("jobs", _initial_jobs, 1_000)
%{
socket: socket,
agent_token: agent_token,
token_string: token_string,
organization: organization,
user: user
}
end
describe "cloud poller join + latency_probe subscription" do
test "join with a cloud-poller token sets the cloud poller branch", %{
agent_token: agent_token,
organization: organization
} do
# Cloud-poller tokens are global (no organization_id), so we use the
# test's user-owned organization for the test device. The setup itself
# proves the join path subscribed to the latency_probe PubSub topic;
# we confirm by broadcasting on it and watching for a "jobs" push.
assert agent_token.is_cloud_poller
{:ok, site} =
Towerops.Sites.create_site(%{
name: "Cloud Site",
organization_id: organization.id
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Probe Device",
ip_address: "10.66.0.1",
site_id: site.id,
organization_id: organization.id
})
Phoenix.PubSub.broadcast(
Towerops.PubSub,
"agent:#{agent_token.id}:latency_probe",
{:latency_probe_jobs, [device]}
)
assert_push("jobs", %{binary: jobs_binary}, 2_000)
{:ok, decoded} = Base.decode64(jobs_binary)
{:ok, job_list} = AgentJobList.decode(decoded)
ping_jobs = Enum.filter(job_list.jobs, &(&1.job_type == :PING))
assert length(ping_jobs) == 5
end
end
end