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

feat: add vendor-specific SNMP modules for WISP equipment

Browse source 
Add 8 new vendor modules for wireless sensor discovery:
- PMP: Cambium PMP 450/430/100 and PTP series
- Mimosa: B5/B5c PTP and A5/A5c PTMP radios
- Siklu: EtherHaul mmWave radios
- Radwin: WinLink and JET series
- UniFi: Ubiquiti UniFi Access Points
- LigoOS: LigoWave and Deliberant devices
- cnPilot: Cambium cnPilot E and R series
- SAF: SAF Tehnika Integra/Lumina/CFIP series

Each module implements the Vendor behaviour with:
- profile_names/0: OS profile identifiers
- detect_hardware/1: device model detection
- wireless_oid_defs/0: sensor OID definitions
- discover_wireless_sensors/1: sensor discovery

Includes comprehensive tests for all modules (104 tests).
This commit is contained in:
Graham McIntire 2026-01-22 08:48:38 -06:00
parent 56bb855996
commit a83e63b3ad
No known key found for this signature in database
18 changed files with 2107 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

67
lib/towerops/snmp/profiles/vendors/cnpilot.ex vendored Normal file
View file

@ -0,0 +1,67 @@
defmodule Towerops.Snmp.Profiles.Vendors.Cnpilot do
@moduledoc """
Cambium cnPilot device-specific SNMP handling.
Mirrors LibreNMS/OS/Cnpilote.php for wireless sensor discovery.
Supported devices:
- cnPilot E-series (E400, E410, E425, E430, E500, E501, E502, E505, E510, E600, E601)
- cnPilot R-series (R190, R195, R200, R201)
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Profiles.Vendors.Vendor
@impl true
def profile_names, do: ["cnpilot", "cnpilote"]
@impl true
def detect_hardware(_client_opts) do
# cnPilot hardware detection is handled by YAML profile
nil
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
[
# Total Clients (AP mode)
%{
oid: "1.3.6.1.4.1.17713.22.1.1.1.14.0",
sensor_type: "clients",
sensor_descr: "Clients",
sensor_unit: "",
sensor_divisor: 1
},
# SNR (SM mode)
%{
oid: "1.3.6.1.4.1.17713.22.1.3.1.11.0",
sensor_type: "snr",
sensor_descr: "SNR",
sensor_unit: "dB",
sensor_divisor: 1
},
# Transmit Power
%{
oid: "1.3.6.1.4.1.17713.22.1.2.1.8.0",
sensor_type: "power",
sensor_descr: "Transmit Power",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Noise Floor
%{
oid: "1.3.6.1.4.1.17713.22.1.2.1.16.0",
sensor_type: "noise-floor",
sensor_descr: "Radio Noise Floor",
sensor_unit: "dBm",
sensor_divisor: 1
}
]
end
end

101
lib/towerops/snmp/profiles/vendors/ligoos.ex vendored Normal file
View file

@ -0,0 +1,101 @@
defmodule Towerops.Snmp.Profiles.Vendors.Ligoos do
@moduledoc """
LigoWave (LigoOS) device-specific SNMP handling.
Mirrors LibreNMS/OS/Ligoos.php for wireless sensor discovery.
Supported devices:
- LigoWave LigoPTP series
- LigoWave DLB series
- Deliberant devices (share same firmware)
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# Table base OID for wireless interface config
@wl_if_table "1.3.6.1.4.1.32750.3.10.1.2.1.1"
@impl true
def profile_names, do: ["ligoos", "deliberant"]
@impl true
def detect_hardware(_client_opts) do
# Hardware detection is handled by YAML profile
nil
end
@impl true
def discover_wireless_sensors(client_opts) do
# LigoOS uses table-based discovery for multiple interfaces
scalar_sensors = Vendor.fetch_sensors(wireless_oid_defs(), client_opts)
# Also try to walk the wireless interface table for per-interface sensors
table_sensors = discover_table_sensors(client_opts)
scalar_sensors ++ table_sensors
end
defp discover_table_sensors(client_opts) do
# Walk the client count column (index 16)
case Client.walk(client_opts, "#{@wl_if_table}.16") do
{:ok, results} when is_map(results) and map_size(results) > 0 ->
Enum.map(results, fn {oid, value} ->
index = oid |> String.split(".") |> List.last()
%{
sensor_type: "clients",
sensor_index: "ligoos_clients_#{index}",
sensor_oid: oid,
sensor_descr: "Wireless Clients (if #{index})",
sensor_unit: "",
sensor_divisor: 1,
last_value: value / 1
}
end)
_ ->
[]
end
end
@impl true
def wireless_oid_defs do
[
# Frequency (first interface)
%{
oid: "1.3.6.1.4.1.32750.3.10.1.2.1.1.6.1",
sensor_type: "frequency",
sensor_descr: "Frequency",
sensor_unit: "MHz",
sensor_divisor: 1
},
# Noise Level (first interface)
%{
oid: "1.3.6.1.4.1.32750.3.10.1.2.1.1.15.1",
sensor_type: "noise-floor",
sensor_descr: "Noise Floor",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Link Quality (first interface) in %
%{
oid: "1.3.6.1.4.1.32750.3.10.1.2.1.1.12.1",
sensor_type: "quality",
sensor_descr: "Link Quality",
sensor_unit: "%",
sensor_divisor: 1
},
# Associated Node Count (first interface)
%{
oid: "1.3.6.1.4.1.32750.3.10.1.2.1.1.16.1",
sensor_type: "clients",
sensor_descr: "Associated Nodes",
sensor_unit: "",
sensor_divisor: 1
}
]
end
end

147
lib/towerops/snmp/profiles/vendors/mimosa.ex vendored Normal file
View file

@ -0,0 +1,147 @@
defmodule Towerops.Snmp.Profiles.Vendors.Mimosa do
@moduledoc """
Mimosa Networks device-specific SNMP handling.
Mirrors LibreNMS/OS/Mimosa.php for wireless sensor discovery.
Supported devices:
- Mimosa B5/B5c (PTP)
- Mimosa A5/A5c (PTMP)
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Profiles.Vendors.Vendor
@impl true
def profile_names, do: ["mimosa"]
@impl true
def detect_hardware(_client_opts) do
# Mimosa hardware detection is handled by YAML profile
nil
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
[
# Tx Packet Error Ratio (divide by 100 for %)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.7.3.0",
sensor_type: "error-ratio",
sensor_descr: "Tx Packet Error Ratio",
sensor_unit: "%",
sensor_divisor: 100
},
# Rx Packet Error Ratio (divide by 100 for %)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.7.4.0",
sensor_type: "error-ratio",
sensor_descr: "Rx Packet Error Ratio",
sensor_unit: "%",
sensor_divisor: 100
},
# Center Frequency Chain 1 (in MHz)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.6.1",
sensor_type: "frequency",
sensor_descr: "Frequency Chain 1",
sensor_unit: "MHz",
sensor_divisor: 1
},
# Center Frequency Chain 2 (in MHz)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.6.2",
sensor_type: "frequency",
sensor_descr: "Frequency Chain 2",
sensor_unit: "MHz",
sensor_divisor: 1
},
# Tx Power Chain 1 (divide by 10 for dBm)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.2.1",
sensor_type: "power",
sensor_descr: "Tx Power Chain 1",
sensor_unit: "dBm",
sensor_divisor: 10
},
# Tx Power Chain 2 (divide by 10 for dBm)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.2.2",
sensor_type: "power",
sensor_descr: "Tx Power Chain 2",
sensor_unit: "dBm",
sensor_divisor: 10
},
# Rx Power Chain 1 (divide by 10 for dBm)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.3.1",
sensor_type: "power",
sensor_descr: "Rx Power Chain 1",
sensor_unit: "dBm",
sensor_divisor: 10
},
# Rx Power Chain 2 (divide by 10 for dBm)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.3.2",
sensor_type: "power",
sensor_descr: "Rx Power Chain 2",
sensor_unit: "dBm",
sensor_divisor: 10
},
# Rx Noise Chain 1 (divide by 10 for dBm)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.4.1",
sensor_type: "noise-floor",
sensor_descr: "Rx Noise Chain 1",
sensor_unit: "dBm",
sensor_divisor: 10
},
# Rx Noise Chain 2 (divide by 10 for dBm)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.4.2",
sensor_type: "noise-floor",
sensor_descr: "Rx Noise Chain 2",
sensor_unit: "dBm",
sensor_divisor: 10
},
# SNR Chain 1 (divide by 10 for dB)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.5.1",
sensor_type: "snr",
sensor_descr: "SNR Chain 1",
sensor_unit: "dB",
sensor_divisor: 10
},
# SNR Chain 2 (divide by 10 for dB)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.1.1.5.2",
sensor_type: "snr",
sensor_descr: "SNR Chain 2",
sensor_unit: "dB",
sensor_divisor: 10
},
# Tx PHY Rate Stream 1 (in Mbps)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.2.1.2.1",
sensor_type: "rate",
sensor_descr: "Tx Rate Stream 1",
sensor_unit: "Mbps",
sensor_divisor: 1
},
# Rx PHY Rate Stream 1 (in Mbps)
%{
oid: "1.3.6.1.4.1.43356.2.1.2.6.2.1.5.1",
sensor_type: "rate",
sensor_descr: "Rx Rate Stream 1",
sensor_unit: "Mbps",
sensor_divisor: 1
}
]
end
end

198
lib/towerops/snmp/profiles/vendors/pmp.ex vendored Normal file
View file

@ -0,0 +1,198 @@
defmodule Towerops.Snmp.Profiles.Vendors.Pmp do
@moduledoc """
Cambium PMP device-specific SNMP handling.
Mirrors LibreNMS/OS/Pmp.php for wireless sensor discovery and hardware detection.
Supported devices:
- PMP 450 (MIMO OFDM)
- PMP 430 (OFDM)
- PMP 100 (FSK)
- PTP 450, PTP 230, PTP 100 (Backhaul)
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# OIDs for hardware detection
@box_device_type "1.3.6.1.4.1.161.19.3.3.1.1.0"
@bh_timing_mode "1.3.6.1.4.1.161.19.3.3.1.2.0"
@impl true
def profile_names, do: ["pmp"]
@impl true
def detect_hardware(client_opts) do
case Client.get(client_opts, @box_device_type) do
{:ok, box_type} when is_binary(box_type) ->
timing_mode =
case Client.get(client_opts, @bh_timing_mode) do
{:ok, mode} when is_binary(mode) -> mode
_ -> ""
end
determine_hardware(box_type, timing_mode)
_ ->
nil
end
end
defp determine_hardware(box_type, timing_mode) do
ptp_model = detect_ptp_model(box_type, timing_mode)
ptp_model || detect_pmp_model(box_type)
end
defp detect_ptp_model(box_type, timing_mode) do
ptp_models = [
{"BHUL450", "PTP 450"},
{"BHUL", "PTP 230"},
{"BH20", "PTP 100"}
]
Enum.find_value(ptp_models, fn {pattern, model} ->
if String.contains?(box_type, pattern) do
"#{model} #{clean_timing_mode(timing_mode)}"
end
end)
end
defp detect_pmp_model(box_type) do
{base_model, suffix} =
cond do
String.contains?(box_type, "MIMO OFDM") -> {"PMP 450", ap_or_sm(box_type)}
String.contains?(box_type, "OFDM") -> {"PMP 430", ap_or_sm(box_type)}
true -> {"PMP 100", ap_or_sm(box_type)}
end
"#{base_model} #{suffix}"
end
defp ap_or_sm(box_type) do
if String.contains?(box_type, "AP"), do: "AP", else: "SM"
end
defp clean_timing_mode(mode) do
mode
|> String.replace("timing", "")
|> String.replace("timeing", "")
|> String.trim()
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
[
# RSSI
%{
oid: "1.3.6.1.4.1.161.19.3.2.2.2.0",
sensor_type: "rssi",
sensor_descr: "RSSI",
sensor_unit: "dBm",
sensor_divisor: 1
},
# SNR Horizontal (SM)
%{
oid: "1.3.6.1.4.1.161.19.3.2.2.106.0",
sensor_type: "snr",
sensor_descr: "SNR Horizontal",
sensor_unit: "dB",
sensor_divisor: 1
},
# SNR Vertical (SM)
%{
oid: "1.3.6.1.4.1.161.19.3.2.2.95.0",
sensor_type: "snr",
sensor_descr: "SNR Vertical",
sensor_unit: "dB",
sensor_divisor: 1
},
# Frequency (in 10s of KHz for OFDM, divide by 1000)
%{
oid: "1.3.6.1.4.1.161.19.3.1.7.37.0",
sensor_type: "frequency",
sensor_descr: "Frequency",
sensor_unit: "MHz",
sensor_divisor: 1000
},
# Client count (AP only)
%{
oid: "1.3.6.1.4.1.161.19.3.1.7.1.0",
sensor_type: "clients",
sensor_descr: "Registered SMs",
sensor_unit: "",
sensor_divisor: 1
},
# 1m Downlink Utilization
%{
oid: "1.3.6.1.4.1.161.19.3.1.12.1.1.0",
sensor_type: "utilization",
sensor_descr: "1m Downlink Utilization",
sensor_unit: "%",
sensor_divisor: 1
},
# 1m Uplink Utilization
%{
oid: "1.3.6.1.4.1.161.19.3.1.12.1.2.0",
sensor_type: "utilization",
sensor_descr: "1m Uplink Utilization",
sensor_unit: "%",
sensor_divisor: 1
},
# Signal Strength Ratio (SSR)
%{
oid: "1.3.6.1.4.1.161.19.3.2.2.108.0",
sensor_type: "ssr",
sensor_descr: "Signal Strength Ratio",
sensor_unit: "dB",
sensor_divisor: 1
},
# CRC Errors
%{
oid: "1.3.6.1.4.1.161.19.3.3.1.223.0",
sensor_type: "errors",
sensor_descr: "CRC Errors",
sensor_unit: "",
sensor_divisor: 1
},
# FEC In Errors
%{
oid: "1.3.6.1.4.1.161.19.3.3.1.95.0",
sensor_type: "errors",
sensor_descr: "FEC In Errors",
sensor_unit: "",
sensor_divisor: 1
},
# FEC Out Errors
%{
oid: "1.3.6.1.4.1.161.19.3.3.1.97.0",
sensor_type: "errors",
sensor_descr: "FEC Out Errors",
sensor_unit: "",
sensor_divisor: 1
},
# GPS Visible Satellites
%{
oid: "1.3.6.1.4.1.161.19.3.4.4.7.0",
sensor_type: "count",
sensor_descr: "GPS Visible Satellites",
sensor_unit: "",
sensor_divisor: 1
},
# GPS Tracked Satellites
%{
oid: "1.3.6.1.4.1.161.19.3.4.4.8.0",
sensor_type: "count",
sensor_descr: "GPS Tracked Satellites",
sensor_unit: "",
sensor_divisor: 1
}
]
end
end

67
lib/towerops/snmp/profiles/vendors/radwin.ex vendored Normal file
View file

@ -0,0 +1,67 @@
defmodule Towerops.Snmp.Profiles.Vendors.Radwin do
@moduledoc """
Radwin device-specific SNMP handling.
Mirrors LibreNMS/OS/Radwin.php for wireless sensor discovery.
Supported devices:
- Radwin WinLink series
- Radwin JET series
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Profiles.Vendors.Vendor
@impl true
def profile_names, do: ["radwin"]
@impl true
def detect_hardware(_client_opts) do
# Radwin hardware detection is handled by YAML profile
nil
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
[
# Link Distance (in meters, divide by 1000 for km)
%{
oid: "1.3.6.1.4.1.4458.1000.1.5.29.0",
sensor_type: "distance",
sensor_descr: "Link Distance",
sensor_unit: "km",
sensor_divisor: 1000
},
# Tx Power in dBm
%{
oid: "1.3.6.1.4.1.4458.1000.1.5.4.0",
sensor_type: "power",
sensor_descr: "Tx Power",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Rx Power in dBm
%{
oid: "1.3.6.1.4.1.4458.1000.1.5.9.1.0",
sensor_type: "power",
sensor_descr: "Rx Power",
sensor_unit: "dBm",
sensor_divisor: 1
},
# RSSI in dBm
%{
oid: "1.3.6.1.4.1.4458.1000.1.1.51.7.0",
sensor_type: "rssi",
sensor_descr: "RSSI",
sensor_unit: "dBm",
sensor_divisor: 1
}
]
end
end

10
lib/towerops/snmp/profiles/vendors/registry.ex vendored
View file

@ -8,10 +8,18 @@ defmodule Towerops.Snmp.Profiles.Vendors.Registry do
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Airos
alias Towerops.Snmp.Profiles.Vendors.Cnpilot
alias Towerops.Snmp.Profiles.Vendors.Epmp
alias Towerops.Snmp.Profiles.Vendors.Ligoos
alias Towerops.Snmp.Profiles.Vendors.Mimosa
alias Towerops.Snmp.Profiles.Vendors.Pmp
alias Towerops.Snmp.Profiles.Vendors.Radwin
alias Towerops.Snmp.Profiles.Vendors.Routeros
alias Towerops.Snmp.Profiles.Vendors.Saf
alias Towerops.Snmp.Profiles.Vendors.Siklu
alias Towerops.Snmp.Profiles.Vendors.Unifi
@vendors [Epmp, Airos, Routeros]
@vendors [Epmp, Airos, Routeros, Pmp, Mimosa, Siklu, Radwin, Unifi, Ligoos, Cnpilot, Saf]
@doc """
Get the vendor module for a given profile name.

92
lib/towerops/snmp/profiles/vendors/saf.ex vendored Normal file
View file

@ -0,0 +1,92 @@
defmodule Towerops.Snmp.Profiles.Vendors.Saf do
@moduledoc """
SAF Tehnika device-specific SNMP handling.
Mirrors LibreNMS/OS/Saf.php for wireless sensor discovery.
Supported devices:
- SAF Tehnika Integra series
- SAF Tehnika Lumina series
- SAF Tehnika CFIP series
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Profiles.Vendors.Vendor
@impl true
def profile_names, do: ["saf"]
@impl true
def detect_hardware(_client_opts) do
# SAF hardware detection is handled by YAML profile
nil
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
[
# Tx Frequency (in kHz, divide by 1000 for MHz)
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.10.1.9.1",
sensor_type: "frequency",
sensor_descr: "Tx Frequency",
sensor_unit: "MHz",
sensor_divisor: 1000
},
# Rx Frequency (in kHz, divide by 1000 for MHz)
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.10.1.10.1",
sensor_type: "frequency",
sensor_descr: "Rx Frequency",
sensor_unit: "MHz",
sensor_divisor: 1000
},
# Radial MSE (Mean Square Error, divide by 10 for dB)
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.12.1.10.1",
sensor_type: "mse",
sensor_descr: "Radial MSE",
sensor_unit: "dB",
sensor_divisor: 10
},
# Rx Power in dBm
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.10.1.5.1",
sensor_type: "power",
sensor_descr: "Rx Power",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Tx Power in dBm
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.10.1.4.1",
sensor_type: "power",
sensor_descr: "Tx Power",
sensor_unit: "dBm",
sensor_divisor: 1
},
# ACM Capacity (in kbps)
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.12.1.18.1",
sensor_type: "rate",
sensor_descr: "ACM Capacity",
sensor_unit: "kbps",
sensor_divisor: 1
},
# Total Capacity (in kbps)
%{
oid: "1.3.6.1.4.1.7571.100.1.1.5.1.1.1.12.1.6.1",
sensor_type: "rate",
sensor_descr: "Total Capacity",
sensor_unit: "kbps",
sensor_divisor: 1
}
]
end
end

86
lib/towerops/snmp/profiles/vendors/siklu.ex vendored Normal file
View file

@ -0,0 +1,86 @@
defmodule Towerops.Snmp.Profiles.Vendors.Siklu do
@moduledoc """
Siklu Communication device-specific SNMP handling.
Mirrors LibreNMS/OS/Siklu.php for wireless sensor discovery.
Supported devices:
- Siklu EtherHaul mmWave radios
- Siklu MultiHaul series
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# OIDs for hardware detection
@sw_bank1_running "1.3.6.1.4.1.31926.1.1.1.2.0"
@sw_bank1_version "1.3.6.1.4.1.31926.1.1.1.3.0"
@sw_bank2_version "1.3.6.1.4.1.31926.1.1.1.5.0"
@impl true
def profile_names, do: ["siklu"]
@impl true
def detect_hardware(client_opts) do
# Siklu hardware is typically detected from sysDescr
# Version comes from software bank
case Client.get(client_opts, @sw_bank1_running) do
{:ok, "running"} ->
case Client.get(client_opts, @sw_bank1_version) do
{:ok, version} when is_binary(version) -> version
_ -> nil
end
_ ->
case Client.get(client_opts, @sw_bank2_version) do
{:ok, version} when is_binary(version) -> version
_ -> nil
end
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
[
# Operational Frequency (in KHz, divide by 1000 for MHz)
%{
oid: "1.3.6.1.4.1.31926.2.1.1.4.1",
sensor_type: "frequency",
sensor_descr: "Frequency",
sensor_unit: "MHz",
sensor_divisor: 1000
},
# Tx Power in dBm
%{
oid: "1.3.6.1.4.1.31926.2.1.1.42.1",
sensor_type: "power",
sensor_descr: "Tx Power",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Average RSSI in dBm
%{
oid: "1.3.6.1.4.1.31926.2.1.1.19.1",
sensor_type: "rssi",
sensor_descr: "RSSI",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Average CINR (SNR) in dB
%{
oid: "1.3.6.1.4.1.31926.2.1.1.18.1",
sensor_type: "snr",
sensor_descr: "CINR",
sensor_unit: "dB",
sensor_divisor: 1
}
]
end
end

124
lib/towerops/snmp/profiles/vendors/unifi.ex vendored Normal file
View file

@ -0,0 +1,124 @@
defmodule Towerops.Snmp.Profiles.Vendors.Unifi do
@moduledoc """
Ubiquiti UniFi device-specific SNMP handling.
Mirrors LibreNMS/OS/Unifi.php for wireless sensor discovery.
Supported devices:
- UniFi Access Points (UAP, UAP-AC, UAP-HD, etc.)
- UniFi switches with wireless capabilities
"""
@behaviour Towerops.Snmp.Profiles.Vendors.Vendor
alias Towerops.Snmp.Client
alias Towerops.Snmp.Profiles.Vendors.Vendor
# OIDs for hardware detection
@system_model "1.3.6.1.4.1.41112.1.6.3.3.0"
@impl true
def profile_names, do: ["unifi"]
@impl true
def detect_hardware(client_opts) do
case Client.get(client_opts, @system_model) do
{:ok, model} when is_binary(model) and 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
[
# Client count radio 1 (ng = 2.4GHz)
%{
oid: "1.3.6.1.4.1.41112.1.6.1.2.1.8.1",
sensor_type: "clients",
sensor_descr: "Clients (ng)",
sensor_unit: "",
sensor_divisor: 1
},
# Client count radio 2 (na = 5GHz)
%{
oid: "1.3.6.1.4.1.41112.1.6.1.2.1.8.2",
sensor_type: "clients",
sensor_descr: "Clients (na)",
sensor_unit: "",
sensor_divisor: 1
},
# CCQ (divide by 10 for %)
%{
oid: "1.3.6.1.4.1.41112.1.6.1.2.1.3.1",
sensor_type: "ccq",
sensor_descr: "CCQ Radio 1",
sensor_unit: "%",
sensor_divisor: 10
},
# CCQ Radio 2
%{
oid: "1.3.6.1.4.1.41112.1.6.1.2.1.3.2",
sensor_type: "ccq",
sensor_descr: "CCQ Radio 2",
sensor_unit: "%",
sensor_divisor: 10
},
# Tx Power Radio 1 in dBm
%{
oid: "1.3.6.1.4.1.41112.1.6.1.2.1.21.1",
sensor_type: "power",
sensor_descr: "Tx Power Radio 1",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Tx Power Radio 2 in dBm
%{
oid: "1.3.6.1.4.1.41112.1.6.1.2.1.21.2",
sensor_type: "power",
sensor_descr: "Tx Power Radio 2",
sensor_unit: "dBm",
sensor_divisor: 1
},
# Total Channel Utilization Radio 1
%{
oid: "1.3.6.1.4.1.41112.1.6.1.1.1.6.1",
sensor_type: "utilization",
sensor_descr: "Total Utilization Radio 1",
sensor_unit: "%",
sensor_divisor: 1
},
# Total Channel Utilization Radio 2
%{
oid: "1.3.6.1.4.1.41112.1.6.1.1.1.6.2",
sensor_type: "utilization",
sensor_descr: "Total Utilization Radio 2",
sensor_unit: "%",
sensor_divisor: 1
},
# Self Rx Utilization Radio 1
%{
oid: "1.3.6.1.4.1.41112.1.6.1.1.1.7.1",
sensor_type: "utilization",
sensor_descr: "Self Rx Utilization Radio 1",
sensor_unit: "%",
sensor_divisor: 1
},
# Self Tx Utilization Radio 1
%{
oid: "1.3.6.1.4.1.41112.1.6.1.1.1.8.1",
sensor_type: "utilization",
sensor_descr: "Self Tx Utilization Radio 1",
sensor_unit: "%",
sensor_divisor: 1
}
]
end
end

112
test/towerops/snmp/profiles/vendors/cnpilot_test.exs vendored Normal file
View file

@ -0,0 +1,112 @@
defmodule Towerops.Snmp.Profiles.Vendors.CnpilotTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Cnpilot
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 cnpilot and cnpilote profile names" do
assert Cnpilot.profile_names() == ["cnpilot", "cnpilote"]
end
end
describe "detect_hardware/1" do
test "returns nil (hardware detection handled by YAML)" do
assert Cnpilot.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Cnpilot.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes clients sensor" do
defs = Cnpilot.wireless_oid_defs()
clients = Enum.find(defs, &(&1.sensor_type == "clients"))
assert clients
assert clients.sensor_descr == "Clients"
end
test "includes SNR sensor" do
defs = Cnpilot.wireless_oid_defs()
snr = Enum.find(defs, &(&1.sensor_type == "snr"))
assert snr
assert snr.sensor_unit == "dB"
end
test "includes power sensor" do
defs = Cnpilot.wireless_oid_defs()
power = Enum.find(defs, &(&1.sensor_type == "power"))
assert power
assert power.sensor_descr == "Transmit Power"
assert power.sensor_unit == "dBm"
end
test "includes noise floor sensor" do
defs = Cnpilot.wireless_oid_defs()
noise = Enum.find(defs, &(&1.sensor_type == "noise-floor"))
assert noise
assert noise.sensor_descr == "Radio Noise Floor"
assert noise.sensor_unit == "dBm"
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, "17713.22.1.1.1.14.0") -> {:ok, {:integer, 25}}
String.contains?(oid, "17713.22.1.3.1.11.0") -> {:ok, {:integer, 30}}
String.contains?(oid, "17713.22.1.2.1.8.0") -> {:ok, {:integer, 20}}
String.contains?(oid, "17713.22.1.2.1.16.0") -> {:ok, {:integer, -95}}
true -> {:error, :no_such_object}
end
end)
sensors = Cnpilot.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 4
# Check clients sensor
clients = Enum.find(sensors, &(&1.sensor_descr == "Clients"))
assert clients
assert clients.last_value == 25.0
# Check noise floor sensor
noise = Enum.find(sensors, &(&1.sensor_descr == "Radio Noise Floor"))
assert noise
assert noise.last_value == -95.0
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 4, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Cnpilot.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

139
test/towerops/snmp/profiles/vendors/ligoos_test.exs vendored Normal file
View file

@ -0,0 +1,139 @@
defmodule Towerops.Snmp.Profiles.Vendors.LigoosTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Ligoos
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 ligoos and deliberant profile names" do
assert Ligoos.profile_names() == ["ligoos", "deliberant"]
end
end
describe "detect_hardware/1" do
test "returns nil (hardware detection handled by YAML)" do
assert Ligoos.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Ligoos.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes frequency sensor" do
defs = Ligoos.wireless_oid_defs()
freq = Enum.find(defs, &(&1.sensor_type == "frequency"))
assert freq
assert freq.sensor_unit == "MHz"
end
test "includes noise floor sensor" do
defs = Ligoos.wireless_oid_defs()
noise = Enum.find(defs, &(&1.sensor_type == "noise-floor"))
assert noise
assert noise.sensor_unit == "dBm"
end
test "includes quality sensor" do
defs = Ligoos.wireless_oid_defs()
quality = Enum.find(defs, &(&1.sensor_type == "quality"))
assert quality
assert quality.sensor_unit == "%"
end
test "includes clients sensor" do
defs = Ligoos.wireless_oid_defs()
clients = Enum.find(defs, &(&1.sensor_type == "clients"))
assert clients
end
end
describe "discover_wireless_sensors/1" do
test "discovers scalar sensors when SNMP responds" do
# Mock scalar sensors
expect(SnmpMock, :get, 4, fn _, oid, _ ->
cond do
String.contains?(oid, "32750.3.10.1.2.1.1.6.1") -> {:ok, {:integer, 5800}}
String.contains?(oid, "32750.3.10.1.2.1.1.15.1") -> {:ok, {:integer, -95}}
String.contains?(oid, "32750.3.10.1.2.1.1.12.1") -> {:ok, {:integer, 85}}
String.contains?(oid, "32750.3.10.1.2.1.1.16.1") -> {:ok, {:integer, 10}}
true -> {:error, :no_such_object}
end
end)
# Mock table walk for per-interface clients
expect(SnmpMock, :walk, fn _, "1.3.6.1.4.1.32750.3.10.1.2.1.1.16", _ ->
{:ok, []}
end)
sensors = Ligoos.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 4
end
test "discovers table sensors for multiple interfaces" do
# Mock scalar sensors
expect(SnmpMock, :get, 4, fn _, _, _ ->
{:error, :no_such_object}
end)
# Mock table walk with multiple interfaces - values are raw (extracted by Client)
expect(SnmpMock, :walk, fn _, "1.3.6.1.4.1.32750.3.10.1.2.1.1.16", _ ->
{:ok,
[
%{oid: "1.3.6.1.4.1.32750.3.10.1.2.1.1.16.1", value: 15},
%{oid: "1.3.6.1.4.1.32750.3.10.1.2.1.1.16.2", value: 8}
]}
end)
sensors = Ligoos.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 2
# Check interface-specific sensors
if1 = Enum.find(sensors, &(&1.sensor_index == "ligoos_clients_1"))
assert if1
assert if1.last_value == 15.0
if2 = Enum.find(sensors, &(&1.sensor_index == "ligoos_clients_2"))
assert if2
assert if2.last_value == 8.0
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 4, fn _, _, _ ->
{:error, :no_such_object}
end)
expect(SnmpMock, :walk, fn _, _, _ ->
{:ok, []}
end)
sensors = Ligoos.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

108
test/towerops/snmp/profiles/vendors/mimosa_test.exs vendored Normal file
View file

@ -0,0 +1,108 @@
defmodule Towerops.Snmp.Profiles.Vendors.MimosaTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Mimosa
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 mimosa profile name" do
assert Mimosa.profile_names() == ["mimosa"]
end
end
describe "detect_hardware/1" do
test "returns nil (hardware detection handled by YAML)" do
assert Mimosa.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Mimosa.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes error-ratio sensors" do
defs = Mimosa.wireless_oid_defs()
error_sensors = Enum.filter(defs, &(&1.sensor_type == "error-ratio"))
assert length(error_sensors) == 2
end
test "includes frequency sensors" do
defs = Mimosa.wireless_oid_defs()
freq_sensors = Enum.filter(defs, &(&1.sensor_type == "frequency"))
assert length(freq_sensors) == 2
end
test "includes power sensors with correct divisor" do
defs = Mimosa.wireless_oid_defs()
power_sensors = Enum.filter(defs, &(&1.sensor_type == "power"))
assert length(power_sensors) >= 4
# Mimosa power values are divided by 10
assert Enum.all?(power_sensors, &(&1.sensor_divisor == 10))
end
test "includes SNR sensors" do
defs = Mimosa.wireless_oid_defs()
snr_sensors = Enum.filter(defs, &(&1.sensor_type == "snr"))
assert length(snr_sensors) == 2
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 14, fn _, oid, _ ->
cond do
String.contains?(oid, "43356.2.1.2.7.3.0") -> {:ok, {:integer, 50}}
String.contains?(oid, "43356.2.1.2.7.4.0") -> {:ok, {:integer, 30}}
String.contains?(oid, "43356.2.1.2.6.1.1.6.1") -> {:ok, {:integer, 5800}}
String.contains?(oid, "43356.2.1.2.6.1.1.6.2") -> {:ok, {:integer, 5805}}
String.contains?(oid, "43356.2.1.2.6.1.1.2.1") -> {:ok, {:integer, 200}}
String.contains?(oid, "43356.2.1.2.6.1.1.2.2") -> {:ok, {:integer, 200}}
String.contains?(oid, "43356.2.1.2.6.1.1.3.1") -> {:ok, {:integer, -500}}
String.contains?(oid, "43356.2.1.2.6.1.1.3.2") -> {:ok, {:integer, -510}}
String.contains?(oid, "43356.2.1.2.6.1.1.4.1") -> {:ok, {:integer, -950}}
String.contains?(oid, "43356.2.1.2.6.1.1.4.2") -> {:ok, {:integer, -960}}
String.contains?(oid, "43356.2.1.2.6.1.1.5.1") -> {:ok, {:integer, 250}}
String.contains?(oid, "43356.2.1.2.6.1.1.5.2") -> {:ok, {:integer, 240}}
String.contains?(oid, "43356.2.1.2.6.2.1.2.1") -> {:ok, {:integer, 800}}
String.contains?(oid, "43356.2.1.2.6.2.1.5.1") -> {:ok, {:integer, 750}}
true -> {:error, :no_such_object}
end
end)
sensors = Mimosa.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 14
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 14, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Mimosa.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

