feat: add APC, Arista, Calix, ADTRAN, Allied Telesis vendor modules

Add 5 new vendor modules for network infrastructure equipment:
- APC: UPS and PDU monitoring (battery, voltage, load, runtime)
- Arista EOS: Data center switches (CPU, memory)
- Calix: GPON/fiber access equipment
- ADTRAN AOS: NetVanta routers and switches
- Allied Telesis: AlliedWare/AlliedWare Plus switches

Total: 53 vendors, 88 profile names

Also cleanup commented AlertNotifier code from previous removal.
This commit is contained in:
Graham McIntire 2026-01-22 11:00:21 -06:00
parent aec9807e97
commit 155b38ab94
No known key found for this signature in database
14 changed files with 1039 additions and 32 deletions

View file

@ -142,20 +142,6 @@ defmodule Towerops.Alerts do
|> Repo.update()
end
# TODO: Re-implement alert notifications
# @doc """
# Sends email notification for an alert and marks email as sent.
# """
# def send_alert_notification(%Alert{} = alert) do
# alias Towerops.Alerts.AlertNotifier
#
# {:ok, _results} = AlertNotifier.deliver_alert_notification(alert)
#
# alert
# |> Alert.changeset(%{email_sent_at: DateTime.truncate(DateTime.utc_now(), :second)})
# |> Repo.update()
# end
@doc """
Checks if there's an active alert of the same type for the device.
"""

View file

@ -0,0 +1,77 @@
defmodule Towerops.Snmp.Profiles.Vendors.Adtran do
@moduledoc """
ADTRAN device-specific SNMP handling.
Supports ADTRAN AOS routers and switches.
Supported devices:
- NetVanta 1500/1600 series switches
- NetVanta 3000/4000 series routers
- NetVanta 5000 series
- ADTRAN Total Access series
Enterprise OID: 1.3.6.1.4.1.664
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# ADTRAN model OID
@model_oid "1.3.6.1.4.1.664.1.1.2.0"
@impl true
def profile_names, do: ["adtran-aos"]
@impl true
def detect_hardware(client_opts) do
case Client.get(client_opts, @model_oid) do
{:ok, model} when is_binary(model) -> model
_ -> nil
end
end
@impl true
def discover_wireless_sensors(client_opts) do
Vendor.fetch_sensors(wireless_oid_defs(), client_opts)
end
@impl true
def wireless_oid_defs do
[
# Current CPU Utilization
%{
oid: "1.3.6.1.4.1.664.5.53.1.4.1.0",
sensor_type: "load",
sensor_descr: "CPU Utilization",
sensor_unit: "%",
sensor_divisor: 1
},
# 5-minute CPU Average
%{
oid: "1.3.6.1.4.1.664.5.53.1.4.4.0",
sensor_type: "load",
sensor_descr: "CPU 5min Average",
sensor_unit: "%",
sensor_divisor: 1
},
# Available Memory
%{
oid: "1.3.6.1.4.1.664.5.53.1.4.8.0",
sensor_type: "count",
sensor_descr: "Available Memory",
sensor_unit: "bytes",
sensor_divisor: 1
},
# System Uptime
%{
oid: "1.3.6.1.2.1.1.3.0",
sensor_type: "count",
sensor_descr: "System Uptime",
sensor_unit: "ticks",
sensor_divisor: 1
}
]
end
end

View file

@ -0,0 +1,93 @@
defmodule Towerops.Snmp.Profiles.Vendors.AlliedTelesis do
@moduledoc """
Allied Telesis device-specific SNMP handling.
Supports Allied Telesis switches and routers.
Supported devices:
- AlliedWare Plus switches (x-series, SBx series)
- AlliedWare switches (AT-8000 series)
- Allied Telesis managed switches
Enterprise OID: 1.3.6.1.4.1.207
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# Allied Telesis sysDescr
@sysdescr_oid "1.3.6.1.2.1.1.1.0"
@impl true
def profile_names, do: ["allied", "awplus", "allied-tq"]
@impl true
def detect_hardware(client_opts) do
# Model often in sysDescr
case Client.get(client_opts, @sysdescr_oid) do
{:ok, descr} when is_binary(descr) ->
cond do
match = Regex.run(~r/(AT-\S+|x\d+\S*)/i, descr) -> List.first(tl(match))
String.contains?(descr, "AlliedWare Plus") -> "AlliedWare Plus"
String.contains?(descr, "AlliedWare") -> "AlliedWare"
true -> nil
end
_ ->
nil
end
end
@impl true
def discover_wireless_sensors(client_opts) do
Vendor.fetch_sensors(wireless_oid_defs(), client_opts)
end
@impl true
def wireless_oid_defs do
[
# CPU Utilization 5-minute Average (AT-SYSINFO-MIB)
%{
oid: "1.3.6.1.4.1.207.8.4.4.3.3.7.0",
sensor_type: "load",
sensor_descr: "CPU 5min Average",
sensor_unit: "%",
sensor_divisor: 1
},
# Free Memory (AT-SYSINFO-MIB)
%{
oid: "1.3.6.1.4.1.207.8.4.4.3.7.1.0",
sensor_type: "count",
sensor_descr: "Free Memory",
sensor_unit: "KB",
sensor_divisor: 1
},
# Temperature Current (AT-ENVMONv2-MIB, index 1)
%{
oid: "1.3.6.1.4.1.207.8.4.4.3.12.3.1.5.1",
sensor_type: "temperature",
sensor_descr: "Temperature",
sensor_unit: "°C",
sensor_divisor: 1
},
# Fan Speed (AT-ENVMONv2-MIB, index 1)
%{
oid: "1.3.6.1.4.1.207.8.4.4.3.12.1.1.5.1",
sensor_type: "fanspeed",
sensor_descr: "Fan Speed",
sensor_unit: "RPM",
sensor_divisor: 1
},
# System Uptime
%{
oid: "1.3.6.1.2.1.1.3.0",
sensor_type: "count",
sensor_descr: "System Uptime",
sensor_unit: "ticks",
sensor_divisor: 1
}
]
end
end

