From f00dc3ff12ed62ccd04b385f14f116038db5b5b7 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Wed, 11 Feb 2026 16:52:12 -0600 Subject: [PATCH] test: expand agent channel coverage from 8% to 66% add 54 tests covering join/terminate lifecycle, heartbeat, error, credential_test_result, monitoring_check, mikrotik_result, catch-all handle_in events, handle_info lifecycle messages, PubSub messages, SNMP result edge cases, poll result sensor/interface data storage, SNMPv3 credential paths, and monitoring-enabled device job building. fix bug where error.error_message was used instead of error.message matching the AgentError protobuf struct field name. --- lib/towerops_web/channels/agent_channel.ex | 4 +- .../channels/agent_channel_test.exs | 986 ++++++++++++++++++ 2 files changed, 988 insertions(+), 2 deletions(-) diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index e99b7ca4..ef4dc2a1 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -456,14 +456,14 @@ defmodule ToweropsWeb.AgentChannel do {:ok, error} <- Validator.validate_agent_error(binary) do maybe_debug_log(socket, "Agent job error", device_id: error.device_id, - error_message: error.error_message, + error_message: error.message, binary_size: byte_size(binary_b64) ) Logger.error("Agent job error", agent_token_id: socket.assigns.agent_token_id, device_id: error.device_id, - error: error.error_message + error: error.message ) {:noreply, socket} diff --git a/test/towerops_web/channels/agent_channel_test.exs b/test/towerops_web/channels/agent_channel_test.exs index cc4f4e66..60b1dcb9 100644 --- a/test/towerops_web/channels/agent_channel_test.exs +++ b/test/towerops_web/channels/agent_channel_test.exs @@ -4,11 +4,21 @@ defmodule ToweropsWeb.AgentChannelTest do import Phoenix.ChannelTest alias Towerops.AccountsFixtures + alias Towerops.Agent.AgentError + alias Towerops.Agent.AgentHeartbeat + alias Towerops.Agent.AgentJob alias Towerops.Agent.AgentJobList + 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.DevicesFixtures alias Towerops.OrganizationsFixtures + alias Towerops.Snmp.AgentDiscovery + alias Towerops.Snmp.Sensor alias ToweropsWeb.AgentSocket @endpoint ToweropsWeb.Endpoint @@ -45,10 +55,986 @@ defmodule ToweropsWeb.AgentChannelTest do 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 + + 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 + heartbeat = build_heartbeat(%{version: "2.0.0"}) + payload = encode_payload(heartbeat) + + push(socket, "heartbeat", payload) + + # Give async processing a moment + Process.sleep(100) + + updated_token = Agents.get_agent_token!(agent_token.id) + assert updated_token.last_seen_at + assert updated_token.metadata["version"] == "2.0.0" + end + + test "valid heartbeat broadcasts to agents:health", %{socket: socket, agent_token: agent_token} do + Phoenix.PubSub.subscribe(Towerops.PubSub, "agents:health") + + agent_token_id = agent_token.id + heartbeat = build_heartbeat() + payload = encode_payload(heartbeat) + + push(socket, "heartbeat", payload) + + assert_receive {:agent_heartbeat, ^agent_token_id, _org_id} + 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 - verify with another message + Process.sleep(50) + 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 + check = build_monitoring_check(device.id, "success") + payload = encode_payload(check) + + push(socket, "monitoring_check", payload) + + Process.sleep(100) + + # Verify check was stored + checks = Towerops.Monitoring.list_devices_checks(device.id) + 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) + + Process.sleep(100) + + updated_device = Towerops.Devices.get_device!(device.id) + 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) + + Process.sleep(100) + + updated_device = Towerops.Devices.get_device!(device.id) + assert updated_device.status == :down + + # Verify alert was created + 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) + + Process.sleep(100) + + updated_device = Towerops.Devices.get_device!(device.id) + 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 + + # ── 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) + job_list = AgentJobList.decode(decoded_binary) + assert is_list(job_list.jobs) + 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 + Process.sleep(50) + 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, _, _}, 5_000 + 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, _, _}, 5_000 + 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, _, _}, 5_000 + 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 500ms debounce + assert_push "jobs", %{binary: _jobs_binary}, 2_000 + 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) + assert_push "jobs", %{binary: _}, 2_000 + refute_push "jobs", %{}, 1_000 + 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) + 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 + Process.sleep(50) + 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) + 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) + 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"} + ) + + Process.sleep(50) + 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, job}" do + test "pushes backup job to agent", %{socket: socket, device: device} do + job = %AgentJob{ + job_id: "backup:#{device.id}", + job_type: :MIKROTIK, + device_id: device.id, + mikrotik_device: nil, + mikrotik_commands: [] + } + + send(socket.channel_pid, {:backup_requested, job}) + + assert_push "backup_job", %{binary: jobs_binary} + + {:ok, decoded_binary} = Base.decode64(jobs_binary) + job_list = AgentJobList.decode(decoded_binary) + + backup_job = hd(job_list.jobs) + assert backup_job.job_id == "backup:#{device.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}, 5_000 + + {:ok, decoded_binary} = Base.decode64(jobs_binary) + 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()}) + + Process.sleep(50) + 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 + Process.sleep(50) + 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 + Process.sleep(50) + 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}, 2_000 + 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) + + %{ + 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)) + Process.sleep(200) + + # Verify sensor reading was stored + updated_sensor = Towerops.Repo.get!(Sensor, sensor.id) + assert updated_sensor.last_value == 45.0 + 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)) + Process.sleep(200) + + # Verify interface stat was stored + stats = Towerops.Snmp.get_interface_stats(interface.id) + 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)) + Process.sleep(200) + + updated_sensor = Towerops.Repo.get!(Sensor, sensor.id) + 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)) + Process.sleep(200) + + updated_sensor = Towerops.Repo.get!(Sensor, sensor_with_divisor.id) + assert updated_sensor.last_value == 12.0 + 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) + 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) + 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 + Process.sleep(50) + assert Process.alive?(socket.channel_pid) + end + end + + # ── Existing tests (preserved) ────────────────────────────────────── + describe "poll after discovery" do test "pushes poll jobs after successful discovery result", %{ socket: socket,