205
test/towerops/snmp/profiles/vendors/pmp_test.exs vendored Normal file
View file

@ -0,0 +1,205 @@
defmodule Towerops.Snmp.Profiles.Vendors.PmpTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Pmp
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 pmp profile name" do
assert Pmp.profile_names() == ["pmp"]
end
end
describe "detect_hardware/1" do
test "detects PTP 450 model" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:ok, {:octet_string, "BHUL450 Connectorized"}}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.2.0", _ ->
{:ok, {:octet_string, "generate timing"}}
end)
assert Pmp.detect_hardware(@client_opts) == "PTP 450 generate"
end
test "detects PTP 230 model" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:ok, {:octet_string, "BHUL Integrated"}}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.2.0", _ ->
{:ok, {:octet_string, "receive timing"}}
end)
assert Pmp.detect_hardware(@client_opts) == "PTP 230 receive"
end
test "detects PMP 450 AP model" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:ok, {:octet_string, "MIMO OFDM AP"}}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.2.0", _ ->
{:error, :no_such_object}
end)
assert Pmp.detect_hardware(@client_opts) == "PMP 450 AP"
end
test "detects PMP 450 SM model" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:ok, {:octet_string, "MIMO OFDM SM"}}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.2.0", _ ->
{:error, :no_such_object}
end)
assert Pmp.detect_hardware(@client_opts) == "PMP 450 SM"
end
test "detects PMP 430 model" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:ok, {:octet_string, "OFDM AP"}}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.2.0", _ ->
{:error, :no_such_object}
end)
assert Pmp.detect_hardware(@client_opts) == "PMP 430 AP"
end
test "detects PMP 100 model" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:ok, {:octet_string, "FSK SM"}}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.2.0", _ ->
{:error, :no_such_object}
end)
assert Pmp.detect_hardware(@client_opts) == "PMP 100 SM"
end
test "returns nil when box type unavailable" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.161.19.3.3.1.1.0", _ ->
{:error, :no_such_object}
end)
assert Pmp.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Pmp.wireless_oid_defs()
assert is_list(defs)
assert defs != []
# Verify structure of first definition
first = hd(defs)
assert Map.has_key?(first, :oid)
assert Map.has_key?(first, :sensor_type)
assert Map.has_key?(first, :sensor_descr)
assert Map.has_key?(first, :sensor_unit)
assert Map.has_key?(first, :sensor_divisor)
end
test "includes RSSI sensor" do
defs = Pmp.wireless_oid_defs()
rssi = Enum.find(defs, &(&1.sensor_type == "rssi"))
assert rssi
assert rssi.sensor_unit == "dBm"
end
test "includes SNR sensors" do
defs = Pmp.wireless_oid_defs()
snr_sensors = Enum.filter(defs, &(&1.sensor_type == "snr"))
assert length(snr_sensors) >= 2
end
test "includes frequency sensor" do
defs = Pmp.wireless_oid_defs()
freq = Enum.find(defs, &(&1.sensor_type == "frequency"))
assert freq
assert freq.sensor_unit == "MHz"
assert freq.sensor_divisor == 1000
end
test "includes clients sensor" do
defs = Pmp.wireless_oid_defs()
clients = Enum.find(defs, &(&1.sensor_type == "clients"))
assert clients
end
test "includes utilization sensors" do
defs = Pmp.wireless_oid_defs()
util_sensors = Enum.filter(defs, &(&1.sensor_type == "utilization"))
assert length(util_sensors) >= 2
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
# Mock all sensor OIDs
expect(SnmpMock, :get, 13, fn _, oid, _ ->
cond do
String.contains?(oid, "161.19.3.2.2.2.0") -> {:ok, {:integer, -65}}
String.contains?(oid, "161.19.3.2.2.106.0") -> {:ok, {:integer, 25}}
String.contains?(oid, "161.19.3.2.2.95.0") -> {:ok, {:integer, 23}}
String.contains?(oid, "161.19.3.1.7.37.0") -> {:ok, {:integer, 5_800_000}}
String.contains?(oid, "161.19.3.1.7.1.0") -> {:ok, {:integer, 15}}
String.contains?(oid, "161.19.3.1.12.1.1.0") -> {:ok, {:integer, 45}}
String.contains?(oid, "161.19.3.1.12.1.2.0") -> {:ok, {:integer, 30}}
String.contains?(oid, "161.19.3.2.2.108.0") -> {:ok, {:integer, 10}}
String.contains?(oid, "161.19.3.3.1.223.0") -> {:ok, {:integer, 5}}
String.contains?(oid, "161.19.3.3.1.95.0") -> {:ok, {:integer, 2}}
String.contains?(oid, "161.19.3.3.1.97.0") -> {:ok, {:integer, 1}}
String.contains?(oid, "161.19.3.4.4.7.0") -> {:ok, {:integer, 12}}
String.contains?(oid, "161.19.3.4.4.8.0") -> {:ok, {:integer, 8}}
true -> {:error, :no_such_object}
end
end)
sensors = Pmp.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 13
# Check RSSI sensor
rssi = Enum.find(sensors, &(&1.sensor_descr == "RSSI"))
assert rssi
assert rssi.last_value == -65.0
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 13, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Pmp.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