View file

@ -0,0 +1,118 @@
defmodule Towerops.Snmp.Profiles.Vendors.Apc do
@moduledoc """
APC (Schneider Electric) device-specific SNMP handling.
Supports APC UPS units and PDUs.
Supported devices:
- Smart-UPS series
- Symmetra UPS
- Back-UPS Pro
- Switched/Metered Rack PDU
- Environmental monitoring cards
Enterprise OID: 1.3.6.1.4.1.318
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# APC model identifier
@model_oid "1.3.6.1.4.1.318.1.1.1.1.1.1.0"
@impl true
def profile_names, do: ["apc", "apc-cpdu", "apc-epdu", "apc-mgeups", "apc-netbotz"]
@impl true
def detect_hardware(client_opts) do
case Client.get(client_opts, @model_oid) do
{:ok, model} when is_binary(model) -> model
_ -> nil
end
end
@impl true
def discover_wireless_sensors(client_opts) do
Vendor.fetch_sensors(wireless_oid_defs(), client_opts)
end
@impl true
def wireless_oid_defs do
[
# Battery Capacity %
%{
oid: "1.3.6.1.4.1.318.1.1.1.2.2.1.0",
sensor_type: "load",
sensor_descr: "Battery Capacity",
sensor_unit: "%",
sensor_divisor: 1
},
# Battery Temperature (Celsius)
%{
oid: "1.3.6.1.4.1.318.1.1.1.2.2.2.0",
sensor_type: "temperature",
sensor_descr: "Battery Temperature",
sensor_unit: "°C",
sensor_divisor: 1
},
# Runtime Remaining (timeticks - needs conversion)
%{
oid: "1.3.6.1.4.1.318.1.1.1.2.2.3.0",
sensor_type: "runtime",
sensor_descr: "Runtime Remaining",
sensor_unit: "min",
sensor_divisor: 6000
},
# Output Voltage
%{
oid: "1.3.6.1.4.1.318.1.1.1.4.2.1.0",
sensor_type: "voltage",
sensor_descr: "Output Voltage",
sensor_unit: "V",
sensor_divisor: 1
},
# Output Frequency
%{
oid: "1.3.6.1.4.1.318.1.1.1.4.2.2.0",
sensor_type: "frequency",
sensor_descr: "Output Frequency",
sensor_unit: "Hz",
sensor_divisor: 1
},
# Output Load %
%{
oid: "1.3.6.1.4.1.318.1.1.1.4.2.3.0",
sensor_type: "load",
sensor_descr: "Output Load",
sensor_unit: "%",
sensor_divisor: 1
},
# Output Current
%{
oid: "1.3.6.1.4.1.318.1.1.1.4.2.4.0",
sensor_type: "current",
sensor_descr: "Output Current",
sensor_unit: "A",
sensor_divisor: 1
},
# Input Voltage
%{
oid: "1.3.6.1.4.1.318.1.1.1.3.2.1.0",
sensor_type: "voltage",
sensor_descr: "Input Voltage",
sensor_unit: "V",
sensor_divisor: 1
},
# Input Frequency
%{
oid: "1.3.6.1.4.1.318.1.1.1.3.2.4.0",
sensor_type: "frequency",
sensor_descr: "Input Frequency",
sensor_unit: "Hz",
sensor_divisor: 1
}
]
end
end

View file

@ -0,0 +1,86 @@
defmodule Towerops.Snmp.Profiles.Vendors.Arista do
@moduledoc """
Arista Networks device-specific SNMP handling.
Supports Arista EOS switches.
Supported devices:
- Arista 7000 series switches
- Arista 7050 series
- Arista 7150 series
- Arista 7280 series
- Arista 7500 series
Enterprise OID: 1.3.6.1.4.1.30065
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# Arista model name from sysDescr parsing
@sysdescr_oid "1.3.6.1.2.1.1.1.0"
@impl true
def profile_names, do: ["arista_eos", "arista-mos"]
@impl true
def detect_hardware(client_opts) do
# Arista includes model in sysDescr: "Arista Networks EOS version ... running on an Arista Networks DCS-7050TX-64"
case Client.get(client_opts, @sysdescr_oid) do
{:ok, descr} when is_binary(descr) ->
case Regex.run(~r/running on an Arista Networks (\S+)/i, descr) do
[_, model] -> model
_ -> nil
end
_ ->
nil
end
end
@impl true
def discover_wireless_sensors(client_opts) do
Vendor.fetch_sensors(wireless_oid_defs(), client_opts)
end
@impl true
def wireless_oid_defs do
[
# CPU Load (HOST-RESOURCES-MIB - average of all CPUs)
%{
oid: "1.3.6.1.2.1.25.3.3.1.2.1",
sensor_type: "load",
sensor_descr: "CPU Load",
sensor_unit: "%",
sensor_divisor: 1
},
# Memory Storage Size (hrStorageSize for RAM, index 1)
# Note: hrStorage values need context - this gives total RAM allocation units
%{
oid: "1.3.6.1.2.1.25.2.3.1.5.1",
sensor_type: "count",
sensor_descr: "Memory Size",
sensor_unit: "units",
sensor_divisor: 1
},
# Memory Storage Used (hrStorageUsed for RAM, index 1)
%{
oid: "1.3.6.1.2.1.25.2.3.1.6.1",
sensor_type: "count",
sensor_descr: "Memory Used",
sensor_unit: "units",
sensor_divisor: 1
},
# System Uptime
%{
oid: "1.3.6.1.2.1.1.3.0",
sensor_type: "count",
sensor_descr: "System Uptime",
sensor_unit: "ticks",
sensor_divisor: 1
}
]
end
end

