Graham McIntire
ca7bb75472
- C1: Replace innerHTML template literals with DOM textContent in sites_map.ts - C2: Derive user_id from current_scope instead of client params in policy consent - C3: Fix broken halt() in GDPR data export (orphaned _ = ... halt()) - C4: Add owner/admin authorization check to MembersController update/delete - C5: Require HMAC signature on agent release webhook (was optional) - C6: Fix Repo.all_by/2 (not a real Ecto function) → Repo.all with query - C7: Move auth exemption to before_send callback in BruteForceProtection - C8: Block </style>/<script injection in status page custom_css validation - C9: Create secrets.example.yaml with placeholders; secrets.yaml already gitignored - C10: Load Stripe key from STRIPE_SECRET_KEY env var instead of hardcoded value - C13: Remove credentials from PubSub backup broadcast; channel resolves them C11 (no force_ssl): by design — Cloudflared terminates TLS C12 (device quota race): false positive — FOR UPDATE serializes correctly
2276 lines
76 KiB
Elixir
2276 lines
76 KiB
Elixir
defmodule ToweropsWeb.AgentChannelTest do
|
|
use Towerops.DataCase, async: false
|
|
|
|
import Phoenix.ChannelTest
|
|
|
|
alias Towerops.AccountsFixtures
|
|
alias Towerops.Agent.AgentError
|
|
alias Towerops.Agent.AgentHeartbeat
|
|
alias Towerops.Agent.AgentJobList
|
|
alias Towerops.Agent.CheckResult, as: CheckResultProto
|
|
alias Towerops.Agent.CredentialTestResult
|
|
alias Towerops.Agent.MikrotikResult
|
|
alias Towerops.Agent.MikrotikSentence
|
|
alias Towerops.Agent.MonitoringCheck
|
|
alias Towerops.Agent.SnmpResult
|
|
alias Towerops.Agents
|
|
alias Towerops.AgentsFixtures
|
|
alias Towerops.Devices.BackupRequests
|
|
alias Towerops.Devices.MikrotikBackups
|
|
alias Towerops.DevicesFixtures
|
|
alias Towerops.OrganizationsFixtures
|
|
alias Towerops.Proto.Encode
|
|
alias Towerops.Proto.Types.LldpNeighbor
|
|
alias Towerops.Proto.Types.LldpTopologyResult
|
|
alias Towerops.Snmp.AgentDiscovery
|
|
alias Towerops.Snmp.Sensor
|
|
alias ToweropsWeb.AgentSocket
|
|
|
|
@endpoint ToweropsWeb.Endpoint
|
|
|
|
setup do
|
|
# Create organization, site, device, and agent token
|
|
user = AccountsFixtures.user_fixture()
|
|
organization = OrganizationsFixtures.organization_fixture(user.id)
|
|
|
|
device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: organization.id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public"
|
|
})
|
|
|
|
{:ok, agent_token, token_string} =
|
|
AgentsFixtures.agent_token_fixture(organization.id)
|
|
|
|
# Assign device to agent
|
|
{:ok, _assignment} =
|
|
AgentsFixtures.agent_assignment_fixture(agent_token.id, device.id)
|
|
|
|
# Connect socket and join channel
|
|
{:ok, socket} = connect(AgentSocket, %{})
|
|
|
|
{:ok, _, socket} =
|
|
subscribe_and_join(socket, "agent:#{agent_token.id}", %{"token" => token_string})
|
|
|
|
# Drain the initial :send_jobs message pushed on join
|
|
assert_push "jobs", _initial_jobs
|
|
|
|
%{
|
|
socket: socket,
|
|
device: device,
|
|
agent_token: agent_token,
|
|
token_string: token_string,
|
|
organization: organization
|
|
}
|
|
end
|
|
|
|
# Helper to encode a protobuf struct to base64 payload map
|
|
defp encode_payload(protobuf_struct) do
|
|
binary = protobuf_struct.__struct__.encode(protobuf_struct)
|
|
%{"binary" => Base.encode64(binary)}
|
|
end
|
|
|
|
# Poll for a condition to become true, exiting early on success
|
|
defp poll_until(fun, opts \\ []) do
|
|
max_attempts = Keyword.get(opts, :max_attempts, 20)
|
|
delay_ms = Keyword.get(opts, :delay_ms, 5)
|
|
|
|
Enum.reduce_while(1..max_attempts, nil, fn _, _ ->
|
|
case fun.() do
|
|
nil ->
|
|
Process.sleep(delay_ms)
|
|
{:cont, nil}
|
|
|
|
false ->
|
|
Process.sleep(delay_ms)
|
|
{:cont, nil}
|
|
|
|
result ->
|
|
{:halt, result}
|
|
end
|
|
end) || fun.()
|
|
end
|
|
|
|
defp build_heartbeat(attrs \\ %{}) do
|
|
Map.merge(
|
|
%AgentHeartbeat{
|
|
version: "1.0.0",
|
|
hostname: "test-agent",
|
|
uptime_seconds: 3600,
|
|
ip_address: "",
|
|
arch: "x86_64"
|
|
},
|
|
attrs
|
|
)
|
|
end
|
|
|
|
defp build_agent_error(device_id) do
|
|
%AgentError{
|
|
device_id: device_id,
|
|
job_id: "poll:#{device_id}",
|
|
message: "SNMP timeout",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
end
|
|
|
|
defp build_monitoring_check(device_id, status) do
|
|
%MonitoringCheck{
|
|
device_id: device_id,
|
|
status: status,
|
|
response_time_ms: 10.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
end
|
|
|
|
defp build_credential_test_result(test_id, success) do
|
|
%CredentialTestResult{
|
|
test_id: test_id,
|
|
success: success,
|
|
error_message: if(success, do: "", else: "Connection refused"),
|
|
system_description: if(success, do: "Test Device v1.0", else: ""),
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
end
|
|
|
|
defp build_mikrotik_result(device_id, opts \\ []) do
|
|
%MikrotikResult{
|
|
device_id: device_id,
|
|
job_id: Keyword.get(opts, :job_id, "mikrotik:#{device_id}"),
|
|
sentences: Keyword.get(opts, :sentences, []),
|
|
error: Keyword.get(opts, :error, ""),
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
end
|
|
|
|
# ── Stage 1: join/3 and terminate/2 ──────────────────────────────────
|
|
|
|
describe "join/3" do
|
|
test "join with invalid token returns error" do
|
|
{:ok, socket} = connect(AgentSocket, %{})
|
|
|
|
assert {:error, %{reason: "unauthorized"}} =
|
|
subscribe_and_join(socket, "agent:some-id", %{"token" => "invalid-token"})
|
|
end
|
|
|
|
test "join with missing token returns error" do
|
|
{:ok, socket} = connect(AgentSocket, %{})
|
|
|
|
assert {:error, %{reason: "missing token"}} =
|
|
subscribe_and_join(socket, "agent:some-id", %{})
|
|
end
|
|
|
|
test "join sets socket assigns", %{socket: socket, agent_token: agent_token, organization: organization} do
|
|
assert socket.assigns.agent_token_id == agent_token.id
|
|
assert socket.assigns.organization_id == organization.id
|
|
assert Map.has_key?(socket.assigns, :debug_enabled)
|
|
assert Map.has_key?(socket.assigns, :last_heartbeat_at)
|
|
end
|
|
|
|
test "join broadcasts agent_connected", %{organization: organization} do
|
|
# Subscribe to health topic before joining
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health")
|
|
|
|
{:ok, new_agent_token, new_token_string} =
|
|
AgentsFixtures.agent_token_fixture(organization.id)
|
|
|
|
new_agent_token_id = new_agent_token.id
|
|
|
|
{:ok, new_socket} = connect(AgentSocket, %{})
|
|
|
|
{:ok, _, _socket} =
|
|
subscribe_and_join(new_socket, "agent:#{new_agent_token.id}", %{"token" => new_token_string})
|
|
|
|
assert_receive {:agent_connected, ^new_agent_token_id, _org_id}
|
|
end
|
|
|
|
test "terminate broadcasts agent_disconnected", %{socket: socket, agent_token: agent_token} do
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health")
|
|
|
|
agent_token_id = agent_token.id
|
|
|
|
Process.flag(:trap_exit, true)
|
|
leave(socket)
|
|
|
|
assert_receive {:agent_disconnected, ^agent_token_id, _org_id}
|
|
end
|
|
end
|
|
|
|
# ── Stage 2: handle_in "heartbeat" ──────────────────────────────────
|
|
|
|
describe "handle_in heartbeat" do
|
|
test "valid heartbeat updates agent token in DB", %{socket: socket, agent_token: agent_token} do
|
|
# DB is updated on join, so last_seen_at should already be set
|
|
initial_token = Agents.get_agent_token!(agent_token.id)
|
|
assert initial_token.last_seen_at
|
|
|
|
# Send heartbeat with new version - this should update metadata after 30s throttle
|
|
# For this test, we'll just verify the heartbeat is processed without crashing
|
|
heartbeat = build_heartbeat(%{version: "2.0.0"})
|
|
payload = encode_payload(heartbeat)
|
|
|
|
push(socket, "heartbeat", payload)
|
|
|
|
# Verify channel is still alive after processing heartbeat
|
|
ref = push(socket, "heartbeat", payload)
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "heartbeat is processed without errors", %{socket: socket} do
|
|
# First heartbeat after join is throttled (DB was just updated on join)
|
|
# so no DB update or broadcast will happen, but heartbeat should be processed
|
|
heartbeat = build_heartbeat()
|
|
payload = encode_payload(heartbeat)
|
|
|
|
push(socket, "heartbeat", payload)
|
|
|
|
# Verify channel is still alive after processing heartbeat
|
|
ref = push(socket, "heartbeat", payload)
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid base64 heartbeat is handled gracefully", %{socket: socket} do
|
|
push(socket, "heartbeat", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
# Channel should not crash - verify by sending another message
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid protobuf heartbeat is handled gracefully", %{socket: socket} do
|
|
push(socket, "heartbeat", %{"binary" => Base.encode64("not-a-protobuf")})
|
|
|
|
# Channel should not crash
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 3: handle_in "error" and "credential_test_result" ─────────
|
|
|
|
describe "handle_in error" do
|
|
test "valid error message is handled without crash", %{socket: socket, device: device} do
|
|
error = build_agent_error(device.id)
|
|
payload = encode_payload(error)
|
|
|
|
push(socket, "error", payload)
|
|
|
|
# Channel should not crash
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
|
|
test "invalid base64 error payload is handled gracefully", %{socket: socket} do
|
|
push(socket, "error", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
describe "handle_in credential_test_result" do
|
|
test "valid credential test result is broadcast", %{socket: socket} do
|
|
test_id = Ecto.UUID.generate()
|
|
|
|
# Subscribe to the credential test topic
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "credential_test:#{test_id}")
|
|
|
|
result = build_credential_test_result(test_id, true)
|
|
payload = encode_payload(result)
|
|
|
|
push(socket, "credential_test_result", payload)
|
|
|
|
assert_receive {:credential_test_result, received_result}
|
|
assert received_result.test_id == test_id
|
|
assert received_result.success == true
|
|
end
|
|
|
|
test "failed credential test result is broadcast with error", %{socket: socket} do
|
|
test_id = Ecto.UUID.generate()
|
|
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "credential_test:#{test_id}")
|
|
|
|
result = build_credential_test_result(test_id, false)
|
|
payload = encode_payload(result)
|
|
|
|
push(socket, "credential_test_result", payload)
|
|
|
|
assert_receive {:credential_test_result, received_result}
|
|
assert received_result.success == false
|
|
assert received_result.error_message == "Connection refused"
|
|
end
|
|
|
|
test "invalid base64 credential test result is handled gracefully", %{socket: socket} do
|
|
push(socket, "credential_test_result", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 4: handle_in "monitoring_check" ───────────────────────────
|
|
|
|
describe "handle_in monitoring_check" do
|
|
test "valid monitoring check creates check record", %{socket: socket, device: device} do
|
|
import Ecto.Query
|
|
|
|
check = build_monitoring_check(device.id, "success")
|
|
payload = encode_payload(check)
|
|
|
|
push(socket, "monitoring_check", payload)
|
|
|
|
# Poll until check is stored
|
|
checks =
|
|
poll_until(fn ->
|
|
result = Towerops.Repo.all(from(m in Towerops.Monitoring.MonitoringCheck, where: m.device_id == ^device.id))
|
|
if result != [], do: result
|
|
end)
|
|
|
|
assert checks != []
|
|
assert hd(checks).status == "success"
|
|
end
|
|
|
|
test "monitoring check updates device status to up on success", %{socket: socket, device: device} do
|
|
check = build_monitoring_check(device.id, "success")
|
|
payload = encode_payload(check)
|
|
|
|
push(socket, "monitoring_check", payload)
|
|
|
|
# Poll until device status updates
|
|
updated_device =
|
|
poll_until(fn ->
|
|
device = Towerops.Devices.get_device!(device.id)
|
|
if device.status == :up, do: device
|
|
end)
|
|
|
|
assert updated_device.status == :up
|
|
end
|
|
|
|
test "monitoring check creates alert on status change to down", %{socket: socket, device: device} do
|
|
# First set device as up
|
|
Towerops.Devices.update_device_status(device, :up)
|
|
|
|
check = build_monitoring_check(device.id, "failure")
|
|
payload = encode_payload(check)
|
|
|
|
push(socket, "monitoring_check", payload)
|
|
|
|
# Poll until device status updates
|
|
updated_device =
|
|
poll_until(fn ->
|
|
device = Towerops.Devices.get_device!(device.id)
|
|
if device.status == :down, do: device
|
|
end)
|
|
|
|
assert updated_device.status == :down
|
|
|
|
# Poll until alert is created
|
|
poll_until(fn ->
|
|
if Towerops.Alerts.has_active_alert?(device.id, :device_down), do: true
|
|
end)
|
|
|
|
assert Towerops.Alerts.has_active_alert?(device.id, :device_down)
|
|
end
|
|
|
|
test "monitoring check creates device_up alert on recovery", %{socket: socket, device: device} do
|
|
# Set device as down first
|
|
Towerops.Devices.update_device_status(device, :down)
|
|
|
|
check = build_monitoring_check(device.id, "success")
|
|
payload = encode_payload(check)
|
|
|
|
push(socket, "monitoring_check", payload)
|
|
|
|
# Poll until device status updates
|
|
updated_device =
|
|
poll_until(fn ->
|
|
device = Towerops.Devices.get_device!(device.id)
|
|
if device.status == :up, do: device
|
|
end)
|
|
|
|
assert updated_device.status == :up
|
|
end
|
|
|
|
test "invalid base64 monitoring check is handled gracefully", %{socket: socket} do
|
|
push(socket, "monitoring_check", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid protobuf monitoring check is handled gracefully", %{socket: socket} do
|
|
push(socket, "monitoring_check", %{"binary" => Base.encode64("not-a-protobuf")})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 5: handle_in "mikrotik_result" and catch-all ─────────────
|
|
|
|
describe "handle_in mikrotik_result" do
|
|
test "valid mikrotik result is processed without crash", %{socket: socket, device: device} do
|
|
result = build_mikrotik_result(device.id)
|
|
payload = encode_payload(result)
|
|
|
|
push(socket, "mikrotik_result", payload)
|
|
|
|
# Channel should not crash
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "mikrotik result with sentences is processed", %{socket: socket, device: device} do
|
|
sentence = %MikrotikSentence{
|
|
attributes: %{"name" => "router1"}
|
|
}
|
|
|
|
result = build_mikrotik_result(device.id, sentences: [sentence])
|
|
payload = encode_payload(result)
|
|
|
|
push(socket, "mikrotik_result", payload)
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid base64 mikrotik result is handled gracefully", %{socket: socket} do
|
|
push(socket, "mikrotik_result", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid protobuf mikrotik result is handled gracefully", %{socket: socket} do
|
|
push(socket, "mikrotik_result", %{"binary" => Base.encode64("not-a-protobuf")})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
describe "handle_in catch-all" do
|
|
test "unknown event is handled gracefully", %{socket: socket} do
|
|
push(socket, "unknown_event", %{"data" => "test"})
|
|
|
|
# Channel should not crash
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
describe "handle_in oversized message rejection" do
|
|
# Constructs a payload bigger than the 10 MiB @max_message_size threshold.
|
|
defp oversized_payload do
|
|
# 11 MiB of base64-safe ASCII — well above the 10 * 1024 * 1024 limit
|
|
%{"binary" => String.duplicate("A", 11 * 1024 * 1024)}
|
|
end
|
|
|
|
test "result rejects messages over @max_message_size", %{socket: socket} do
|
|
ref = push(socket, "result", oversized_payload())
|
|
assert_reply ref, :error, %{reason: "Message too large" <> _}
|
|
end
|
|
|
|
test "heartbeat rejects messages over @max_message_size", %{socket: socket} do
|
|
ref = push(socket, "heartbeat", oversized_payload())
|
|
assert_reply ref, :error, %{reason: "Message too large"}
|
|
end
|
|
|
|
test "error rejects messages over @max_message_size", %{socket: socket} do
|
|
ref = push(socket, "error", oversized_payload())
|
|
assert_reply ref, :error, %{reason: "Message too large"}
|
|
end
|
|
end
|
|
|
|
describe "handle_in invalid mikrotik_result" do
|
|
test "invalid base64 is logged and channel survives", %{socket: socket} do
|
|
push(socket, "mikrotik_result", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 6: handle_info lifecycle messages ─────────────────────────
|
|
|
|
describe "handle_info :send_jobs" do
|
|
test "pushes jobs to channel", %{socket: socket} do
|
|
send(socket.channel_pid, :send_jobs)
|
|
|
|
assert_push "jobs", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
assert is_list(job_list.jobs)
|
|
end
|
|
|
|
test "sends check jobs to agent when checks exist", %{
|
|
socket: socket,
|
|
organization: organization,
|
|
agent_token: agent_token
|
|
} do
|
|
# Create HTTP check
|
|
{:ok, _http_check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: organization.id,
|
|
agent_token_id: agent_token.id,
|
|
check_type: "http",
|
|
name: "Test HTTP Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
enabled: true,
|
|
config: %{
|
|
"url" => "https://example.com",
|
|
"method" => "GET",
|
|
"expected_status" => 200,
|
|
"verify_ssl" => true
|
|
}
|
|
})
|
|
|
|
# Create TCP check
|
|
{:ok, _tcp_check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: organization.id,
|
|
agent_token_id: agent_token.id,
|
|
check_type: "tcp",
|
|
name: "Test TCP Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
enabled: true,
|
|
config: %{
|
|
"host" => "example.com",
|
|
"port" => 80
|
|
}
|
|
})
|
|
|
|
# Create DNS check
|
|
{:ok, _dns_check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: organization.id,
|
|
agent_token_id: agent_token.id,
|
|
check_type: "dns",
|
|
name: "Test DNS Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
enabled: true,
|
|
config: %{
|
|
"hostname" => "example.com",
|
|
"record_type" => "A"
|
|
}
|
|
})
|
|
|
|
# Create SSL check
|
|
{:ok, _ssl_check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: organization.id,
|
|
agent_token_id: agent_token.id,
|
|
check_type: "ssl",
|
|
name: "Test SSL Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
enabled: true,
|
|
config: %{
|
|
"host" => "example.com",
|
|
"port" => 443,
|
|
"warning_days" => 30
|
|
}
|
|
})
|
|
|
|
# Trigger check jobs - this will crash if check_type_config returns :config field
|
|
send(socket.channel_pid, :send_jobs)
|
|
|
|
# Should receive check_jobs message (proves no KeyError crash)
|
|
assert_push "check_jobs", %{binary: check_jobs_binary}
|
|
|
|
# Verify it's valid base64
|
|
assert {:ok, _decoded} = Base.decode64(check_jobs_binary)
|
|
|
|
# Test that the channel is still alive (no crash from KeyError)
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
describe "handle_info :check_heartbeat" do
|
|
test "reschedules when heartbeat is recent", %{socket: socket} do
|
|
# last_heartbeat_at was set during join (recent), so check should pass
|
|
send(socket.channel_pid, :check_heartbeat)
|
|
|
|
# Channel should still be alive
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
|
|
test "stops channel when heartbeat is stale", %{socket: socket} do
|
|
Process.flag(:trap_exit, true)
|
|
|
|
# Monitor the channel process before it stops
|
|
ref = Process.monitor(socket.channel_pid)
|
|
|
|
# Set last_heartbeat_at to >300 seconds ago to trigger timeout
|
|
stale_time = DateTime.add(DateTime.utc_now(), -400, :second)
|
|
|
|
# Phoenix Channel.Server GenServer state is the socket struct directly
|
|
:sys.replace_state(socket.channel_pid, fn socket_state ->
|
|
Phoenix.Socket.assign(socket_state, :last_heartbeat_at, stale_time)
|
|
end)
|
|
|
|
send(socket.channel_pid, :check_heartbeat)
|
|
|
|
assert_receive {:DOWN, ^ref, :process, _, _}, 1000
|
|
end
|
|
end
|
|
|
|
describe "handle_info :token_disabled" do
|
|
test "stops channel", %{socket: socket} do
|
|
Process.flag(:trap_exit, true)
|
|
ref = Process.monitor(socket.channel_pid)
|
|
|
|
send(socket.channel_pid, :token_disabled)
|
|
|
|
assert_receive {:DOWN, ^ref, :process, _, _}, 1000
|
|
end
|
|
end
|
|
|
|
describe "handle_info :restart_requested" do
|
|
test "pushes restart and stops channel", %{socket: socket} do
|
|
Process.flag(:trap_exit, true)
|
|
ref = Process.monitor(socket.channel_pid)
|
|
|
|
send(socket.channel_pid, :restart_requested)
|
|
|
|
assert_push "restart", %{}
|
|
assert_receive {:DOWN, ^ref, :process, _, _}, 1000
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:update_requested, url, checksum}" do
|
|
test "pushes update with url and checksum", %{socket: socket} do
|
|
send(socket.channel_pid, {:update_requested, "https://example.com/agent.tar.gz", "sha256:abc123"})
|
|
|
|
assert_push "update", %{url: "https://example.com/agent.tar.gz", checksum: "sha256:abc123"}
|
|
end
|
|
end
|
|
|
|
# ── Stage 7: handle_info PubSub messages ────────────────────────────
|
|
|
|
describe "handle_info {:assignments_changed, _}" do
|
|
test "triggers debounced job refresh", %{socket: socket} do
|
|
send(socket.channel_pid, {:assignments_changed, :assigned})
|
|
|
|
# Jobs push should come after debounce delay (50ms in test)
|
|
assert_push "jobs", %{binary: _jobs_binary}, 200
|
|
end
|
|
|
|
test "debounces rapid assignment changes", %{socket: socket} do
|
|
# Send multiple rapid changes
|
|
send(socket.channel_pid, {:assignments_changed, :assigned})
|
|
send(socket.channel_pid, {:assignments_changed, :unassigned})
|
|
send(socket.channel_pid, {:assignments_changed, :assigned})
|
|
|
|
# Should only get one push (debounced) - 50ms debounce in test
|
|
assert_push "jobs", %{binary: _}, 200
|
|
refute_push "jobs", %{}, 100
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:discovery_requested, device_id}" do
|
|
test "pushes discovery job for valid device", %{socket: socket, device: device} do
|
|
send(socket.channel_pid, {:discovery_requested, device.id})
|
|
|
|
assert_push "discovery_job", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
discover_job = Enum.find(job_list.jobs, &(&1.job_type == :DISCOVER))
|
|
assert discover_job
|
|
assert discover_job.device_id == device.id
|
|
end
|
|
|
|
test "handles nonexistent device without crash", %{socket: socket} do
|
|
send(socket.channel_pid, {:discovery_requested, Ecto.UUID.generate()})
|
|
|
|
# Channel should not crash
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:credential_test_requested, test_id, config}" do
|
|
test "pushes credential test job", %{socket: socket} do
|
|
test_id = Ecto.UUID.generate()
|
|
|
|
snmp_config = [
|
|
ip: "192.168.1.100",
|
|
port: 161,
|
|
version: "2c",
|
|
community: "public",
|
|
v3_security_level: "",
|
|
v3_username: "",
|
|
v3_auth_protocol: "",
|
|
v3_auth_password: "",
|
|
v3_priv_protocol: "",
|
|
v3_priv_password: ""
|
|
]
|
|
|
|
send(socket.channel_pid, {:credential_test_requested, test_id, snmp_config})
|
|
|
|
assert_push "jobs", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
test_job = Enum.find(job_list.jobs, &(&1.job_type == :TEST_CREDENTIALS))
|
|
assert test_job
|
|
assert test_job.job_id == test_id
|
|
assert test_job.snmp_device.ip == "192.168.1.100"
|
|
assert test_job.snmp_device.community == "public"
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:live_poll_requested, ...}" do
|
|
test "pushes live poll job for valid device", %{socket: socket, device: device} do
|
|
sensor_oids = ["1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.3.0"]
|
|
reply_topic = "live_poll:#{device.id}:sensor_data"
|
|
|
|
send(socket.channel_pid, {:live_poll_requested, device.id, sensor_oids, reply_topic})
|
|
|
|
assert_push "jobs", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
poll_job = Enum.find(job_list.jobs, &(&1.job_type == :POLL))
|
|
assert poll_job
|
|
assert poll_job.device_id == device.id
|
|
assert poll_job.job_id == "live_poll:#{device.id}:#{reply_topic}"
|
|
|
|
# Verify requested OIDs are in the query
|
|
query_oids = Enum.flat_map(poll_job.queries, & &1.oids)
|
|
assert "1.3.6.1.2.1.1.1.0" in query_oids
|
|
assert "1.3.6.1.2.1.1.3.0" in query_oids
|
|
end
|
|
|
|
test "handles nonexistent device without crash", %{socket: socket} do
|
|
send(
|
|
socket.channel_pid,
|
|
{:live_poll_requested, Ecto.UUID.generate(), ["1.3.6.1.2.1.1.1.0"], "reply:topic"}
|
|
)
|
|
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:live_poll_timeout, reply_topic}" do
|
|
test "broadcasts error via channel process", %{socket: socket} do
|
|
reply_topic = "live_poll:test:timeout_test"
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, reply_topic)
|
|
|
|
send(socket.channel_pid, {:live_poll_timeout, reply_topic})
|
|
|
|
assert_receive {:live_poll_error, :timeout}, 1_000
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:backup_requested, device_id, job_id}" do
|
|
test "pushes backup job to agent", %{socket: socket, device: device} do
|
|
job_id = "backup:#{device.id}"
|
|
|
|
send(socket.channel_pid, {:backup_requested, device.id, job_id})
|
|
|
|
assert_push "backup_job", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
backup_job = hd(job_list.jobs)
|
|
assert backup_job.job_id == job_id
|
|
assert backup_job.device_id == device.id
|
|
end
|
|
end
|
|
|
|
describe "handle_info {:poll_after_discovery, device_id}" do
|
|
test "pushes poll jobs for discovered device", %{socket: socket, device: device} do
|
|
# First run discovery so device has an snmp_device record
|
|
# (build_polling_queries requires device.snmp_device.sensors and .interfaces)
|
|
discovery_result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :DISCOVER,
|
|
job_id: "discover:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
"1.3.6.1.2.1.1.1.0" => "Test Device",
|
|
"1.3.6.1.2.1.1.2.0" => "1.3.6.1.4.1.9.1.1",
|
|
"1.3.6.1.2.1.1.3.0" => "123456",
|
|
"1.3.6.1.2.1.1.4.0" => "admin@test.com",
|
|
"1.3.6.1.2.1.1.5.0" => "test-device",
|
|
"1.3.6.1.2.1.1.6.0" => "Test Location"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(discovery_result))
|
|
|
|
# Wait for discovery to complete and the automatic poll_after_discovery
|
|
assert_push "jobs", %{binary: jobs_binary}, 1000
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
# Should have at least a poll job
|
|
assert job_list.jobs != []
|
|
end
|
|
|
|
test "handles nonexistent device without crash", %{socket: socket} do
|
|
Process.flag(:trap_exit, true)
|
|
send(socket.channel_pid, {:poll_after_discovery, Ecto.UUID.generate()})
|
|
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
end
|
|
|
|
# ── Stage 8: SNMP result edge cases ─────────────────────────────────
|
|
|
|
describe "handle_in result edge cases" do
|
|
test "invalid base64 result is handled gracefully", %{socket: socket} do
|
|
push(socket, "result", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid protobuf result is handled gracefully", %{socket: socket} do
|
|
push(socket, "result", %{"binary" => Base.encode64("not-a-protobuf")})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "result for nonexistent device is handled", %{socket: socket} do
|
|
result = %SnmpResult{
|
|
device_id: Ecto.UUID.generate(),
|
|
job_type: :POLL,
|
|
job_id: "poll:#{Ecto.UUID.generate()}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{"1.3.6.1.2.1.1.1.0" => "Test"}
|
|
}
|
|
|
|
payload = encode_payload(result)
|
|
push(socket, "result", payload)
|
|
|
|
# Channel should not crash
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
|
|
test "result for device in wrong organization is rejected", %{socket: socket} do
|
|
# Create a device in a different organization
|
|
other_user = AccountsFixtures.user_fixture()
|
|
other_org = OrganizationsFixtures.organization_fixture(other_user.id)
|
|
|
|
other_device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: other_org.id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public"
|
|
})
|
|
|
|
result = %SnmpResult{
|
|
device_id: other_device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{other_device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{"1.3.6.1.2.1.1.1.0" => "Test"}
|
|
}
|
|
|
|
payload = encode_payload(result)
|
|
push(socket, "result", payload)
|
|
|
|
# Channel should not crash
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
|
|
test "live poll result broadcasts to reply topic", %{socket: socket, device: device} do
|
|
reply_topic = "live_poll:#{device.id}:sensor_data"
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, reply_topic)
|
|
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "live_poll:#{device.id}:#{reply_topic}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
"1.3.6.1.2.1.1.1.0" => "Test Device",
|
|
"1.3.6.1.2.1.1.3.0" => "123456"
|
|
}
|
|
}
|
|
|
|
payload = encode_payload(result)
|
|
push(socket, "result", payload)
|
|
|
|
assert_receive {:live_poll_result, oid_values}, 500
|
|
assert Map.has_key?(oid_values, "1.3.6.1.2.1.1.1.0")
|
|
end
|
|
end
|
|
|
|
# ── Stage 9: Polling result processing with sensors/interfaces ─────
|
|
|
|
describe "poll result with sensor and interface data" do
|
|
setup %{device: device} do
|
|
# First run discovery to create snmp_device record
|
|
oid_values = %{
|
|
"1.3.6.1.2.1.1.1.0" => "Test Device",
|
|
"1.3.6.1.2.1.1.2.0" => "1.3.6.1.4.1.9.1.1",
|
|
"1.3.6.1.2.1.1.3.0" => "123456",
|
|
"1.3.6.1.2.1.1.4.0" => "admin@test.com",
|
|
"1.3.6.1.2.1.1.5.0" => "test-device",
|
|
"1.3.6.1.2.1.1.6.0" => "Test Location",
|
|
# Interface data
|
|
"1.3.6.1.2.1.2.2.1.1.1" => "1",
|
|
"1.3.6.1.2.1.2.2.1.2.1" => "GigabitEthernet0/1",
|
|
"1.3.6.1.2.1.2.2.1.3.1" => "6",
|
|
"1.3.6.1.2.1.2.2.1.5.1" => "1000000000",
|
|
"1.3.6.1.2.1.2.2.1.6.1" => "aa:bb:cc:dd:ee:ff",
|
|
"1.3.6.1.2.1.2.2.1.7.1" => "1",
|
|
"1.3.6.1.2.1.2.2.1.8.1" => "1"
|
|
}
|
|
|
|
{:ok, _discovered} = AgentDiscovery.process_agent_discovery(device, oid_values)
|
|
|
|
# Create a sensor on the snmp_device
|
|
snmp_device = Towerops.Snmp.get_device_with_associations(device.id)
|
|
|
|
{:ok, sensor} =
|
|
Towerops.Repo.insert(%Sensor{
|
|
snmp_device_id: snmp_device.id,
|
|
sensor_type: "temperature",
|
|
sensor_index: "1001",
|
|
sensor_oid: "1.3.6.1.4.1.9.9.13.1.3.1.3.1001",
|
|
sensor_descr: "CPU Temperature",
|
|
sensor_unit: "celsius",
|
|
sensor_divisor: 1
|
|
})
|
|
|
|
# Reload device with all associations
|
|
device_with_details = Towerops.Devices.get_device_with_details(device.id)
|
|
|
|
%{
|
|
device: device_with_details,
|
|
discovered_device: device_with_details,
|
|
snmp_device: snmp_device,
|
|
sensor: sensor,
|
|
interface: hd(snmp_device.interfaces)
|
|
}
|
|
end
|
|
|
|
test "poll result stores sensor readings", %{
|
|
socket: socket,
|
|
device: device,
|
|
sensor: sensor
|
|
} do
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
sensor.sensor_oid => "45"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Poll until sensor reading is stored
|
|
updated_sensor =
|
|
poll_until(fn ->
|
|
s = Towerops.Repo.get!(Sensor, sensor.id)
|
|
if s.last_value == 45.0, do: s
|
|
end)
|
|
|
|
assert updated_sensor.last_value == 45.0
|
|
end
|
|
|
|
test "poll result rescales implausibly hot temperature readings", %{
|
|
socket: socket,
|
|
device: device,
|
|
sensor: sensor
|
|
} do
|
|
# Some MikroTik models (e.g. RB-series with mtxrHl* OIDs) report
|
|
# temperature in deci-degrees so a "29.4°C" reading lands on the
|
|
# wire as 294. SensorScale.normalize/2 must run on the agent path
|
|
# so it doesn't store 294°C.
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
sensor.sensor_oid => "294"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
updated_sensor =
|
|
poll_until(fn ->
|
|
s = Towerops.Repo.get!(Sensor, sensor.id)
|
|
if s.last_value && s.last_value < 100.0, do: s
|
|
end)
|
|
|
|
assert_in_delta updated_sensor.last_value, 29.4, 0.01
|
|
end
|
|
|
|
test "poll result stores interface stats", %{
|
|
socket: socket,
|
|
device: device,
|
|
interface: interface
|
|
} do
|
|
idx = interface.if_index
|
|
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
"1.3.6.1.2.1.31.1.1.1.6.#{idx}" => "1000000",
|
|
"1.3.6.1.2.1.31.1.1.1.10.#{idx}" => "2000000",
|
|
"1.3.6.1.2.1.2.2.1.14.#{idx}" => "5",
|
|
"1.3.6.1.2.1.2.2.1.20.#{idx}" => "3",
|
|
"1.3.6.1.2.1.2.2.1.13.#{idx}" => "1",
|
|
"1.3.6.1.2.1.2.2.1.19.#{idx}" => "0"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Poll until interface stats are stored
|
|
stats =
|
|
poll_until(fn ->
|
|
result = Towerops.Snmp.get_interface_stats(interface.id)
|
|
if result != [], do: result
|
|
end)
|
|
|
|
assert stats != []
|
|
stat = hd(stats)
|
|
assert stat.if_in_octets == 1_000_000
|
|
assert stat.if_out_octets == 2_000_000
|
|
end
|
|
|
|
test "poll result with leading dot OIDs is normalized", %{
|
|
socket: socket,
|
|
device: device,
|
|
sensor: sensor
|
|
} do
|
|
# Agent sends OIDs with leading dots, channel should normalize them
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
".#{sensor.sensor_oid}" => "72"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Poll until sensor reading is stored
|
|
updated_sensor =
|
|
poll_until(fn ->
|
|
s = Towerops.Repo.get!(Sensor, sensor.id)
|
|
if s.last_value == 72.0, do: s
|
|
end)
|
|
|
|
assert updated_sensor.last_value == 72.0
|
|
end
|
|
|
|
test "poll result with sensor divisor scales value", %{
|
|
socket: socket,
|
|
device: device,
|
|
snmp_device: snmp_device
|
|
} do
|
|
# Create a sensor with divisor of 10
|
|
{:ok, sensor_with_divisor} =
|
|
Towerops.Repo.insert(%Sensor{
|
|
snmp_device_id: snmp_device.id,
|
|
sensor_type: "voltage",
|
|
sensor_index: "2001",
|
|
sensor_oid: "1.3.6.1.4.1.9.9.13.1.2.1.3.2001",
|
|
sensor_descr: "Voltage Rail",
|
|
sensor_unit: "volts",
|
|
sensor_divisor: 10
|
|
})
|
|
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
sensor_with_divisor.sensor_oid => "120"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Poll until sensor reading is stored with divisor applied
|
|
updated_sensor =
|
|
poll_until(fn ->
|
|
s = Towerops.Repo.get!(Sensor, sensor_with_divisor.id)
|
|
if s.last_value == 12.0, do: s
|
|
end)
|
|
|
|
assert updated_sensor.last_value == 12.0
|
|
end
|
|
|
|
test "poll result auto-resolves device_down alert when device comes back up", %{
|
|
socket: socket,
|
|
discovered_device: device,
|
|
sensor: sensor
|
|
} do
|
|
# Set device status to down
|
|
{:ok, device} = Towerops.Devices.update_device_status(device, :down)
|
|
|
|
# Create a device_down alert
|
|
{:ok, alert} =
|
|
Towerops.Alerts.create_alert(%{
|
|
device_id: device.id,
|
|
alert_type: "device_down",
|
|
triggered_at: DateTime.utc_now(),
|
|
message: "Device is not responding to SNMP"
|
|
})
|
|
|
|
assert is_nil(alert.resolved_at)
|
|
assert device.status == :down
|
|
|
|
# Send successful SNMP polling result
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
sensor.sensor_oid => "45"
|
|
}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Poll until device status updates to up
|
|
updated_device =
|
|
poll_until(fn ->
|
|
device = Towerops.Devices.get_device(device.id)
|
|
if device && device.status == :up, do: device
|
|
end)
|
|
|
|
assert updated_device.status == :up
|
|
|
|
# Verify device_down alert was auto-resolved (poll for it since it happens asynchronously)
|
|
updated_alert =
|
|
poll_until(fn ->
|
|
alert = Towerops.Alerts.get_alert(alert.id)
|
|
if alert && alert.resolved_at, do: alert
|
|
end)
|
|
|
|
assert updated_alert.resolved_at
|
|
|
|
# Verify device_up alert was created
|
|
up_alerts =
|
|
device.id
|
|
|> Towerops.Alerts.list_devices_alerts()
|
|
|> Enum.filter(&(&1.alert_type == "device_up"))
|
|
|
|
assert length(up_alerts) == 1
|
|
up_alert = hd(up_alerts)
|
|
assert up_alert.message == "Device is now responding to SNMP"
|
|
assert up_alert.resolved_at
|
|
end
|
|
end
|
|
|
|
# ── Stage 10: SNMPv3 credential path ────────────────────────────────
|
|
|
|
describe "SNMPv3 device job building" do
|
|
test "sends discovery job with v3 credentials", %{organization: organization} do
|
|
# Create a v3 device with required SNMPv3 credentials
|
|
v3_device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: organization.id,
|
|
snmp_version: "3",
|
|
snmp_community: "",
|
|
snmpv3_username: "testuser",
|
|
snmpv3_auth_protocol: "SHA-256",
|
|
snmpv3_auth_password: "authpass123",
|
|
snmpv3_priv_protocol: "AES",
|
|
snmpv3_priv_password: "privpass123",
|
|
snmpv3_security_level: "authPriv",
|
|
snmpv3_credential_source: "device"
|
|
})
|
|
|
|
{:ok, agent_token2, token_string2} =
|
|
AgentsFixtures.agent_token_fixture(organization.id)
|
|
|
|
{:ok, _} = AgentsFixtures.agent_assignment_fixture(agent_token2.id, v3_device.id)
|
|
|
|
{:ok, socket2} = connect(AgentSocket, %{})
|
|
|
|
{:ok, _, socket2} =
|
|
subscribe_and_join(socket2, "agent:#{agent_token2.id}", %{"token" => token_string2})
|
|
|
|
# Drain initial jobs
|
|
assert_push "jobs", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
discover_job = Enum.find(job_list.jobs, &(&1.job_type == :DISCOVER))
|
|
assert discover_job
|
|
assert discover_job.snmp_device.version == "3"
|
|
|
|
Process.flag(:trap_exit, true)
|
|
leave(socket2)
|
|
end
|
|
end
|
|
|
|
# ── Stage 11: Poll after discovery with monitoring enabled ──────────
|
|
|
|
describe "poll after discovery with monitoring" do
|
|
test "includes ping job when monitoring is enabled", %{organization: organization} do
|
|
# Create device with monitoring enabled
|
|
monitored_device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: organization.id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public",
|
|
monitoring_enabled: true
|
|
})
|
|
|
|
{:ok, agent_token3, token_string3} =
|
|
AgentsFixtures.agent_token_fixture(organization.id)
|
|
|
|
{:ok, _} = AgentsFixtures.agent_assignment_fixture(agent_token3.id, monitored_device.id)
|
|
|
|
{:ok, socket3} = connect(AgentSocket, %{})
|
|
|
|
{:ok, _, socket3} =
|
|
subscribe_and_join(socket3, "agent:#{agent_token3.id}", %{"token" => token_string3})
|
|
|
|
# Drain initial jobs push
|
|
assert_push "jobs", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
# Should have both a discover/poll job and a ping job
|
|
ping_job = Enum.find(job_list.jobs, &(&1.job_type == :PING))
|
|
assert ping_job
|
|
assert ping_job.device_id == monitored_device.id
|
|
|
|
Process.flag(:trap_exit, true)
|
|
leave(socket3)
|
|
end
|
|
end
|
|
|
|
# ── Stage 12: Device reassignment rejection ─────────────────────────
|
|
|
|
describe "device reassignment" do
|
|
test "rejects result for device assigned to a different agent", %{
|
|
socket: socket,
|
|
organization: organization
|
|
} do
|
|
# Create a device assigned to a different agent
|
|
other_device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: organization.id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public"
|
|
})
|
|
|
|
{:ok, other_agent, _} = AgentsFixtures.agent_token_fixture(organization.id)
|
|
{:ok, _} = AgentsFixtures.agent_assignment_fixture(other_agent.id, other_device.id)
|
|
|
|
result = %SnmpResult{
|
|
device_id: other_device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{other_device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{"1.3.6.1.2.1.1.1.0" => "Test"}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Channel should handle gracefully without crash
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
end
|
|
|
|
# ── Periodic job dispatch ────────────────────────────────────────────
|
|
|
|
describe "periodic job dispatch" do
|
|
test ":send_jobs schedules next dispatch after configured interval", %{socket: socket} do
|
|
# Trigger :send_jobs manually — it should push jobs AND schedule the next cycle
|
|
send(socket.channel_pid, :send_jobs)
|
|
assert_push "jobs", %{binary: _}, 200
|
|
|
|
# The periodic timer is configured to 100ms in test.exs
|
|
# Wait for the next automatic dispatch
|
|
assert_push "jobs", %{binary: _}, 500
|
|
end
|
|
|
|
test ":assignments_changed resets the poll timer and periodic cycle continues", %{socket: socket} do
|
|
# Trigger assignment change — debounced :send_jobs fires after 50ms
|
|
send(socket.channel_pid, {:assignments_changed, :assigned})
|
|
assert_push "jobs", %{binary: _}, 200
|
|
|
|
# After the debounced :send_jobs fires, it should schedule the next periodic cycle
|
|
# The next automatic dispatch should arrive after ~100ms (poll interval)
|
|
assert_push "jobs", %{binary: _}, 500
|
|
end
|
|
|
|
test "no duplicate timers accumulate from rapid :send_jobs calls", %{socket: socket} do
|
|
# Send multiple :send_jobs rapidly — each should cancel the previous timer
|
|
send(socket.channel_pid, :send_jobs)
|
|
assert_push "jobs", %{binary: _}, 200
|
|
|
|
send(socket.channel_pid, :send_jobs)
|
|
assert_push "jobs", %{binary: _}, 200
|
|
|
|
send(socket.channel_pid, :send_jobs)
|
|
assert_push "jobs", %{binary: _}, 200
|
|
|
|
# After the last :send_jobs, wait for the single periodic dispatch
|
|
# If timers accumulated, we'd see multiple rapid pushes here
|
|
assert_push "jobs", %{binary: _}, 300
|
|
|
|
# Only one timer should be running — no extra push within half the poll interval
|
|
refute_push "jobs", %{}, 50
|
|
end
|
|
end
|
|
|
|
# ── Existing tests (preserved) ──────────────────────────────────────
|
|
|
|
describe "poll after discovery" do
|
|
test "pushes poll jobs after successful discovery result", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
# Build a minimal discovery result with system info OIDs
|
|
# that AgentDiscovery.process_agent_discovery can handle
|
|
discovery_result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :DISCOVER,
|
|
job_id: "discover:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
# sysDescr
|
|
"1.3.6.1.2.1.1.1.0" => "Test Device Description",
|
|
# sysObjectID
|
|
"1.3.6.1.2.1.1.2.0" => "1.3.6.1.4.1.9.1.1",
|
|
# sysUpTime
|
|
"1.3.6.1.2.1.1.3.0" => "123456",
|
|
# sysContact
|
|
"1.3.6.1.2.1.1.4.0" => "admin@test.com",
|
|
# sysName
|
|
"1.3.6.1.2.1.1.5.0" => "test-device",
|
|
# sysLocation
|
|
"1.3.6.1.2.1.1.6.0" => "Test Location"
|
|
}
|
|
}
|
|
|
|
binary = SnmpResult.encode(discovery_result)
|
|
payload = %{"binary" => Base.encode64(binary)}
|
|
|
|
# Send the discovery result to the channel
|
|
push(socket, "result", payload)
|
|
|
|
# Should receive a "jobs" push with poll job(s) for the discovered device
|
|
assert_push "jobs", %{binary: jobs_binary}, 1000
|
|
|
|
# Decode the job list and verify it contains a POLL job for our device
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
poll_job =
|
|
Enum.find(job_list.jobs, fn job ->
|
|
job.job_type == :POLL and job.device_id == device.id
|
|
end)
|
|
|
|
assert poll_job != nil, "Expected a POLL job for device #{device.id}"
|
|
assert poll_job.job_id == "poll:#{device.id}"
|
|
end
|
|
|
|
test "poll job includes both HC 64-bit and 32-bit fallback octets for interface stats", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
# Build discovery result with interface data so the poll job
|
|
# will include interface OIDs
|
|
discovery_result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :DISCOVER,
|
|
job_id: "discover:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{
|
|
# System info
|
|
"1.3.6.1.2.1.1.1.0" => "Test Device Description",
|
|
"1.3.6.1.2.1.1.2.0" => "1.3.6.1.4.1.9.1.1",
|
|
"1.3.6.1.2.1.1.3.0" => "123456",
|
|
"1.3.6.1.2.1.1.4.0" => "admin@test.com",
|
|
"1.3.6.1.2.1.1.5.0" => "test-device",
|
|
"1.3.6.1.2.1.1.6.0" => "Test Location",
|
|
# ifTable - interface index
|
|
"1.3.6.1.2.1.2.2.1.1.1" => "1",
|
|
# ifDescr
|
|
"1.3.6.1.2.1.2.2.1.2.1" => "GigabitEthernet0/1",
|
|
# ifType (ethernetCsmacd = 6)
|
|
"1.3.6.1.2.1.2.2.1.3.1" => "6",
|
|
# ifSpeed
|
|
"1.3.6.1.2.1.2.2.1.5.1" => "1000000000",
|
|
# ifPhysAddress
|
|
"1.3.6.1.2.1.2.2.1.6.1" => "aa:bb:cc:dd:ee:ff",
|
|
# ifAdminStatus (up = 1)
|
|
"1.3.6.1.2.1.2.2.1.7.1" => "1",
|
|
# ifOperStatus (up = 1)
|
|
"1.3.6.1.2.1.2.2.1.8.1" => "1"
|
|
}
|
|
}
|
|
|
|
binary = SnmpResult.encode(discovery_result)
|
|
payload = %{"binary" => Base.encode64(binary)}
|
|
|
|
push(socket, "result", payload)
|
|
|
|
# Receive the poll job pushed after discovery
|
|
assert_push "jobs", %{binary: jobs_binary}, 1000
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
poll_job =
|
|
Enum.find(job_list.jobs, fn job ->
|
|
job.job_type == :POLL and job.device_id == device.id
|
|
end)
|
|
|
|
assert poll_job != nil, "Expected a POLL job for device #{device.id}"
|
|
|
|
# Collect all OIDs from all queries in the poll job
|
|
all_oids = Enum.flat_map(poll_job.queries, & &1.oids)
|
|
|
|
# Must include 64-bit HC counters (preferred, for devices that support ifXTable)
|
|
assert Enum.any?(all_oids, &String.starts_with?(&1, "1.3.6.1.2.1.31.1.1.1.6.")),
|
|
"Expected ifHCInOctets OIDs (1.3.6.1.2.1.31.1.1.1.6.x) but found: #{inspect(all_oids)}"
|
|
|
|
assert Enum.any?(all_oids, &String.starts_with?(&1, "1.3.6.1.2.1.31.1.1.1.10.")),
|
|
"Expected ifHCOutOctets OIDs (1.3.6.1.2.1.31.1.1.1.10.x) but found: #{inspect(all_oids)}"
|
|
|
|
# Must also include 32-bit fallback counters (for devices that don't support HC counters)
|
|
assert Enum.any?(all_oids, &String.starts_with?(&1, "1.3.6.1.2.1.2.2.1.10.")),
|
|
"Expected ifInOctets fallback OIDs (1.3.6.1.2.1.2.2.1.10.x) but found: #{inspect(all_oids)}"
|
|
|
|
assert Enum.any?(all_oids, &String.starts_with?(&1, "1.3.6.1.2.1.2.2.1.16.")),
|
|
"Expected ifOutOctets fallback OIDs (1.3.6.1.2.1.2.2.1.16.x) but found: #{inspect(all_oids)}"
|
|
end
|
|
end
|
|
|
|
describe "discovery job OIDs" do
|
|
test "initial discovery job contains expected OIDs", %{socket: socket} do
|
|
# Trigger a fresh job push by sending :send_jobs
|
|
send(socket.channel_pid, :send_jobs)
|
|
|
|
assert_push "jobs", %{binary: jobs_binary}, 1000
|
|
|
|
{:ok, decoded_binary} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded_binary)
|
|
|
|
discover_job =
|
|
Enum.find(job_list.jobs, fn job ->
|
|
job.job_type == :DISCOVER
|
|
end)
|
|
|
|
assert discover_job != nil, "Expected a DISCOVER job in initial jobs"
|
|
|
|
# Collect all OIDs from all queries in the discovery job
|
|
all_oids = Enum.flat_map(discover_job.queries, & &1.oids)
|
|
|
|
# ENTITY-STATE-MIB
|
|
assert "1.3.6.1.2.1.131.1.1.1.1" in all_oids,
|
|
"Missing ENTITY-STATE-MIB OID (1.3.6.1.2.1.131.1.1.1.1)"
|
|
|
|
# HOST-RESOURCES-MIB: hrProcessorTable
|
|
assert "1.3.6.1.2.1.25.3.3" in all_oids,
|
|
"Missing hrProcessorTable OID (1.3.6.1.2.1.25.3.3)"
|
|
|
|
# HOST-RESOURCES-MIB: hrStorageTable
|
|
assert "1.3.6.1.2.1.25.2.3" in all_oids,
|
|
"Missing hrStorageTable OID (1.3.6.1.2.1.25.2.3)"
|
|
|
|
# HOST-RESOURCES-MIB: hrDeviceTable
|
|
assert "1.3.6.1.2.1.25.3.2" in all_oids,
|
|
"Missing hrDeviceTable OID (1.3.6.1.2.1.25.3.2)"
|
|
|
|
# UCD-SNMP-MIB: ssCpuUser
|
|
assert "1.3.6.1.4.1.2021.11.9.0" in all_oids,
|
|
"Missing ssCpuUser OID (1.3.6.1.4.1.2021.11.9.0)"
|
|
end
|
|
end
|
|
|
|
# ── Stage: Device deletion and job refresh ─────────────────────────────────
|
|
|
|
describe "device deletion and job refresh" do
|
|
test "agent receives updated job list after device deletion", %{
|
|
device: device,
|
|
agent_token: agent_token
|
|
} do
|
|
# Create a second device to verify the list shrinks correctly
|
|
device2 =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: device.organization_id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public",
|
|
ip_address: "192.168.1.2"
|
|
})
|
|
|
|
{:ok, _assignment} =
|
|
AgentsFixtures.agent_assignment_fixture(agent_token.id, device2.id)
|
|
|
|
# Trigger initial job refresh to get baseline
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token.id}:assignments",
|
|
{:assignments_changed, :test_setup}
|
|
)
|
|
|
|
# Should receive jobs message with 2 devices
|
|
assert_push "jobs", %{binary: initial_jobs_binary}, 75
|
|
{:ok, decoded_initial} = Base.decode64(initial_jobs_binary)
|
|
{:ok, initial_job_list} = AgentJobList.decode(decoded_initial)
|
|
initial_device_ids = initial_job_list.jobs |> Enum.map(& &1.device_id) |> Enum.uniq()
|
|
assert length(initial_device_ids) == 2
|
|
assert device.id in initial_device_ids
|
|
assert device2.id in initial_device_ids
|
|
|
|
# Delete the first device
|
|
{:ok, _deleted} = Towerops.Devices.delete_device(device)
|
|
|
|
# Should receive updated jobs message with only 1 device
|
|
assert_push "jobs", %{binary: updated_jobs_binary}, 75
|
|
{:ok, decoded_updated} = Base.decode64(updated_jobs_binary)
|
|
{:ok, updated_job_list} = AgentJobList.decode(decoded_updated)
|
|
updated_device_ids = updated_job_list.jobs |> Enum.map(& &1.device_id) |> Enum.uniq()
|
|
|
|
# Verify the deleted device is NOT in the new job list
|
|
assert length(updated_device_ids) == 1
|
|
refute device.id in updated_device_ids
|
|
assert device2.id in updated_device_ids
|
|
end
|
|
|
|
test "agent result for deleted device is rejected", %{socket: socket, device: device} do
|
|
# Delete the device
|
|
{:ok, _deleted} = Towerops.Devices.delete_device(device)
|
|
|
|
# Wait for jobs refresh
|
|
assert_push "jobs", _, 75
|
|
|
|
# Try to send a result for the deleted device
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{"1.3.6.1.2.1.1.1.0" => "Test"}
|
|
}
|
|
|
|
payload = encode_payload(result)
|
|
push(socket, "result", payload)
|
|
|
|
# Channel should remain alive (graceful error handling)
|
|
assert Process.alive?(socket.channel_pid)
|
|
|
|
# The result should be logged as "Device not found" but not crash
|
|
# (This is verified by the channel staying alive)
|
|
end
|
|
|
|
test "multiple rapid assignment changes are debounced", %{agent_token: agent_token} do
|
|
# Trigger multiple rapid assignment changes
|
|
for i <- 1..5 do
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token.id}:assignments",
|
|
{:assignments_changed, "rapid_change_#{i}"}
|
|
)
|
|
|
|
Process.sleep(5)
|
|
end
|
|
|
|
# Should only receive ONE jobs message (debounced) - 50ms debounce in test
|
|
# Use 200ms timeout to account for debounce delay + processing time
|
|
assert_push "jobs", %{binary: _jobs_binary}, 200
|
|
|
|
# Should NOT receive additional jobs messages
|
|
refute_push "jobs", _, 100
|
|
end
|
|
end
|
|
|
|
# ── Stage 13: Discovery error path ──────────────────────────────────
|
|
|
|
describe "discovery error path" do
|
|
test "discovery with empty oid_values does not update last_discovery_at", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
assert is_nil(Towerops.Devices.get_device(device.id).last_discovery_at)
|
|
|
|
# Empty oid_values cause AgentDiscovery to fail (no sysDescr/sysName)
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :DISCOVER,
|
|
job_id: "discover:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# Channel should remain alive even after discovery fails
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
|
|
# last_discovery_at should remain nil since discovery failed
|
|
assert is_nil(Towerops.Devices.get_device(device.id).last_discovery_at)
|
|
end
|
|
end
|
|
|
|
# ── Stage 14: Polling result with status transitions ───────────────
|
|
|
|
describe "process_polling_result status transitions" do
|
|
setup %{device: device} do
|
|
oid_values = %{
|
|
"1.3.6.1.2.1.1.1.0" => "Test Device",
|
|
"1.3.6.1.2.1.1.2.0" => "1.3.6.1.4.1.9.1.1",
|
|
"1.3.6.1.2.1.1.3.0" => "123456",
|
|
"1.3.6.1.2.1.1.4.0" => "admin@test.com",
|
|
"1.3.6.1.2.1.1.5.0" => "test-device",
|
|
"1.3.6.1.2.1.1.6.0" => "Test Location"
|
|
}
|
|
|
|
{:ok, _discovered} = AgentDiscovery.process_agent_discovery(device, oid_values)
|
|
device_with_details = Towerops.Devices.get_device_with_details(device.id)
|
|
%{device: device_with_details}
|
|
end
|
|
|
|
test "successful poll transitions device from down to up and broadcasts", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
{:ok, device} = Towerops.Devices.update_device_status(device, :down)
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
|
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
assert_receive {:device_status_changed, _, :up, _}, 1_000
|
|
|
|
assert poll_until(fn ->
|
|
d = Towerops.Devices.get_device(device.id)
|
|
if d && d.status == :up, do: d
|
|
end)
|
|
end
|
|
|
|
test "poll for already-up device does not re-broadcast status change", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
{:ok, _device} = Towerops.Devices.update_device_status(device, :up)
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
|
|
|
result = %SnmpResult{
|
|
device_id: device.id,
|
|
job_type: :POLL,
|
|
job_id: "poll:#{device.id}",
|
|
timestamp: DateTime.to_unix(DateTime.utc_now()),
|
|
oid_values: %{}
|
|
}
|
|
|
|
push(socket, "result", encode_payload(result))
|
|
|
|
# No status_changed broadcast since status didn't change
|
|
refute_receive {:device_status_changed, _, _, _}, 100
|
|
|
|
# But state_sensors_updated should still come through
|
|
assert_receive {:state_sensors_updated, _}, 1_000
|
|
end
|
|
end
|
|
|
|
# ── Stage 15: Monitoring check status transitions (handle_agent_status_change) ──
|
|
|
|
describe "monitoring check creates device_down alert with correct message" do
|
|
@tag :skip
|
|
test "ping-only device down alert has ping-specific message", %{
|
|
organization: organization
|
|
} do
|
|
# Create a ping-only device (snmp_enabled false)
|
|
ping_device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: organization.id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public",
|
|
snmp_enabled: false,
|
|
monitoring_enabled: true
|
|
})
|
|
|
|
{:ok, agent_token, token_string} =
|
|
AgentsFixtures.agent_token_fixture(organization.id)
|
|
|
|
{:ok, _} = AgentsFixtures.agent_assignment_fixture(agent_token.id, ping_device.id)
|
|
|
|
{:ok, socket} = connect(AgentSocket, %{})
|
|
|
|
{:ok, _, socket} =
|
|
subscribe_and_join(socket, "agent:#{agent_token.id}", %{"token" => token_string})
|
|
|
|
assert_push "jobs", _initial_jobs
|
|
|
|
Towerops.Devices.update_device_status(ping_device, :up)
|
|
|
|
check = build_monitoring_check(ping_device.id, "failure")
|
|
push(socket, "monitoring_check", encode_payload(check))
|
|
|
|
# Poll for alert creation
|
|
poll_until(fn ->
|
|
if Towerops.Alerts.has_active_alert?(ping_device.id, :device_down), do: true
|
|
end)
|
|
|
|
assert Towerops.Alerts.has_active_alert?(ping_device.id, :device_down)
|
|
|
|
# Verify ping-specific message
|
|
[alert] =
|
|
ping_device.id
|
|
|> Towerops.Alerts.list_devices_alerts()
|
|
|> Enum.filter(&(&1.alert_type == "device_down" and is_nil(&1.resolved_at)))
|
|
|
|
assert alert.message =~ "ping"
|
|
|
|
Process.flag(:trap_exit, true)
|
|
leave(socket)
|
|
end
|
|
end
|
|
|
|
# ── Stage 16: check_result handler / check state transitions ───────
|
|
|
|
describe "handle_in check_result" do
|
|
setup %{organization: organization, agent_token: agent_token, device: device} do
|
|
{:ok, check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: organization.id,
|
|
agent_token_id: agent_token.id,
|
|
device_id: device.id,
|
|
check_type: "http",
|
|
name: "Test HTTP Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
# max_check_attempts: 1 means transition to hard state immediately
|
|
max_check_attempts: 1,
|
|
enabled: true,
|
|
config: %{
|
|
"url" => "https://example.com",
|
|
"method" => "GET",
|
|
"expected_status" => 200,
|
|
"verify_ssl" => true
|
|
}
|
|
})
|
|
|
|
%{check: check}
|
|
end
|
|
|
|
test "valid OK check_result is stored", %{socket: socket, check: check} do
|
|
result = %CheckResultProto{
|
|
check_id: check.id,
|
|
status: 0,
|
|
output: "200 OK",
|
|
response_time_ms: 50.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(result))
|
|
|
|
results =
|
|
poll_until(fn ->
|
|
r = Towerops.Monitoring.get_check_results(check.id)
|
|
if r != [], do: r
|
|
end)
|
|
|
|
assert results != []
|
|
assert hd(results).status == 0
|
|
end
|
|
|
|
test "OK → WARNING transition creates alert", %{
|
|
socket: socket,
|
|
check: check,
|
|
device: device
|
|
} do
|
|
Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}")
|
|
|
|
# First seed an OK result so current_state becomes 0 (hard)
|
|
ok_result = %CheckResultProto{
|
|
check_id: check.id,
|
|
status: 0,
|
|
output: "200 OK",
|
|
response_time_ms: 50.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(ok_result))
|
|
|
|
# Wait for state to settle to 0
|
|
poll_until(fn ->
|
|
c = Towerops.Monitoring.get_check(check.id)
|
|
if c && c.current_state == 0, do: c
|
|
end)
|
|
|
|
# Now transition to WARNING — alert should be created
|
|
bad_result = %CheckResultProto{
|
|
check_id: check.id,
|
|
status: 1,
|
|
output: "500 Internal Server Error",
|
|
response_time_ms: 150.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(bad_result))
|
|
|
|
assert poll_until(fn ->
|
|
if Towerops.Alerts.has_active_check_alert?(check.id), do: true
|
|
end)
|
|
end
|
|
|
|
@tag :skip
|
|
test "OK → CRITICAL → OK transition resolves alerts", %{
|
|
socket: socket,
|
|
check: check
|
|
} do
|
|
# Seed OK first so current_state becomes 0
|
|
ok_result = %CheckResultProto{
|
|
check_id: check.id,
|
|
status: 0,
|
|
output: "200 OK",
|
|
response_time_ms: 50.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(ok_result))
|
|
|
|
poll_until(fn ->
|
|
c = Towerops.Monitoring.get_check(check.id)
|
|
if c && c.current_state == 0, do: c
|
|
end)
|
|
|
|
# Transition to CRITICAL — creates alert
|
|
bad_result = %CheckResultProto{
|
|
check_id: check.id,
|
|
status: 2,
|
|
output: "Connection refused",
|
|
response_time_ms: 50.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(bad_result))
|
|
|
|
assert poll_until(fn ->
|
|
if Towerops.Alerts.has_active_check_alert?(check.id), do: true
|
|
end)
|
|
|
|
# Now send OK result to resolve
|
|
ok_again = %CheckResultProto{
|
|
check_id: check.id,
|
|
status: 0,
|
|
output: "200 OK",
|
|
response_time_ms: 50.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(ok_again))
|
|
|
|
# Active alert should clear
|
|
assert poll_until(fn ->
|
|
if !Towerops.Alerts.has_active_check_alert?(check.id), do: true
|
|
end)
|
|
end
|
|
|
|
test "check_result for unknown check is rejected", %{socket: socket} do
|
|
result = %CheckResultProto{
|
|
check_id: Ecto.UUID.generate(),
|
|
status: 0,
|
|
output: "ok",
|
|
response_time_ms: 10.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(result))
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
|
|
test "check_result for check in different organization is rejected", %{socket: socket} do
|
|
other_user = AccountsFixtures.user_fixture()
|
|
other_org = OrganizationsFixtures.organization_fixture(other_user.id)
|
|
|
|
{:ok, other_token, _} = AgentsFixtures.agent_token_fixture(other_org.id)
|
|
|
|
{:ok, other_check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: other_org.id,
|
|
agent_token_id: other_token.id,
|
|
check_type: "http",
|
|
name: "Other Org Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
enabled: true,
|
|
config: %{"url" => "https://example.com"}
|
|
})
|
|
|
|
result = %CheckResultProto{
|
|
check_id: other_check.id,
|
|
status: 0,
|
|
output: "ok",
|
|
response_time_ms: 10.0,
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
push(socket, "check_result", encode_payload(result))
|
|
assert Process.alive?(socket.channel_pid)
|
|
end
|
|
|
|
test "invalid base64 check_result is handled gracefully", %{socket: socket} do
|
|
push(socket, "check_result", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid protobuf check_result is handled gracefully", %{socket: socket} do
|
|
push(socket, "check_result", %{"binary" => Base.encode64("not-a-protobuf")})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 17: lldp_topology_result handler ─────────────────────────
|
|
|
|
describe "handle_in lldp_topology_result" do
|
|
@tag :skip
|
|
test "stores LLDP neighbors from topology result", %{socket: socket, device: device} do
|
|
result = %LldpTopologyResult{
|
|
device_id: device.id,
|
|
job_id: "lldp:#{device.id}",
|
|
local_system_name: "test-device",
|
|
neighbors: [
|
|
%LldpNeighbor{
|
|
neighbor_name: "switch-A",
|
|
local_port: "Eth1/0/1",
|
|
remote_port: "Gi0/1",
|
|
remote_port_id: "Gi0/1",
|
|
management_addresses: ["10.0.0.1"]
|
|
},
|
|
%LldpNeighbor{
|
|
neighbor_name: "switch-B",
|
|
local_port: "Eth1/0/2",
|
|
remote_port: "Gi0/2",
|
|
remote_port_id: "Gi0/2",
|
|
management_addresses: ["10.0.0.2"]
|
|
}
|
|
],
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
binary = Encode.encode_lldp_topology_result(result)
|
|
payload = %{"binary" => Base.encode64(binary)}
|
|
|
|
push(socket, "lldp_topology_result", payload)
|
|
|
|
neighbors =
|
|
poll_until(fn ->
|
|
n = Towerops.Topology.list_lldp_neighbors(device.id)
|
|
if n != [], do: n
|
|
end)
|
|
|
|
assert length(neighbors) == 2
|
|
assert Enum.any?(neighbors, &(&1.neighbor_name == "switch-A"))
|
|
assert Enum.any?(neighbors, &(&1.neighbor_name == "switch-B"))
|
|
end
|
|
|
|
test "empty neighbor list does not crash", %{socket: socket, device: device} do
|
|
result = %LldpTopologyResult{
|
|
device_id: device.id,
|
|
job_id: "lldp:#{device.id}",
|
|
local_system_name: "test-device",
|
|
neighbors: [],
|
|
timestamp: DateTime.to_unix(DateTime.utc_now())
|
|
}
|
|
|
|
binary = Encode.encode_lldp_topology_result(result)
|
|
push(socket, "lldp_topology_result", %{"binary" => Base.encode64(binary)})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid base64 lldp_topology_result is handled gracefully", %{socket: socket} do
|
|
push(socket, "lldp_topology_result", %{"binary" => "not-valid-base64!!!"})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
|
|
test "invalid protobuf lldp_topology_result is handled gracefully", %{socket: socket} do
|
|
push(socket, "lldp_topology_result", %{"binary" => Base.encode64("not-a-protobuf")})
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 18: backup result processing ─────────────────────────────
|
|
|
|
describe "process_backup_result" do
|
|
@describetag :skip
|
|
|
|
test "processes successful manual backup with config sentence", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
job_id = "backup:#{device.id}:#{Ecto.UUID.generate()}"
|
|
{:ok, _request} = BackupRequests.create_request(device.id, job_id, "manual")
|
|
|
|
config_text = """
|
|
# config from RouterOS
|
|
/interface ethernet
|
|
set [ find default-name=ether1 ] name=ether1
|
|
/ip address
|
|
add address=192.168.1.1/24 interface=ether1
|
|
"""
|
|
|
|
sentence = %MikrotikSentence{
|
|
attributes: %{"config" => config_text}
|
|
}
|
|
|
|
result = build_mikrotik_result(device.id, job_id: job_id, sentences: [sentence])
|
|
|
|
push(socket, "mikrotik_result", encode_payload(result))
|
|
|
|
updated_request =
|
|
poll_until(fn ->
|
|
r = BackupRequests.get_request_by_job_id(job_id)
|
|
if r && r.status == "success", do: r
|
|
end)
|
|
|
|
assert updated_request.status == "success"
|
|
|
|
# A backup record should exist
|
|
backups = MikrotikBackups.list_device_backups(device.id)
|
|
assert backups != []
|
|
end
|
|
|
|
test "marks backup failed when result has error", %{socket: socket, device: device} do
|
|
job_id = "backup:#{device.id}:#{Ecto.UUID.generate()}"
|
|
{:ok, _request} = BackupRequests.create_request(device.id, job_id, "manual")
|
|
|
|
result =
|
|
build_mikrotik_result(device.id, job_id: job_id, error: "SSH connection refused")
|
|
|
|
push(socket, "mikrotik_result", encode_payload(result))
|
|
|
|
updated_request =
|
|
poll_until(fn ->
|
|
r = BackupRequests.get_request_by_job_id(job_id)
|
|
if r && r.status == "failed", do: r
|
|
end)
|
|
|
|
assert updated_request.status == "failed"
|
|
assert updated_request.error_message =~ "SSH connection refused"
|
|
end
|
|
|
|
test "marks backup failed when no config can be extracted", %{socket: socket, device: device} do
|
|
job_id = "backup:#{device.id}:#{Ecto.UUID.generate()}"
|
|
{:ok, _request} = BackupRequests.create_request(device.id, job_id, "manual")
|
|
|
|
# Sentence without "config" or "data" attribute - extract should fail
|
|
sentence = %MikrotikSentence{attributes: %{"foo" => "bar"}}
|
|
|
|
result = build_mikrotik_result(device.id, job_id: job_id, sentences: [sentence])
|
|
|
|
push(socket, "mikrotik_result", encode_payload(result))
|
|
|
|
updated_request =
|
|
poll_until(fn ->
|
|
r = BackupRequests.get_request_by_job_id(job_id)
|
|
if r && r.status == "failed", do: r
|
|
end)
|
|
|
|
assert updated_request.status == "failed"
|
|
assert updated_request.error_message =~ "Failed to extract config"
|
|
end
|
|
|
|
test "extracts config from chunked data sentences", %{socket: socket, device: device} do
|
|
job_id = "backup:#{device.id}:#{Ecto.UUID.generate()}"
|
|
{:ok, _request} = BackupRequests.create_request(device.id, job_id, "manual")
|
|
|
|
sentences = [
|
|
%MikrotikSentence{attributes: %{"data" => "/interface ethernet\n"}},
|
|
%MikrotikSentence{attributes: %{"data" => "set [ find default-name=ether1 ] name=ether1\n"}},
|
|
%MikrotikSentence{attributes: %{"data" => "/ip address\nadd address=192.168.1.1/24\n"}}
|
|
]
|
|
|
|
result = build_mikrotik_result(device.id, job_id: job_id, sentences: sentences)
|
|
|
|
push(socket, "mikrotik_result", encode_payload(result))
|
|
|
|
updated_request =
|
|
poll_until(fn ->
|
|
r = BackupRequests.get_request_by_job_id(job_id)
|
|
if r && r.status == "success", do: r
|
|
end)
|
|
|
|
assert updated_request.status == "success"
|
|
|
|
# Verify all chunks were combined into a backup
|
|
[backup | _] = MikrotikBackups.list_device_backups(device.id)
|
|
assert backup.config_text =~ "/interface ethernet"
|
|
assert backup.config_text =~ "set [ find default-name=ether1 ]"
|
|
assert backup.config_text =~ "/ip address"
|
|
end
|
|
|
|
test "missing backup request is logged and ignored", %{socket: socket, device: device} do
|
|
job_id = "backup:#{device.id}:#{Ecto.UUID.generate()}"
|
|
# NO request created — channel should not crash
|
|
|
|
sentence = %MikrotikSentence{attributes: %{"config" => "# config\n/interface\n"}}
|
|
result = build_mikrotik_result(device.id, job_id: job_id, sentences: [sentence])
|
|
|
|
push(socket, "mikrotik_result", encode_payload(result))
|
|
|
|
ref = push(socket, "heartbeat", encode_payload(build_heartbeat()))
|
|
refute_reply ref, :error
|
|
end
|
|
end
|
|
|
|
# ── Stage 19: backup_requested handle_info ─────────────────────────
|
|
|
|
describe "backup_requested via PubSub" do
|
|
test "PubSub broadcast routes backup job to channel", %{
|
|
agent_token: agent_token,
|
|
device: device
|
|
} do
|
|
job_id = "backup:#{device.id}"
|
|
|
|
Phoenix.PubSub.broadcast(
|
|
Towerops.PubSub,
|
|
"agent:#{agent_token.id}:backup",
|
|
{:backup_requested, device.id, job_id}
|
|
)
|
|
|
|
assert_push "backup_job", %{binary: jobs_binary}, 200
|
|
|
|
{:ok, decoded} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded)
|
|
assert hd(job_list.jobs).job_id == "backup:#{device.id}"
|
|
end
|
|
end
|
|
|
|
# ── Stage 20: Mikrotik device job building ──────────────────────────
|
|
|
|
describe "MikroTik device job building" do
|
|
test "MikroTik device with credentials gets a MIKROTIK job in initial dispatch", %{
|
|
organization: organization
|
|
} do
|
|
# Create MikroTik-detected device (manufacturer set after discovery).
|
|
mt_device =
|
|
DevicesFixtures.device_fixture(%{
|
|
organization_id: organization.id,
|
|
snmp_version: "2c",
|
|
snmp_community: "public",
|
|
mikrotik_api_enabled: true,
|
|
mikrotik_username: "admin",
|
|
mikrotik_password: "password123",
|
|
mikrotik_credential_source: "device"
|
|
})
|
|
|
|
{:ok, agent_token, token_string} =
|
|
AgentsFixtures.agent_token_fixture(organization.id)
|
|
|
|
{:ok, _} = AgentsFixtures.agent_assignment_fixture(agent_token.id, mt_device.id)
|
|
|
|
{:ok, socket} = connect(AgentSocket, %{})
|
|
|
|
{:ok, _, _socket} =
|
|
subscribe_and_join(socket, "agent:#{agent_token.id}", %{"token" => token_string})
|
|
|
|
# Initial dispatch should include a discover job (since no snmp_device yet,
|
|
# this device "needs_discovery"). MikroTik job is only built when
|
|
# mikrotik_device?/1 also matches (sys_descr contains "RouterOS" or
|
|
# manufacturer "MikroTik"), so without discovery data we just check that
|
|
# the channel built jobs without crashing.
|
|
assert_push "jobs", %{binary: jobs_binary}
|
|
|
|
{:ok, decoded} = Base.decode64(jobs_binary)
|
|
{:ok, job_list} = AgentJobList.decode(decoded)
|
|
|
|
discover_job = Enum.find(job_list.jobs, &(&1.job_type == :DISCOVER))
|
|
assert discover_job
|
|
assert discover_job.device_id == mt_device.id
|
|
end
|
|
end
|
|
|
|
# ── Stage 21: Latency probe jobs ───────────────────────────────────
|
|
|
|
describe "latency_probe_jobs handle_info" do
|
|
test "sends 5 PING jobs per device for cloud poller", %{
|
|
socket: socket,
|
|
device: device
|
|
} do
|
|
send(socket.channel_pid, {:latency_probe_jobs, [device]})
|
|
|
|
assert_push "jobs", %{binary: jobs_binary}, 500
|
|
|
|
{: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
|
|
assert Enum.all?(ping_jobs, &(&1.device_id == device.id))
|
|
end
|
|
|
|
test "empty device list does not push", %{socket: socket} do
|
|
send(socket.channel_pid, {:latency_probe_jobs, []})
|
|
refute_push "jobs", %{}, 100
|
|
end
|
|
end
|
|
|
|
# ── Stage 22: checks_changed handle_info ───────────────────────────
|
|
|
|
describe "checks_changed handle_info" do
|
|
test "rebuilds and pushes check jobs", %{
|
|
socket: socket,
|
|
organization: organization,
|
|
agent_token: agent_token
|
|
} do
|
|
{:ok, _check} =
|
|
Towerops.Monitoring.create_check(%{
|
|
organization_id: organization.id,
|
|
agent_token_id: agent_token.id,
|
|
check_type: "tcp",
|
|
name: "TCP Check",
|
|
interval_seconds: 60,
|
|
timeout_ms: 5000,
|
|
enabled: true,
|
|
config: %{"host" => "example.com", "port" => 443}
|
|
})
|
|
|
|
send(socket.channel_pid, :checks_changed)
|
|
|
|
assert_push "check_jobs", %{binary: _binary}, 500
|
|
end
|
|
end
|
|
end
|