97
test/towerops/snmp/profiles/vendors/radwin_test.exs vendored Normal file
View file

@ -0,0 +1,97 @@
defmodule Towerops.Snmp.Profiles.Vendors.RadwinTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Radwin
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 radwin profile name" do
assert Radwin.profile_names() == ["radwin"]
end
end
describe "detect_hardware/1" do
test "returns nil (hardware detection handled by YAML)" do
assert Radwin.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Radwin.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes distance sensor with km unit" do
defs = Radwin.wireless_oid_defs()
distance = Enum.find(defs, &(&1.sensor_type == "distance"))
assert distance
assert distance.sensor_unit == "km"
assert distance.sensor_divisor == 1000
end
test "includes power sensors" do
defs = Radwin.wireless_oid_defs()
power_sensors = Enum.filter(defs, &(&1.sensor_type == "power"))
assert length(power_sensors) == 2
end
test "includes RSSI sensor" do
defs = Radwin.wireless_oid_defs()
rssi = Enum.find(defs, &(&1.sensor_type == "rssi"))
assert rssi
assert rssi.sensor_unit == "dBm"
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, "4458.1000.1.5.29.0") -> {:ok, {:integer, 5200}}
String.contains?(oid, "4458.1000.1.5.4.0") -> {:ok, {:integer, 20}}
String.contains?(oid, "4458.1000.1.5.9.1.0") -> {:ok, {:integer, -50}}
String.contains?(oid, "4458.1000.1.1.51.7.0") -> {:ok, {:integer, -55}}
true -> {:error, :no_such_object}
end
end)
sensors = Radwin.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 4
# Check distance is in km
distance = Enum.find(sensors, &(&1.sensor_descr == "Link Distance"))
assert distance
assert distance.last_value == 5200.0
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 4, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Radwin.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