View file

@ -0,0 +1,73 @@
defmodule Towerops.Snmp.Profiles.Vendors.Calix do
@moduledoc """
Calix device-specific SNMP handling.
Supports Calix GPON/fiber access equipment.
Supported devices:
- Calix E7-2 AXOS systems
- Calix E7 EXA nodes
- Calix C7 platforms
- Calix GigaSpire/GigaPoint ONTs
Enterprise OID: 1.3.6.1.4.1.6321
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# Calix system description
@sysdescr_oid "1.3.6.1.2.1.1.1.0"
@impl true
def profile_names, do: ["calix"]
@impl true
def detect_hardware(client_opts) do
# Calix model typically in sysDescr
case Client.get(client_opts, @sysdescr_oid) do
{:ok, descr} when is_binary(descr) ->
cond do
String.contains?(descr, "E7-2") -> "E7-2"
String.contains?(descr, "E7") -> "E7"
String.contains?(descr, "C7") -> "C7"
String.contains?(descr, "AXOS") -> "AXOS"
true -> nil
end
_ ->
nil
end
end
@impl true
def discover_wireless_sensors(client_opts) do
Vendor.fetch_sensors(wireless_oid_defs(), client_opts)
end
@impl true
def wireless_oid_defs do
# Calix GPON OIDs are not publicly documented
# Using standard MIB OIDs for basic system metrics
[
# CPU Load (HOST-RESOURCES-MIB)
%{
oid: "1.3.6.1.2.1.25.3.3.1.2.1",
sensor_type: "load",
sensor_descr: "CPU Load",
sensor_unit: "%",
sensor_divisor: 1
},
# System Uptime
%{
oid: "1.3.6.1.2.1.1.3.0",
sensor_type: "count",
sensor_descr: "System Uptime",
sensor_unit: "ticks",
sensor_divisor: 1
}
]
end
end

View file

@ -7,14 +7,19 @@ defmodule Towerops.Snmp.Profiles.Vendors.Registry do
"""
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Adtran
alias Towerops.Snmp.Profiles.Vendors.Aerohive
alias Towerops.Snmp.Profiles.Vendors.Airfiber
alias Towerops.Snmp.Profiles.Vendors.Airos
alias Towerops.Snmp.Profiles.Vendors.AlliedTelesis
alias Towerops.Snmp.Profiles.Vendors.Altalabs
alias Towerops.Snmp.Profiles.Vendors.Alvarion
alias Towerops.Snmp.Profiles.Vendors.Apc
alias Towerops.Snmp.Profiles.Vendors.Arista
alias Towerops.Snmp.Profiles.Vendors.Aruba
alias Towerops.Snmp.Profiles.Vendors.Aviat
alias Towerops.Snmp.Profiles.Vendors.Baicells
alias Towerops.Snmp.Profiles.Vendors.Calix
alias Towerops.Snmp.Profiles.Vendors.CambiumPtp
alias Towerops.Snmp.Profiles.Vendors.Ceragon
alias Towerops.Snmp.Profiles.Vendors.Ciscowap
@ -57,14 +62,19 @@ defmodule Towerops.Snmp.Profiles.Vendors.Registry do
alias Towerops.Snmp.Profiles.Vendors.ZyxelWlc
@vendors [
Adtran,
Aerohive,
Airfiber,
Airos,
AlliedTelesis,
Altalabs,
Alvarion,
Apc,
Arista,
Aruba,
Aviat,
Baicells,
Calix,
CambiumPtp,
Ceragon,
Ciscowap,

View file

@ -328,24 +328,6 @@ defmodule Towerops.AlertsTest do
end
end
# TODO: Re-implement alert notification test
# test "send_alert_notification/1 sends notification and marks email as sent", %{
# device: device
# } do
# attrs = Map.put(@valid_attrs, :device_id, device.id)
# {:ok, alert} = Alerts.create_alert(attrs)
#
# # Reload to get associations
# alert = Alerts.get_alert!(alert.id)
#
# refute alert.email_sent_at
#
# {:ok, updated_alert} = Alerts.send_alert_notification(alert)
#
# assert updated_alert.email_sent_at
# assert %DateTime{} = updated_alert.email_sent_at
# end
test "get_active_alert/2 returns most recent when multiple active alerts", %{
device: device
} do

View file

@ -0,0 +1,97 @@
defmodule Towerops.Snmp.Profiles.Vendors.AdtranTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Adtran
alias Towerops.Snmp.SnmpMock
setup :verify_on_exit!
@client_opts [
ip: "192.168.1.1",
community: "public",
version: "2c",
port: 161,
timeout: 5000
]
describe "profile_names/0" do
test "returns ADTRAN profile names" do
assert Adtran.profile_names() == ["adtran-aos"]
end
end
describe "detect_hardware/1" do
test "returns model when SNMP responds" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.664.1.1.2.0", _ ->
{:ok, "NetVanta 1531P"}
end)
assert Adtran.detect_hardware(@client_opts) == "NetVanta 1531P"
end
test "returns nil when SNMP fails" do
expect(SnmpMock, :get, fn _, _, _ ->
{:error, :timeout}
end)
assert Adtran.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of sensor definitions" do
defs = Adtran.wireless_oid_defs()
assert is_list(defs)
assert [_ | _] = defs
end
test "includes CPU utilization sensor" do
defs = Adtran.wireless_oid_defs()
cpu = Enum.find(defs, &(&1.sensor_descr == "CPU Utilization"))
assert cpu
assert cpu.sensor_type == "load"
assert cpu.sensor_unit == "%"
end
test "includes CPU 5min average sensor" do
defs = Adtran.wireless_oid_defs()
cpu_avg = Enum.find(defs, &(&1.sensor_descr == "CPU 5min Average"))
assert cpu_avg
assert cpu_avg.sensor_type == "load"
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 4, fn _, oid, _ ->
cond do
String.contains?(oid, "664.5.53.1.4.1.0") -> {:ok, 25}
String.contains?(oid, "664.5.53.1.4.4.0") -> {:ok, 20}
String.contains?(oid, "664.5.53.1.4.8.0") -> {:ok, 512_000_000}
String.contains?(oid, "1.3.0") -> {:ok, 123_456_789}
true -> {:error, :no_such_object}
end
end)
sensors = Adtran.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 4
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 4, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Adtran.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

