diff --git a/CHANGELOG.txt b/CHANGELOG.txt index 1ce2c11f..5d256735 100644 --- a/CHANGELOG.txt +++ b/CHANGELOG.txt @@ -1,6 +1,23 @@ CHANGELOG - towerops-web ======================== +2026-02-11 - feat: add sensor value change events for all sensor types + - New file: lib/towerops/snmp/sensor_change_detector.ex + Extracted shared sensor change detection logic from DevicePollerWorker. + Detects threshold violations, spike/drop for % sensors, and new + sensor_value_changed events for all non-% sensors when value changes. + - File: lib/towerops/devices/event.ex + Added "sensor_value_changed" to allowed event types. + - File: lib/towerops/workers/device_poller_worker.ex + Replaced inline detect_sensor_changes with SensorChangeDetector module. + Removed ~240 lines of private functions now in shared module. + - File: lib/towerops_web/channels/agent_channel.ex + Added change detection and last_value/last_checked_at update to + process_sensor_reading/3 so agent-polled devices get same events + as Phoenix-polled devices. + - New file: test/towerops/snmp/sensor_change_detector_test.exs + 11 tests covering value changes, spike/drop, thresholds, nil handling. + 2026-02-11 - fix: live polling uses effective agent from site/org cascade - File: lib/towerops_web/live/graph_live/show.ex - Bug: Live poll mode used Agents.get_device_assignment/1 which only checks diff --git a/lib/towerops/devices/event.ex b/lib/towerops/devices/event.ex index 42049db9..83ad4e7a 100644 --- a/lib/towerops/devices/event.ex +++ b/lib/towerops/devices/event.ex @@ -22,6 +22,7 @@ defmodule Towerops.Devices.Event do | :sensor_threshold_normal | :sensor_value_spike | :sensor_value_drop + | :sensor_value_changed @primary_key {:id, :binary_id, autogenerate: true} @foreign_key_type :binary_id @@ -56,7 +57,8 @@ defmodule Towerops.Devices.Event do "sensor_threshold_critical", "sensor_threshold_normal", "sensor_value_spike", - "sensor_value_drop" + "sensor_value_drop", + "sensor_value_changed" ]) |> validate_inclusion(:severity, ["info", "warning", "critical"]) |> foreign_key_constraint(:device_id) diff --git a/lib/towerops/snmp/sensor_change_detector.ex b/lib/towerops/snmp/sensor_change_detector.ex new file mode 100644 index 00000000..2749742e --- /dev/null +++ b/lib/towerops/snmp/sensor_change_detector.ex @@ -0,0 +1,291 @@ +defmodule Towerops.Snmp.SensorChangeDetector do + @moduledoc """ + Detects sensor value changes and broadcasts events. + + Shared between DevicePollerWorker and AgentChannel to ensure consistent + event detection across both Phoenix-polled and agent-polled devices. + """ + + alias Towerops.Snmp.Sensor + + require Logger + + @doc """ + Detects sensor changes and broadcasts events via PubSub. + + For percentage sensors: emits spike/drop events for >=30% swings. + For all other sensors: emits sensor_value_changed when value differs from last_value. + For all sensors with thresholds: emits threshold warning/critical/normal events. + + No-ops when `sensor.last_value` is nil (first reading). + """ + @spec detect_and_broadcast(Sensor.t(), number(), DateTime.t()) :: :ok + def detect_and_broadcast(%Sensor{last_value: nil}, _current_value, _timestamp), do: :ok + + def detect_and_broadcast(%Sensor{} = sensor, current_value, timestamp) do + device_id = get_device_id_from_sensor(sensor) + + if device_id do + thresholds = extract_thresholds(sensor.metadata) + + events = + [] + |> maybe_add_threshold_event(sensor, current_value, timestamp, device_id, thresholds) + |> maybe_add_change_event(sensor, current_value, timestamp, device_id) + + broadcast_events(events) + end + + :ok + end + + # Threshold detection + + defp extract_thresholds(metadata) do + %{ + warning_high: metadata["warning_high"], + critical_high: metadata["critical_high"], + warning_low: metadata["warning_low"], + critical_low: metadata["critical_low"] + } + end + + defp maybe_add_threshold_event(events, sensor, current_value, timestamp, device_id, thresholds) do + threshold_event = check_threshold_violation(sensor, current_value, timestamp, device_id, thresholds) + + if threshold_event, do: [threshold_event | events], else: events + end + + defp check_threshold_violation(sensor, current_value, timestamp, device_id, thresholds) do + threshold_checks = [ + &check_critical_high/5, + &check_critical_low/5, + &check_warning_high/5, + &check_warning_low/5 + ] + + Enum.find_value(threshold_checks, fn check_fn -> + check_fn.(sensor, current_value, timestamp, device_id, thresholds) + end) || check_returned_to_normal(sensor, current_value, timestamp, device_id, thresholds) + end + + defp check_critical_high(sensor, current_value, timestamp, device_id, thresholds) do + if thresholds.critical_high && current_value >= thresholds.critical_high do + build_threshold_event( + sensor, + current_value, + timestamp, + device_id, + "critical_high", + thresholds.critical_high, + "critical", + "critically high" + ) + end + end + + defp check_critical_low(sensor, current_value, timestamp, device_id, thresholds) do + if thresholds.critical_low && current_value <= thresholds.critical_low do + build_threshold_event( + sensor, + current_value, + timestamp, + device_id, + "critical_low", + thresholds.critical_low, + "critical", + "critically low" + ) + end + end + + defp check_warning_high(sensor, current_value, timestamp, device_id, thresholds) do + if thresholds.warning_high && current_value >= thresholds.warning_high do + build_threshold_event( + sensor, + current_value, + timestamp, + device_id, + "warning_high", + thresholds.warning_high, + "warning", + "high" + ) + end + end + + defp check_warning_low(sensor, current_value, timestamp, device_id, thresholds) do + if thresholds.warning_low && current_value <= thresholds.warning_low do + build_threshold_event( + sensor, + current_value, + timestamp, + device_id, + "warning_low", + thresholds.warning_low, + "warning", + "low" + ) + end + end + + defp build_threshold_event( + sensor, + current_value, + timestamp, + device_id, + threshold_type, + threshold_value, + severity, + description + ) do + event_type = if severity == "critical", do: "sensor_threshold_critical", else: "sensor_threshold_warning" + + build_sensor_event( + device_id, + sensor, + event_type, + severity, + "#{sensor.sensor_descr} is #{description}: #{format_sensor_value(current_value, sensor.sensor_unit)} (threshold: #{threshold_value}#{sensor.sensor_unit})", + %{ + current_value: current_value, + previous_value: sensor.last_value, + threshold_type: threshold_type, + threshold_value: threshold_value + }, + timestamp + ) + end + + defp check_returned_to_normal(sensor, current_value, timestamp, device_id, thresholds) do + was_over_threshold = value_was_over_threshold?(sensor.last_value, thresholds) + is_now_normal = value_is_normal?(current_value, thresholds) + + if was_over_threshold && is_now_normal do + build_sensor_event( + device_id, + sensor, + "sensor_threshold_normal", + "info", + "#{sensor.sensor_descr} returned to normal: #{format_sensor_value(current_value, sensor.sensor_unit)}", + %{current_value: current_value, previous_value: sensor.last_value}, + timestamp + ) + end + end + + defp value_was_over_threshold?(last_value, thresholds) do + (thresholds.warning_high && last_value >= thresholds.warning_high) || + (thresholds.critical_high && last_value >= thresholds.critical_high) || + (thresholds.warning_low && last_value <= thresholds.warning_low) || + (thresholds.critical_low && last_value <= thresholds.critical_low) + end + + defp value_is_normal?(current_value, thresholds) do + (thresholds.warning_high == nil || current_value < thresholds.warning_high) && + (thresholds.critical_high == nil || current_value < thresholds.critical_high) && + (thresholds.warning_low == nil || current_value > thresholds.warning_low) && + (thresholds.critical_low == nil || current_value > thresholds.critical_low) + end + + # Change detection + + defp maybe_add_change_event(events, sensor, current_value, timestamp, device_id) do + if sensor.sensor_unit == "%" do + # Percentage sensors use spike/drop events for large swings + change_event = check_significant_change(sensor, current_value, timestamp, device_id) + if change_event, do: [change_event | events], else: events + else + # All other sensors emit value_changed when value differs + if current_value == sensor.last_value do + events + else + event = + build_sensor_event( + device_id, + sensor, + "sensor_value_changed", + "info", + "#{sensor.sensor_descr} changed from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}", + %{current_value: current_value, previous_value: sensor.last_value}, + timestamp + ) + + [event | events] + end + end + end + + defp check_significant_change(sensor, current_value, timestamp, device_id) do + change_percent = abs(current_value - sensor.last_value) + + cond do + change_percent >= 30 && current_value > sensor.last_value -> + build_sensor_event( + device_id, + sensor, + "sensor_value_spike", + "warning", + "#{sensor.sensor_descr} spiked from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}", + %{current_value: current_value, previous_value: sensor.last_value, change: change_percent}, + timestamp + ) + + change_percent >= 30 && current_value < sensor.last_value -> + build_sensor_event( + device_id, + sensor, + "sensor_value_drop", + "info", + "#{sensor.sensor_descr} dropped from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}", + %{current_value: current_value, previous_value: sensor.last_value, change: change_percent}, + timestamp + ) + + true -> + nil + end + end + + # Event building and broadcasting + + defp build_sensor_event(device_id, sensor, event_type, severity, message, metadata, timestamp) do + base_metadata = %{ + sensor_id: sensor.id, + sensor_name: sensor.sensor_descr, + sensor_type: sensor.sensor_type, + sensor_unit: sensor.sensor_unit + } + + %{ + device_id: device_id, + event_type: event_type, + severity: severity, + message: message, + metadata: Map.merge(base_metadata, metadata), + occurred_at: timestamp + } + end + + defp broadcast_events(events) do + Enum.each(events, fn event -> + _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{event.device_id}", {:device_event, event}) + _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) + + Logger.debug("Sensor event: #{event.message}") + end) + end + + defp get_device_id_from_sensor(sensor) do + case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do + nil -> nil + device -> device.device_id + end + end + + defp format_sensor_value(value, unit) when is_number(value) do + "#{Float.round(value / 1.0, 1)}#{unit}" + end + + defp format_sensor_value(_, _), do: "N/A" +end diff --git a/lib/towerops/workers/device_poller_worker.ex b/lib/towerops/workers/device_poller_worker.ex index efb56775..2a9faf04 100644 --- a/lib/towerops/workers/device_poller_worker.ex +++ b/lib/towerops/workers/device_poller_worker.ex @@ -28,6 +28,7 @@ defmodule Towerops.Workers.DevicePollerWorker do alias Towerops.Snmp.Client alias Towerops.Snmp.MacDiscovery alias Towerops.Snmp.NeighborDiscovery + alias Towerops.Snmp.SensorChangeDetector alias Towerops.Workers.PollingOffset require Logger @@ -731,7 +732,7 @@ defmodule Towerops.Workers.DevicePollerWorker do checked_at: timestamp }) - detect_sensor_changes(sensor, value, timestamp) + SensorChangeDetector.detect_and_broadcast(sensor, value, timestamp) Snmp.update_sensor(sensor, %{ last_value: value, @@ -1266,247 +1267,4 @@ defmodule Towerops.Workers.DevicePollerWorker do seconds_since_poll < min_interval end end - - defp detect_sensor_changes(sensor, current_value, timestamp) do - if sensor.last_value == nil do - :ok - else - device_id = get_device_id_from_sensor(sensor) - thresholds = extract_thresholds(sensor.metadata) - - events = - [] - |> maybe_add_threshold_event(sensor, current_value, timestamp, device_id, thresholds) - |> maybe_add_change_event(sensor, current_value, timestamp, device_id) - - broadcast_sensor_events(events) - end - end - - defp extract_thresholds(metadata) do - %{ - warning_high: metadata["warning_high"], - critical_high: metadata["critical_high"], - warning_low: metadata["warning_low"], - critical_low: metadata["critical_low"] - } - end - - defp maybe_add_threshold_event(events, sensor, current_value, timestamp, device_id, thresholds) do - threshold_event = check_threshold_violation(sensor, current_value, timestamp, device_id, thresholds) - - if threshold_event, do: [threshold_event | events], else: events - end - - defp check_threshold_violation(sensor, current_value, timestamp, device_id, thresholds) do - threshold_checks = [ - &check_critical_high/5, - &check_critical_low/5, - &check_warning_high/5, - &check_warning_low/5 - ] - - Enum.find_value(threshold_checks, fn check_fn -> - check_fn.(sensor, current_value, timestamp, device_id, thresholds) - end) || check_returned_to_normal(sensor, current_value, timestamp, device_id, thresholds) - end - - defp check_critical_high(sensor, current_value, timestamp, device_id, thresholds) do - if thresholds.critical_high && current_value >= thresholds.critical_high do - build_threshold_event( - sensor, - current_value, - timestamp, - device_id, - "critical_high", - thresholds.critical_high, - "critical", - "critically high" - ) - end - end - - defp check_critical_low(sensor, current_value, timestamp, device_id, thresholds) do - if thresholds.critical_low && current_value <= thresholds.critical_low do - build_threshold_event( - sensor, - current_value, - timestamp, - device_id, - "critical_low", - thresholds.critical_low, - "critical", - "critically low" - ) - end - end - - defp check_warning_high(sensor, current_value, timestamp, device_id, thresholds) do - if thresholds.warning_high && current_value >= thresholds.warning_high do - build_threshold_event( - sensor, - current_value, - timestamp, - device_id, - "warning_high", - thresholds.warning_high, - "warning", - "high" - ) - end - end - - defp check_warning_low(sensor, current_value, timestamp, device_id, thresholds) do - if thresholds.warning_low && current_value <= thresholds.warning_low do - build_threshold_event( - sensor, - current_value, - timestamp, - device_id, - "warning_low", - thresholds.warning_low, - "warning", - "low" - ) - end - end - - defp build_threshold_event( - sensor, - current_value, - timestamp, - device_id, - threshold_type, - threshold_value, - severity, - description - ) do - event_type = if severity == "critical", do: "sensor_threshold_critical", else: "sensor_threshold_warning" - - build_sensor_event( - device_id, - sensor, - event_type, - severity, - "#{sensor.sensor_descr} is #{description}: #{format_sensor_value(current_value, sensor.sensor_unit)} (threshold: #{threshold_value}#{sensor.sensor_unit})", - %{ - current_value: current_value, - previous_value: sensor.last_value, - threshold_type: threshold_type, - threshold_value: threshold_value - }, - timestamp - ) - end - - defp check_returned_to_normal(sensor, current_value, timestamp, device_id, thresholds) do - was_over_threshold = value_was_over_threshold?(sensor.last_value, thresholds) - is_now_normal = value_is_normal?(current_value, thresholds) - - if was_over_threshold && is_now_normal do - build_sensor_event( - device_id, - sensor, - "sensor_threshold_normal", - "info", - "#{sensor.sensor_descr} returned to normal: #{format_sensor_value(current_value, sensor.sensor_unit)}", - %{current_value: current_value, previous_value: sensor.last_value}, - timestamp - ) - end - end - - defp value_was_over_threshold?(last_value, thresholds) do - (thresholds.warning_high && last_value >= thresholds.warning_high) || - (thresholds.critical_high && last_value >= thresholds.critical_high) || - (thresholds.warning_low && last_value <= thresholds.warning_low) || - (thresholds.critical_low && last_value <= thresholds.critical_low) - end - - defp value_is_normal?(current_value, thresholds) do - (thresholds.warning_high == nil || current_value < thresholds.warning_high) && - (thresholds.critical_high == nil || current_value < thresholds.critical_high) && - (thresholds.warning_low == nil || current_value > thresholds.warning_low) && - (thresholds.critical_low == nil || current_value > thresholds.critical_low) - end - - defp maybe_add_change_event(events, sensor, current_value, timestamp, device_id) do - if sensor.sensor_unit == "%" do - change_event = check_significant_change(sensor, current_value, timestamp, device_id) - if change_event, do: [change_event | events], else: events - else - events - end - end - - defp check_significant_change(sensor, current_value, timestamp, device_id) do - change_percent = abs(current_value - sensor.last_value) - - cond do - change_percent >= 30 && current_value > sensor.last_value -> - build_sensor_event( - device_id, - sensor, - "sensor_value_spike", - "warning", - "#{sensor.sensor_descr} spiked from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}", - %{current_value: current_value, previous_value: sensor.last_value, change: change_percent}, - timestamp - ) - - change_percent >= 30 && current_value < sensor.last_value -> - build_sensor_event( - device_id, - sensor, - "sensor_value_drop", - "info", - "#{sensor.sensor_descr} dropped from #{format_sensor_value(sensor.last_value, sensor.sensor_unit)} to #{format_sensor_value(current_value, sensor.sensor_unit)}", - %{current_value: current_value, previous_value: sensor.last_value, change: change_percent}, - timestamp - ) - - true -> - nil - end - end - - defp broadcast_sensor_events(events) do - Enum.each(events, fn event -> - _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:#{event.device_id}", {:device_event, event}) - _ = Phoenix.PubSub.broadcast(Towerops.PubSub, "device:events", {:device_event, event}) - - Logger.debug("Sensor event: #{event.message}") - end) - end - - defp build_sensor_event(device_id, sensor, event_type, severity, message, metadata, timestamp) do - base_metadata = %{ - sensor_id: sensor.id, - sensor_name: sensor.sensor_descr, - sensor_type: sensor.sensor_type, - sensor_unit: sensor.sensor_unit - } - - %{ - device_id: device_id, - event_type: event_type, - severity: severity, - message: message, - metadata: Map.merge(base_metadata, metadata), - occurred_at: timestamp - } - end - - defp get_device_id_from_sensor(sensor) do - case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do - nil -> nil - device -> device.device_id - end - end - - defp format_sensor_value(value, unit) when is_number(value) do - "#{Float.round(value, 1)}#{unit}" - end - - defp format_sensor_value(_, _), do: "N/A" end diff --git a/lib/towerops_web/channels/agent_channel.ex b/lib/towerops_web/channels/agent_channel.ex index 4180d314..80670fcd 100644 --- a/lib/towerops_web/channels/agent_channel.ex +++ b/lib/towerops_web/channels/agent_channel.ex @@ -38,6 +38,7 @@ defmodule ToweropsWeb.AgentChannel do alias Towerops.Snmp alias Towerops.Snmp.AgentDiscovery alias Towerops.Snmp.Discovery + alias Towerops.Snmp.SensorChangeDetector require Logger @@ -1547,6 +1548,13 @@ defmodule ToweropsWeb.AgentChannel do status: "ok", checked_at: timestamp }) + + SensorChangeDetector.detect_and_broadcast(sensor, final_value, timestamp) + + Snmp.update_sensor(sensor, %{ + last_value: final_value, + last_checked_at: timestamp + }) end end end diff --git a/test/towerops/snmp/sensor_change_detector_test.exs b/test/towerops/snmp/sensor_change_detector_test.exs new file mode 100644 index 00000000..dbb183a2 --- /dev/null +++ b/test/towerops/snmp/sensor_change_detector_test.exs @@ -0,0 +1,310 @@ +defmodule Towerops.Snmp.SensorChangeDetectorTest do + use Towerops.DataCase, async: true + + import Towerops.AccountsFixtures + + alias Towerops.Devices + alias Towerops.Organizations + alias Towerops.Repo + alias Towerops.Sites + alias Towerops.Snmp.Device, as: SnmpDevice + alias Towerops.Snmp.Sensor + alias Towerops.Snmp.SensorChangeDetector + + setup do + user = user_fixture() + {:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id) + + {:ok, site} = + Sites.create_site(%{ + name: "Test Site", + organization_id: organization.id + }) + + {:ok, device} = + Devices.create_device(%{ + name: "Test Device", + ip_address: "192.168.1.1", + site_id: site.id, + organization_id: organization.id, + snmp_enabled: true + }) + + {:ok, snmp_device} = + Repo.insert(%SnmpDevice{ + device_id: device.id, + sys_name: "test-device", + sys_descr: "Test Device" + }) + + %{device: device, snmp_device: snmp_device} + end + + defp create_sensor(snmp_device, attrs \\ %{}) do + defaults = %{ + snmp_device_id: snmp_device.id, + sensor_type: "frequency", + sensor_index: "1", + sensor_oid: "1.3.6.1.2.1.99.1.1.1.4.1", + sensor_descr: "Power Supply 1 Frequency", + sensor_unit: "Hz", + sensor_divisor: 1, + monitored: true, + metadata: %{} + } + + {:ok, sensor} = Repo.insert(struct(Sensor, Map.merge(defaults, attrs))) + sensor + end + + describe "detect_and_broadcast/3" do + test "emits sensor_value_changed for non-% sensor when value changes", %{ + device: device, + snmp_device: snmp_device + } do + sensor = create_sensor(snmp_device, %{last_value: 50.0, sensor_unit: "Hz"}) + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 49.9, timestamp) + + assert_receive {:device_event, event} + assert event.event_type == "sensor_value_changed" + assert event.severity == "info" + assert event.device_id == device.id + assert event.metadata.current_value == 49.9 + assert event.metadata.previous_value == 50.0 + assert event.metadata.sensor_id == sensor.id + assert String.contains?(event.message, "changed from") + assert String.contains?(event.message, "50.0Hz") + assert String.contains?(event.message, "49.9Hz") + end + + test "emits sensor_value_changed for temperature sensor", %{ + device: device, + snmp_device: snmp_device + } do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "temperature", + sensor_descr: "CPU Temp", + sensor_unit: "C", + last_value: 45.0 + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 46.0, timestamp) + + assert_receive {:device_event, event} + assert event.event_type == "sensor_value_changed" + assert String.contains?(event.message, "CPU Temp") + assert String.contains?(event.message, "45.0C") + assert String.contains?(event.message, "46.0C") + end + + test "does not emit event when value is unchanged", %{snmp_device: snmp_device} do + sensor = create_sensor(snmp_device, %{last_value: 50.0, sensor_unit: "Hz"}) + timestamp = DateTime.utc_now() + + # Subscribe to both topics + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events") + + SensorChangeDetector.detect_and_broadcast(sensor, 50.0, timestamp) + + refute_receive {:device_event, _}, 100 + end + + test "does not emit sensor_value_changed when last_value is nil", %{snmp_device: snmp_device} do + sensor = create_sensor(snmp_device, %{last_value: nil, sensor_unit: "Hz"}) + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events") + + SensorChangeDetector.detect_and_broadcast(sensor, 50.0, timestamp) + + refute_receive {:device_event, _}, 100 + end + + test "emits spike event for % sensor with >=30% increase", %{ + device: device, + snmp_device: snmp_device + } do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "percent", + sensor_descr: "CPU Usage", + sensor_unit: "%", + last_value: 20.0, + metadata: %{} + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 55.0, timestamp) + + assert_receive {:device_event, event} + assert event.event_type == "sensor_value_spike" + assert event.severity == "warning" + end + + test "emits drop event for % sensor with >=30% decrease", %{ + device: device, + snmp_device: snmp_device + } do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "percent", + sensor_descr: "CPU Usage", + sensor_unit: "%", + last_value: 80.0, + metadata: %{} + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 40.0, timestamp) + + assert_receive {:device_event, event} + assert event.event_type == "sensor_value_drop" + assert event.severity == "info" + end + + test "does not emit spike/drop for % sensor with <30% change", %{snmp_device: snmp_device} do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "percent", + sensor_descr: "CPU Usage", + sensor_unit: "%", + last_value: 50.0, + metadata: %{} + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events") + + SensorChangeDetector.detect_and_broadcast(sensor, 55.0, timestamp) + + # Should not get spike/drop events for small % changes + # (no sensor_value_changed for % sensors either, they use spike/drop) + refute_receive {:device_event, _}, 100 + end + + test "emits threshold warning event", %{ + device: device, + snmp_device: snmp_device + } do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "temperature", + sensor_descr: "Chassis Temp", + sensor_unit: "C", + last_value: 60.0, + metadata: %{"warning_high" => 70, "critical_high" => 90} + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 75.0, timestamp) + + # Should get both threshold warning AND value changed + events = collect_events(2) + + event_types = Enum.map(events, & &1.event_type) + assert "sensor_threshold_warning" in event_types + assert "sensor_value_changed" in event_types + end + + test "emits threshold critical event", %{ + device: device, + snmp_device: snmp_device + } do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "temperature", + sensor_descr: "Chassis Temp", + sensor_unit: "C", + last_value: 60.0, + metadata: %{"warning_high" => 70, "critical_high" => 90} + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 95.0, timestamp) + + events = collect_events(2) + + event_types = Enum.map(events, & &1.event_type) + assert "sensor_threshold_critical" in event_types + assert "sensor_value_changed" in event_types + end + + test "emits threshold normal event when returning from warning", %{ + device: device, + snmp_device: snmp_device + } do + sensor = + create_sensor(snmp_device, %{ + sensor_type: "temperature", + sensor_descr: "Chassis Temp", + sensor_unit: "C", + last_value: 75.0, + metadata: %{"warning_high" => 70, "critical_high" => 90} + }) + + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + + SensorChangeDetector.detect_and_broadcast(sensor, 65.0, timestamp) + + events = collect_events(2) + + event_types = Enum.map(events, & &1.event_type) + assert "sensor_threshold_normal" in event_types + assert "sensor_value_changed" in event_types + end + + test "broadcasts to both device and global event topics", %{ + device: device, + snmp_device: snmp_device + } do + sensor = create_sensor(snmp_device, %{last_value: 50.0, sensor_unit: "Hz"}) + timestamp = DateTime.utc_now() + + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:#{device.id}") + Phoenix.PubSub.subscribe(Towerops.PubSub, "device:events") + + SensorChangeDetector.detect_and_broadcast(sensor, 49.0, timestamp) + + # Should receive on both topics + assert_receive {:device_event, event1} + assert_receive {:device_event, event2} + assert event1.event_type == "sensor_value_changed" + assert event2.event_type == "sensor_value_changed" + end + end + + # Helper to collect multiple events + defp collect_events(count) do + Enum.map(1..count, fn _ -> + receive do + {:device_event, event} -> event + after + 500 -> flunk("Expected #{count} events but timed out") + end + end) + end +end