173
test/towerops/snmp/profiles/vendors/registry_test.exs vendored Normal file
View file

@ -0,0 +1,173 @@
defmodule Towerops.Snmp.Profiles.Vendors.RegistryTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Airos
alias Towerops.Snmp.Profiles.Vendors.Cnpilot
alias Towerops.Snmp.Profiles.Vendors.Epmp
alias Towerops.Snmp.Profiles.Vendors.Ligoos
alias Towerops.Snmp.Profiles.Vendors.Mimosa
alias Towerops.Snmp.Profiles.Vendors.Pmp
alias Towerops.Snmp.Profiles.Vendors.Radwin
alias Towerops.Snmp.Profiles.Vendors.Registry
alias Towerops.Snmp.Profiles.Vendors.Routeros
alias Towerops.Snmp.Profiles.Vendors.Saf
alias Towerops.Snmp.Profiles.Vendors.Siklu
alias Towerops.Snmp.Profiles.Vendors.Unifi
alias Towerops.Snmp.SnmpMock
setup :verify_on_exit!
@client_opts [
ip: "192.168.1.1",
community: "public",
version: "2c",
port: 161,
timeout: 5000
]
describe "list_vendors/0" do
test "returns all registered vendor modules" do
vendors = Registry.list_vendors()
assert is_list(vendors)
assert Epmp in vendors
assert Airos in vendors
assert Routeros in vendors
assert Pmp in vendors
assert Mimosa in vendors
assert Siklu in vendors
assert Radwin in vendors
assert Unifi in vendors
assert Ligoos in vendors
assert Cnpilot in vendors
assert Saf in vendors
end
end
describe "list_profile_names/0" do
test "returns all profile names from all vendors" do
names = Registry.list_profile_names()
assert is_list(names)
# Check for some known profile names
assert "epmp" in names
assert "airos" in names
assert "routeros" in names
assert "pmp" in names
assert "mimosa" in names
assert "siklu" in names
assert "radwin" in names
assert "unifi" in names
assert "ligoos" in names
assert "deliberant" in names
assert "cnpilot" in names
assert "saf" in names
end
end
describe "get_vendor/1" do
test "returns Epmp for epmp profile" do
assert Registry.get_vendor("epmp") == Epmp
end
test "returns Airos for airos profile" do
assert Registry.get_vendor("airos") == Airos
end
test "returns Routeros for routeros profile" do
assert Registry.get_vendor("routeros") == Routeros
end
test "returns Pmp for pmp profile" do
assert Registry.get_vendor("pmp") == Pmp
end
test "returns Mimosa for mimosa profile" do
assert Registry.get_vendor("mimosa") == Mimosa
end
test "returns Siklu for siklu profile" do
assert Registry.get_vendor("siklu") == Siklu
end
test "returns Radwin for radwin profile" do
assert Registry.get_vendor("radwin") == Radwin
end
test "returns Unifi for unifi profile" do
assert Registry.get_vendor("unifi") == Unifi
end
test "returns Ligoos for ligoos profile" do
assert Registry.get_vendor("ligoos") == Ligoos
end
test "returns Ligoos for deliberant profile" do
assert Registry.get_vendor("deliberant") == Ligoos
end
test "returns Cnpilot for cnpilot profile" do
assert Registry.get_vendor("cnpilot") == Cnpilot
end
test "returns Saf for saf profile" do
assert Registry.get_vendor("saf") == Saf
end
test "returns nil for unknown profile" do
assert Registry.get_vendor("unknown_vendor") == nil
end
end
describe "detect_hardware/2" do
test "delegates to vendor module" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.41112.1.6.3.3.0", _ ->
{:ok, {:octet_string, "UAP-AC-HD"}}
end)
assert Registry.detect_hardware("unifi", @client_opts) == "UAP-AC-HD"
end
test "returns nil for unknown profile" do
assert Registry.detect_hardware("unknown", @client_opts) == nil
end
end
describe "discover_wireless_sensors/2" do
test "delegates to vendor module" do
expect(SnmpMock, :get, 4, fn _, oid, _ ->
cond do
String.contains?(oid, "17713.22.1.1.1.14.0") -> {:ok, {:integer, 25}}
String.contains?(oid, "17713.22.1.3.1.11.0") -> {:ok, {:integer, 30}}
String.contains?(oid, "17713.22.1.2.1.8.0") -> {:ok, {:integer, 20}}
String.contains?(oid, "17713.22.1.2.1.16.0") -> {:ok, {:integer, -95}}
true -> {:error, :no_such_object}
end
end)
sensors = Registry.discover_wireless_sensors("cnpilot", @client_opts)
assert is_list(sensors)
assert length(sensors) == 4
end
test "returns empty list for unknown profile" do
assert Registry.discover_wireless_sensors("unknown", @client_opts) == []
end
end
describe "get_wireless_oid_defs/1" do
test "returns OID definitions for known profile" do
defs = Registry.get_wireless_oid_defs("pmp")
assert is_list(defs)
assert defs != []
end
test "returns empty list for unknown profile" do
assert Registry.get_wireless_oid_defs("unknown") == []
end
end
end