View file

@ -0,0 +1,123 @@
defmodule Towerops.Snmp.Profiles.Vendors.AlliedTelesisTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.AlliedTelesis
alias Towerops.Snmp.SnmpMock
setup :verify_on_exit!
@client_opts [
ip: "192.168.1.1",
community: "public",
version: "2c",
port: 161,
timeout: 5000
]
describe "profile_names/0" do
test "returns Allied Telesis profile names" do
assert AlliedTelesis.profile_names() == ["allied", "awplus", "allied-tq"]
end
end
describe "detect_hardware/1" do
test "returns model when AT- prefix found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Allied Telesis AT-x530-28GTXm switch"}
end)
assert AlliedTelesis.detect_hardware(@client_opts) == "AT-x530-28GTXm"
end
test "returns model when x-series found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Allied Telesis x310-26FT switch"}
end)
assert AlliedTelesis.detect_hardware(@client_opts) == "x310-26FT"
end
test "returns AlliedWare Plus when found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "AlliedWare Plus software version 5.4.9"}
end)
assert AlliedTelesis.detect_hardware(@client_opts) == "AlliedWare Plus"
end
test "returns nil when model not in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Some other device"}
end)
assert AlliedTelesis.detect_hardware(@client_opts) == nil
end
test "returns nil when SNMP fails" do
expect(SnmpMock, :get, fn _, _, _ ->
{:error, :timeout}
end)
assert AlliedTelesis.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of sensor definitions" do
defs = AlliedTelesis.wireless_oid_defs()
assert is_list(defs)
assert [_ | _] = defs
end
test "includes CPU 5min average sensor" do
defs = AlliedTelesis.wireless_oid_defs()
cpu = Enum.find(defs, &(&1.sensor_descr == "CPU 5min Average"))
assert cpu
assert cpu.sensor_type == "load"
assert cpu.sensor_unit == "%"
end
test "includes temperature sensor" do
defs = AlliedTelesis.wireless_oid_defs()
temp = Enum.find(defs, &(&1.sensor_descr == "Temperature"))
assert temp
assert temp.sensor_type == "temperature"
assert temp.sensor_unit == "°C"
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 5, fn _, oid, _ ->
cond do
String.contains?(oid, "207.8.4.4.3.3.7.0") -> {:ok, 15}
String.contains?(oid, "207.8.4.4.3.7.1.0") -> {:ok, 1_024_000}
String.contains?(oid, "207.8.4.4.3.12.3.1.5.1") -> {:ok, 42}
String.contains?(oid, "207.8.4.4.3.12.1.1.5.1") -> {:ok, 3500}
String.contains?(oid, "1.3.0") -> {:ok, 987_654_321}
true -> {:error, :no_such_object}
end
end)
sensors = AlliedTelesis.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 5
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 5, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = AlliedTelesis.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

View file

@ -0,0 +1,115 @@
defmodule Towerops.Snmp.Profiles.Vendors.ApcTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Apc
alias Towerops.Snmp.SnmpMock
setup :verify_on_exit!
@client_opts [
ip: "192.168.1.1",
community: "public",
version: "2c",
port: 161,
timeout: 5000
]
describe "profile_names/0" do
test "returns APC profile names" do
assert Apc.profile_names() == ["apc", "apc-cpdu", "apc-epdu", "apc-mgeups", "apc-netbotz"]
end
end
describe "detect_hardware/1" do
test "returns model when SNMP responds" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.318.1.1.1.1.1.1.0", _ ->
{:ok, "Smart-UPS 3000"}
end)
assert Apc.detect_hardware(@client_opts) == "Smart-UPS 3000"
end
test "returns nil when SNMP fails" do
expect(SnmpMock, :get, fn _, _, _ ->
{:error, :timeout}
end)
assert Apc.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of sensor definitions" do
defs = Apc.wireless_oid_defs()
assert is_list(defs)
assert length(defs) > 0
end
test "includes battery capacity sensor" do
defs = Apc.wireless_oid_defs()
battery = Enum.find(defs, &(&1.sensor_descr == "Battery Capacity"))
assert battery
assert battery.sensor_type == "load"
assert battery.sensor_unit == "%"
end
test "includes output voltage sensor" do
defs = Apc.wireless_oid_defs()
voltage = Enum.find(defs, &(&1.sensor_descr == "Output Voltage"))
assert voltage
assert voltage.sensor_type == "voltage"
assert voltage.sensor_unit == "V"
end
test "includes output load sensor" do
defs = Apc.wireless_oid_defs()
load = Enum.find(defs, &(&1.sensor_descr == "Output Load"))
assert load
assert load.sensor_type == "load"
assert load.sensor_unit == "%"
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 9, fn _, oid, _ ->
cond do
String.contains?(oid, "318.1.1.1.2.2.1.0") -> {:ok, 95}
String.contains?(oid, "318.1.1.1.2.2.2.0") -> {:ok, 25}
String.contains?(oid, "318.1.1.1.2.2.3.0") -> {:ok, 36_000}
String.contains?(oid, "318.1.1.1.4.2.1.0") -> {:ok, 120}
String.contains?(oid, "318.1.1.1.4.2.2.0") -> {:ok, 60}
String.contains?(oid, "318.1.1.1.4.2.3.0") -> {:ok, 45}
String.contains?(oid, "318.1.1.1.4.2.4.0") -> {:ok, 5}
String.contains?(oid, "318.1.1.1.3.2.1.0") -> {:ok, 121}
String.contains?(oid, "318.1.1.1.3.2.4.0") -> {:ok, 60}
true -> {:error, :no_such_object}
end
end)
sensors = Apc.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 9
battery = Enum.find(sensors, &(&1.sensor_descr == "Battery Capacity"))
assert battery.last_value == 95.0
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 9, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Apc.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