115
test/towerops/snmp/profiles/vendors/saf_test.exs vendored Normal file
View file

@ -0,0 +1,115 @@
defmodule Towerops.Snmp.Profiles.Vendors.SafTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Saf
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 saf profile name" do
assert Saf.profile_names() == ["saf"]
end
end
describe "detect_hardware/1" do
test "returns nil (hardware detection handled by YAML)" do
assert Saf.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Saf.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes frequency sensors with correct divisor" do
defs = Saf.wireless_oid_defs()
freq_sensors = Enum.filter(defs, &(&1.sensor_type == "frequency"))
assert length(freq_sensors) == 2
# SAF frequency is in kHz, divide by 1000 for MHz
assert Enum.all?(freq_sensors, &(&1.sensor_divisor == 1000))
end
test "includes MSE sensor with correct divisor" do
defs = Saf.wireless_oid_defs()
mse = Enum.find(defs, &(&1.sensor_type == "mse"))
assert mse
assert mse.sensor_descr == "Radial MSE"
# MSE is divided by 10
assert mse.sensor_divisor == 10
end
test "includes power sensors" do
defs = Saf.wireless_oid_defs()
power_sensors = Enum.filter(defs, &(&1.sensor_type == "power"))
assert length(power_sensors) == 2
# Rx and Tx power
assert Enum.any?(power_sensors, &String.contains?(&1.sensor_descr, "Rx"))
assert Enum.any?(power_sensors, &String.contains?(&1.sensor_descr, "Tx"))
end
test "includes rate sensors" do
defs = Saf.wireless_oid_defs()
rate_sensors = Enum.filter(defs, &(&1.sensor_type == "rate"))
assert length(rate_sensors) == 2
# ACM and Total capacity
assert Enum.any?(rate_sensors, &String.contains?(&1.sensor_descr, "ACM"))
assert Enum.any?(rate_sensors, &String.contains?(&1.sensor_descr, "Total"))
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 7, fn _, oid, _ ->
cond do
String.contains?(oid, "7571.100.1.1.5.1.1.1.10.1.9.1") -> {:ok, {:integer, 71_000_000}}
String.contains?(oid, "7571.100.1.1.5.1.1.1.10.1.10.1") -> {:ok, {:integer, 81_000_000}}
String.contains?(oid, "7571.100.1.1.5.1.1.1.12.1.10.1") -> {:ok, {:integer, -150}}
String.contains?(oid, "7571.100.1.1.5.1.1.1.10.1.5.1") -> {:ok, {:integer, -45}}
String.contains?(oid, "7571.100.1.1.5.1.1.1.10.1.4.1") -> {:ok, {:integer, 20}}
String.contains?(oid, "7571.100.1.1.5.1.1.1.12.1.18.1") -> {:ok, {:integer, 500_000}}
String.contains?(oid, "7571.100.1.1.5.1.1.1.12.1.6.1") -> {:ok, {:integer, 1_000_000}}
true -> {:error, :no_such_object}
end
end)
sensors = Saf.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 7
# Check Tx frequency
tx_freq = Enum.find(sensors, &(&1.sensor_descr == "Tx Frequency"))
assert tx_freq
assert tx_freq.last_value == 71_000_000.0
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 7, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Saf.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

140
test/towerops/snmp/profiles/vendors/siklu_test.exs vendored Normal file
View file

@ -0,0 +1,140 @@
defmodule Towerops.Snmp.Profiles.Vendors.SikluTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Siklu
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 siklu profile name" do
assert Siklu.profile_names() == ["siklu"]
end
end
describe "detect_hardware/1" do
test "detects EtherHaul model from software bank 1 version when running" do
# First checks if bank 1 is running
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.31926.1.1.1.2.0", _ ->
{:ok, {:octet_string, "running"}}
end)
# Then gets bank 1 version
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.31926.1.1.1.3.0", _ ->
{:ok, {:octet_string, "EH-1200F v9.0.1"}}
end)
assert Siklu.detect_hardware(@client_opts) == "EH-1200F v9.0.1"
end
test "falls back to bank 2 version when bank 1 not running" do
# Bank 1 is not running
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.31926.1.1.1.2.0", _ ->
{:ok, {:octet_string, "not_running"}}
end)
# Gets bank 2 version
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.31926.1.1.1.5.0", _ ->
{:ok, {:octet_string, "EH-600T v8.5.0"}}
end)
assert Siklu.detect_hardware(@client_opts) == "EH-600T v8.5.0"
end
test "returns nil when OIDs unavailable" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.31926.1.1.1.2.0", _ ->
{:error, :no_such_object}
end)
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.31926.1.1.1.5.0", _ ->
{:error, :no_such_object}
end)
assert Siklu.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Siklu.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes frequency sensor" do
defs = Siklu.wireless_oid_defs()
freq = Enum.find(defs, &(&1.sensor_type == "frequency"))
assert freq
assert freq.sensor_unit == "MHz"
end
test "includes power sensor" do
defs = Siklu.wireless_oid_defs()
power = Enum.find(defs, &(&1.sensor_type == "power"))
assert power
assert power.sensor_descr == "Tx Power"
end
test "includes RSSI sensor" do
defs = Siklu.wireless_oid_defs()
rssi = Enum.find(defs, &(&1.sensor_type == "rssi"))
assert rssi
assert rssi.sensor_unit == "dBm"
end
test "includes CINR sensor" do
defs = Siklu.wireless_oid_defs()
cinr = Enum.find(defs, &(&1.sensor_descr == "CINR"))
assert cinr
assert cinr.sensor_type == "snr"
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 4, fn _, oid, _ ->
cond do
# Frequency OID
String.contains?(oid, "31926.2.1.1.4.1") -> {:ok, {:integer, 71_000}}
# Tx Power OID
String.contains?(oid, "31926.2.1.1.42.1") -> {:ok, {:integer, 18}}
# RSSI OID
String.contains?(oid, "31926.2.1.1.19.1") -> {:ok, {:integer, -45}}
# CINR (SNR) OID
String.contains?(oid, "31926.2.1.1.18.1") -> {:ok, {:integer, 25}}
true -> {:error, :no_such_object}
end
end)
sensors = Siklu.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 = Siklu.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end