View file

@ -0,0 +1,97 @@
defmodule Towerops.Snmp.Profiles.Vendors.AristaTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Arista
alias Towerops.Snmp.SnmpMock
setup :verify_on_exit!
@client_opts [
ip: "192.168.1.1",
community: "public",
version: "2c",
port: 161,
timeout: 5000
]
describe "profile_names/0" do
test "returns Arista profile names" do
assert Arista.profile_names() == ["arista_eos", "arista-mos"]
end
end
describe "detect_hardware/1" do
test "returns model when found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Arista Networks EOS version 4.28.1F running on an Arista Networks DCS-7050TX-64"}
end)
assert Arista.detect_hardware(@client_opts) == "DCS-7050TX-64"
end
test "returns nil when model not in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Some other device"}
end)
assert Arista.detect_hardware(@client_opts) == nil
end
test "returns nil when SNMP fails" do
expect(SnmpMock, :get, fn _, _, _ ->
{:error, :timeout}
end)
assert Arista.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of sensor definitions" do
defs = Arista.wireless_oid_defs()
assert is_list(defs)
assert length(defs) > 0
end
test "includes CPU load sensor" do
defs = Arista.wireless_oid_defs()
cpu = Enum.find(defs, &(&1.sensor_descr == "CPU Load"))
assert cpu
assert cpu.sensor_type == "load"
assert cpu.sensor_unit == "%"
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 4, fn _, oid, _ ->
cond do
String.contains?(oid, "25.3.3.1.2.1") -> {:ok, 15}
String.contains?(oid, "25.2.3.1.5.1") -> {:ok, 8_388_608}
String.contains?(oid, "25.2.3.1.6.1") -> {:ok, 4_194_304}
String.contains?(oid, "1.3.0") -> {:ok, 123_456_789}
true -> {:error, :no_such_object}
end
end)
sensors = Arista.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 4
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 4, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Arista.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

View file