127
test/towerops/snmp/profiles/vendors/unifi_test.exs vendored Normal file
View file

@ -0,0 +1,127 @@
defmodule Towerops.Snmp.Profiles.Vendors.UnifiTest do
use Towerops.DataCase, async: true
import Mox
alias Towerops.Snmp.Profiles.Vendors.Unifi
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 unifi profile name" do
assert Unifi.profile_names() == ["unifi"]
end
end
describe "detect_hardware/1" do
test "detects UniFi model from system model OID" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.41112.1.6.3.3.0", _ ->
{:ok, {:octet_string, "UAP-AC-HD"}}
end)
assert Unifi.detect_hardware(@client_opts) == "UAP-AC-HD"
end
test "returns nil when model unavailable" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.41112.1.6.3.3.0", _ ->
{:error, :no_such_object}
end)
assert Unifi.detect_hardware(@client_opts) == nil
end
test "returns nil when model is empty string" do
expect(SnmpMock, :get, fn _, "1.3.6.1.4.1.41112.1.6.3.3.0", _ ->
{:ok, {:octet_string, ""}}
end)
assert Unifi.detect_hardware(@client_opts) == nil
end
end
describe "wireless_oid_defs/0" do
test "returns list of wireless sensor definitions" do
defs = Unifi.wireless_oid_defs()
assert is_list(defs)
assert defs != []
end
test "includes client count sensors for both radios" do
defs = Unifi.wireless_oid_defs()
client_sensors = Enum.filter(defs, &(&1.sensor_type == "clients"))
assert length(client_sensors) == 2
# ng = 2.4GHz, na = 5GHz
assert Enum.any?(client_sensors, &String.contains?(&1.sensor_descr, "ng"))
assert Enum.any?(client_sensors, &String.contains?(&1.sensor_descr, "na"))
end
test "includes CCQ sensors with correct divisor" do
defs = Unifi.wireless_oid_defs()
ccq_sensors = Enum.filter(defs, &(&1.sensor_type == "ccq"))
assert length(ccq_sensors) == 2
# CCQ values are divided by 10 for percentage
assert Enum.all?(ccq_sensors, &(&1.sensor_divisor == 10))
end
test "includes power sensors" do
defs = Unifi.wireless_oid_defs()
power_sensors = Enum.filter(defs, &(&1.sensor_type == "power"))
assert length(power_sensors) == 2
end
test "includes utilization sensors" do
defs = Unifi.wireless_oid_defs()
util_sensors = Enum.filter(defs, &(&1.sensor_type == "utilization"))
assert length(util_sensors) >= 2
end
end
describe "discover_wireless_sensors/1" do
test "discovers sensors when SNMP responds" do
expect(SnmpMock, :get, 10, fn _, oid, _ ->
cond do
String.contains?(oid, "41112.1.6.1.2.1.8.1") -> {:ok, {:integer, 25}}
String.contains?(oid, "41112.1.6.1.2.1.8.2") -> {:ok, {:integer, 15}}
String.contains?(oid, "41112.1.6.1.2.1.3.1") -> {:ok, {:integer, 980}}
String.contains?(oid, "41112.1.6.1.2.1.3.2") -> {:ok, {:integer, 950}}
String.contains?(oid, "41112.1.6.1.2.1.21.1") -> {:ok, {:integer, 20}}
String.contains?(oid, "41112.1.6.1.2.1.21.2") -> {:ok, {:integer, 23}}
String.contains?(oid, "41112.1.6.1.1.1.6.1") -> {:ok, {:integer, 45}}
String.contains?(oid, "41112.1.6.1.1.1.6.2") -> {:ok, {:integer, 30}}
String.contains?(oid, "41112.1.6.1.1.1.7.1") -> {:ok, {:integer, 15}}
String.contains?(oid, "41112.1.6.1.1.1.8.1") -> {:ok, {:integer, 10}}
true -> {:error, :no_such_object}
end
end)
sensors = Unifi.discover_wireless_sensors(@client_opts)
assert is_list(sensors)
assert length(sensors) == 10
end
test "returns empty list when no sensors respond" do
expect(SnmpMock, :get, 10, fn _, _, _ ->
{:error, :no_such_object}
end)
sensors = Unifi.discover_wireless_sensors(@client_opts)
assert sensors == []
end
end
end