@ -0,0 +1,110 @@
defmodule Towerops.Snmp.Profiles.Vendors.CalixTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Calix
alias Towerops.Snmp.SnmpMock
setup :verify_on_exit!
@client_opts [
ip: "192.168.1.1",
community: "public",
version: "2c",
port: 161,
timeout: 5000
]
describe "profile_names/0" do
test "returns Calix profile names" do
assert Calix.profile_names() == ["calix"]
end
end
describe "detect_hardware/1" do
test "returns E7-2 when found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Calix E7-2 AXOS 21.3"}
end)
assert Calix.detect_hardware(@client_opts) == "E7-2"
end
test "returns E7 when found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Calix E7 EXA"}
end)
assert Calix.detect_hardware(@client_opts) == "E7"
end
test "returns AXOS when found in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "AXOS Platform"}
end)
assert Calix.detect_hardware(@client_opts) == "AXOS"
end
test "returns nil when model not in sysDescr" do
expect(SnmpMock, :get, fn _, "1.3.6.1.2.1.1.1.0", _ ->
{:ok, "Some other device"}
end)
assert Calix.detect_hardware(@client_opts) == nil
end
test "returns nil when SNMP fails" do
expect(SnmpMock, :get, fn _, _, _ ->
{:error, :timeout}
end)
assert Calix.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of sensor definitions" do
defs = Calix.wireless_oid_defs()
assert is_list(defs)
assert length(defs) > 0
end
test "includes CPU load sensor" do
defs = Calix.wireless_oid_defs()
cpu = Enum.find(defs, &(&1.sensor_descr == "CPU Load"))
assert cpu
assert cpu.sensor_type == "load"
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 2, fn _, oid, _ ->
cond do
String.contains?(oid, "25.3.3.1.2.1") -> {:ok, 25}
String.contains?(oid, "1.3.0") -> {:ok, 987_654_321}
true -> {:error, :no_such_object}
end
end)
sensors = Calix.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 2
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 2, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Calix.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

View file

@ -3,10 +3,15 @@ defmodule Towerops.Snmp.Profiles.Vendors.RegistryTest do
import Mox
alias Towerops.Snmp.Profiles.Vendors.Adtran
alias Towerops.Snmp.Profiles.Vendors.Airfiber
alias Towerops.Snmp.Profiles.Vendors.Airos
alias Towerops.Snmp.Profiles.Vendors.AlliedTelesis
alias Towerops.Snmp.Profiles.Vendors.Altalabs
alias Towerops.Snmp.Profiles.Vendors.Alvarion
alias Towerops.Snmp.Profiles.Vendors.Apc
alias Towerops.Snmp.Profiles.Vendors.Arista
alias Towerops.Snmp.Profiles.Vendors.Calix
alias Towerops.Snmp.Profiles.Vendors.Ceragon
alias Towerops.Snmp.Profiles.Vendors.Cmm
alias Towerops.Snmp.Profiles.Vendors.Cnpilot
@ -86,6 +91,11 @@ defmodule Towerops.Snmp.Profiles.Vendors.RegistryTest do
assert Tranzeo in vendors
assert Fortiwlc in vendors
assert Meraki in vendors
assert Apc in vendors
assert Arista in vendors
assert Calix in vendors
assert Adtran in vendors
assert AlliedTelesis in vendors
end
end
@ -131,6 +141,12 @@ defmodule Towerops.Snmp.Profiles.Vendors.RegistryTest do
assert "merakims" in names
assert "merakimx" in names
assert "cnpilotr" in names
assert "apc" in names
assert "arista_eos" in names
assert "calix" in names
assert "adtran-aos" in names
assert "allied" in names
assert "awplus" in names
end
end
@ -275,6 +291,30 @@ defmodule Towerops.Snmp.Profiles.Vendors.RegistryTest do
assert Registry.get_vendor("merakimx") == Meraki
end
test "returns Apc for apc profile" do
assert Registry.get_vendor("apc") == Apc
end
test "returns Arista for arista_eos profile" do
assert Registry.get_vendor("arista_eos") == Arista
end
test "returns Calix for calix profile" do
assert Registry.get_vendor("calix") == Calix
end
test "returns Adtran for adtran-aos profile" do
assert Registry.get_vendor("adtran-aos") == Adtran
end
test "returns AlliedTelesis for allied profile" do
assert Registry.get_vendor("allied") == AlliedTelesis
end
test "returns AlliedTelesis for awplus profile" do
assert Registry.get_vendor("awplus") == AlliedTelesis
end
test "returns nil for unknown profile" do
assert Registry.get_vendor("unknown_vendor") == nil
end