more tests and fixes
This commit is contained in:
parent
dff9c26905
commit
021c86a130
12 changed files with 926 additions and 57 deletions
|
|
@ -437,10 +437,10 @@ defmodule SnmpKit.SnmpLib.ManagerTest do
|
|||
|
||||
# This verifies the logic works but we can't test the private function directly
|
||||
# Instead verify that SNMPv2c exception types are recognized
|
||||
assert Types.is_exception_type?(:no_such_object) == true
|
||||
assert Types.is_exception_type?(:no_such_instance) == true
|
||||
assert Types.is_exception_type?(:end_of_mib_view) == true
|
||||
assert Types.is_exception_type?(:integer) == false
|
||||
assert Types.exception_type?(:no_such_object) == true
|
||||
assert Types.exception_type?(:no_such_instance) == true
|
||||
assert Types.exception_type?(:end_of_mib_view) == true
|
||||
assert Types.exception_type?(:integer) == false
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ defmodule SnmpKit.SnmpLib.MIB.ParserTest do
|
|||
|
||||
alias SnmpKit.SnmpLib.MIB.Parser
|
||||
|
||||
@tag :yecc_required
|
||||
test "parser has known bug with index 10 (newline character)" do
|
||||
mib_path = Path.join(:code.priv_dir(:towerops), "mibs/ubnt/UBNT-MIB")
|
||||
{:ok, content} = File.read(mib_path)
|
||||
|
|
|
|||
|
|
@ -84,31 +84,31 @@ defmodule SnmpKit.SnmpLib.OIDTest do
|
|||
parent = [1, 3, 6, 1, 2, 1]
|
||||
child = [1, 3, 6, 1, 2, 1, 1, 1, 0]
|
||||
|
||||
assert OID.is_child_of?(child, parent) == true
|
||||
assert OID.is_child_of?(parent, child) == false
|
||||
assert OID.child_of?(child, parent) == true
|
||||
assert OID.child_of?(parent, child) == false
|
||||
end
|
||||
|
||||
test "correctly identifies parent relationships" do
|
||||
parent = [1, 3, 6, 1, 2, 1]
|
||||
child = [1, 3, 6, 1, 2, 1, 1, 1, 0]
|
||||
|
||||
assert OID.is_parent_of?(parent, child) == true
|
||||
assert OID.is_parent_of?(child, parent) == false
|
||||
assert OID.parent_of?(parent, child) == true
|
||||
assert OID.parent_of?(child, parent) == false
|
||||
end
|
||||
|
||||
test "rejects equal OIDs as child/parent" do
|
||||
oid = [1, 3, 6, 1, 2, 1]
|
||||
|
||||
assert OID.is_child_of?(oid, oid) == false
|
||||
assert OID.is_parent_of?(oid, oid) == false
|
||||
assert OID.child_of?(oid, oid) == false
|
||||
assert OID.parent_of?(oid, oid) == false
|
||||
end
|
||||
|
||||
test "rejects sibling OIDs as child/parent" do
|
||||
oid1 = [1, 3, 6, 1, 2, 1]
|
||||
oid2 = [1, 3, 6, 1, 2, 2]
|
||||
|
||||
assert OID.is_child_of?(oid1, oid2) == false
|
||||
assert OID.is_child_of?(oid2, oid1) == false
|
||||
assert OID.child_of?(oid1, oid2) == false
|
||||
assert OID.child_of?(oid2, oid1) == false
|
||||
end
|
||||
|
||||
test "gets parent OID correctly" do
|
||||
|
|
@ -322,24 +322,24 @@ defmodule SnmpKit.SnmpLib.OIDTest do
|
|||
end
|
||||
|
||||
test "identifies MIB-2 OIDs" do
|
||||
assert OID.is_mib_2?([1, 3, 6, 1, 2, 1]) == true
|
||||
assert OID.is_mib_2?([1, 3, 6, 1, 2, 1, 1, 1, 0]) == true
|
||||
assert OID.is_mib_2?([1, 3, 6, 1, 4, 1, 9]) == false
|
||||
assert OID.mib_2?([1, 3, 6, 1, 2, 1]) == true
|
||||
assert OID.mib_2?([1, 3, 6, 1, 2, 1, 1, 1, 0]) == true
|
||||
assert OID.mib_2?([1, 3, 6, 1, 4, 1, 9]) == false
|
||||
end
|
||||
|
||||
test "identifies enterprise OIDs" do
|
||||
assert OID.is_enterprise?([1, 3, 6, 1, 4, 1, 9, 1, 1]) == true
|
||||
assert OID.is_enterprise?([1, 3, 6, 1, 2, 1, 1, 1, 0]) == false
|
||||
assert OID.enterprise?([1, 3, 6, 1, 4, 1, 9, 1, 1]) == true
|
||||
assert OID.enterprise?([1, 3, 6, 1, 2, 1, 1, 1, 0]) == false
|
||||
end
|
||||
|
||||
test "identifies experimental OIDs" do
|
||||
assert OID.is_experimental?([1, 3, 6, 1, 3, 1]) == true
|
||||
assert OID.is_experimental?([1, 3, 6, 1, 2, 1]) == false
|
||||
assert OID.experimental?([1, 3, 6, 1, 3, 1]) == true
|
||||
assert OID.experimental?([1, 3, 6, 1, 2, 1]) == false
|
||||
end
|
||||
|
||||
test "identifies private OIDs" do
|
||||
assert OID.is_private?([1, 3, 6, 1, 4, 2]) == true
|
||||
assert OID.is_private?([1, 3, 6, 1, 2, 1]) == false
|
||||
assert OID.private?([1, 3, 6, 1, 4, 2]) == true
|
||||
assert OID.private?([1, 3, 6, 1, 2, 1]) == false
|
||||
end
|
||||
|
||||
test "extracts enterprise numbers" do
|
||||
|
|
@ -361,8 +361,8 @@ defmodule SnmpKit.SnmpLib.OIDTest do
|
|||
test "handles invalid inputs gracefully" do
|
||||
assert {:error, :invalid_input} = OID.string_to_list(:not_binary)
|
||||
assert {:error, :invalid_input} = OID.list_to_string(:not_list)
|
||||
assert OID.is_child_of?(:not_list, [1, 2, 3]) == false
|
||||
assert OID.is_child_of?([1, 2, 3], :not_list) == false
|
||||
assert OID.child_of?(:not_list, [1, 2, 3]) == false
|
||||
assert OID.child_of?([1, 2, 3], :not_list) == false
|
||||
assert {:error, :invalid_input} = OID.get_parent(:not_list)
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@ defmodule SnmpKit.SnmpLib.PDUTest do
|
|||
assert {:ok, _} = PDU.validate(pdu)
|
||||
|
||||
invalid_pdu = %{type: :invalid_type, request_id: 123, varbinds: []}
|
||||
assert {:error, :invalid_pdu_type} = PDU.validate(invalid_pdu)
|
||||
assert :error = PDU.validate(invalid_pdu)
|
||||
end
|
||||
|
||||
test "validates varbinds format" do
|
||||
|
|
|
|||
|
|
@ -262,10 +262,10 @@ defmodule SnmpKit.SnmpLib.TransportTest do
|
|||
end
|
||||
|
||||
test "identifies SNMP ports" do
|
||||
assert Transport.is_snmp_port?(161) == true
|
||||
assert Transport.is_snmp_port?(162) == true
|
||||
assert Transport.is_snmp_port?(80) == false
|
||||
assert Transport.is_snmp_port?(443) == false
|
||||
assert Transport.snmp_port?(161) == true
|
||||
assert Transport.snmp_port?(162) == true
|
||||
assert Transport.snmp_port?(80) == false
|
||||
assert Transport.snmp_port?(443) == false
|
||||
end
|
||||
|
||||
test "returns maximum SNMP payload size" do
|
||||
|
|
|
|||
|
|
@ -79,7 +79,7 @@ defmodule SnmpKit.SnmpLib.TypesTest do
|
|||
test "rejects invalid IP addresses" do
|
||||
assert {:error, :invalid_length} = Types.validate_ip_address(<<192, 168, 1>>)
|
||||
assert {:error, :invalid_length} = Types.validate_ip_address(<<192, 168, 1, 1, 1>>)
|
||||
assert {:error, :invalid_format} = Types.validate_ip_address({256, 1, 1, 1})
|
||||
assert {:error, :out_of_range} = Types.validate_ip_address({256, 1, 1, 1})
|
||||
assert {:error, :invalid_format} = Types.validate_ip_address({1, 2, 3})
|
||||
assert {:error, :invalid_format} = Types.validate_ip_address("192.168.1.1")
|
||||
end
|
||||
|
|
@ -382,32 +382,32 @@ defmodule SnmpKit.SnmpLib.TypesTest do
|
|||
|
||||
describe "Type classification utilities" do
|
||||
test "identifies numeric types correctly" do
|
||||
assert Types.is_numeric_type?(:integer) == true
|
||||
assert Types.is_numeric_type?(:counter32) == true
|
||||
assert Types.is_numeric_type?(:gauge32) == true
|
||||
assert Types.is_numeric_type?(:timeticks) == true
|
||||
assert Types.is_numeric_type?(:counter64) == true
|
||||
assert Types.numeric_type?(:integer) == true
|
||||
assert Types.numeric_type?(:counter32) == true
|
||||
assert Types.numeric_type?(:gauge32) == true
|
||||
assert Types.numeric_type?(:timeticks) == true
|
||||
assert Types.numeric_type?(:counter64) == true
|
||||
|
||||
assert Types.is_numeric_type?(:string) == false
|
||||
assert Types.is_numeric_type?(:oid) == false
|
||||
assert Types.numeric_type?(:string) == false
|
||||
assert Types.numeric_type?(:oid) == false
|
||||
end
|
||||
|
||||
test "identifies binary types correctly" do
|
||||
assert Types.is_binary_type?(:string) == true
|
||||
assert Types.is_binary_type?(:opaque) == true
|
||||
assert Types.is_binary_type?(:ip_address) == true
|
||||
assert Types.binary_type?(:string) == true
|
||||
assert Types.binary_type?(:opaque) == true
|
||||
assert Types.binary_type?(:ip_address) == true
|
||||
|
||||
assert Types.is_binary_type?(:integer) == false
|
||||
assert Types.is_binary_type?(:counter32) == false
|
||||
assert Types.binary_type?(:integer) == false
|
||||
assert Types.binary_type?(:counter32) == false
|
||||
end
|
||||
|
||||
test "identifies exception types correctly" do
|
||||
assert Types.is_exception_type?(:no_such_object) == true
|
||||
assert Types.is_exception_type?(:no_such_instance) == true
|
||||
assert Types.is_exception_type?(:end_of_mib_view) == true
|
||||
assert Types.exception_type?(:no_such_object) == true
|
||||
assert Types.exception_type?(:no_such_instance) == true
|
||||
assert Types.exception_type?(:end_of_mib_view) == true
|
||||
|
||||
assert Types.is_exception_type?(:integer) == false
|
||||
assert Types.is_exception_type?(:string) == false
|
||||
assert Types.exception_type?(:integer) == false
|
||||
assert Types.exception_type?(:string) == false
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -20,31 +20,31 @@ defmodule SnmpKit.SnmpMgr.FormatTest do
|
|||
result = {"1.3.6.1.2.1.2.2.1.10.1", :counter32, 42_000_000}
|
||||
{_oid, _type, formatted} = Format.pretty_print(result)
|
||||
|
||||
assert formatted == "42000000 (Counter32)"
|
||||
assert formatted == "42.0 Mbps"
|
||||
end
|
||||
|
||||
test "formats small counter types as integers" do
|
||||
# Counter32 values <= 1,000,000 are formatted with type label
|
||||
# Counter32 values <= 1,000,000 are formatted as plain integers
|
||||
result = {"1.3.6.1.2.1.2.2.1.10.1", :counter32, 500_000}
|
||||
{_oid, _type, formatted} = Format.pretty_print(result)
|
||||
|
||||
assert formatted == "500000 (Counter32)"
|
||||
assert formatted == "500000"
|
||||
end
|
||||
|
||||
test "formats large gauge types as bytes" do
|
||||
# Gauge32 values > 1,000,000 are formatted with type label
|
||||
# Gauge32 values > 1,000,000 are formatted as bytes
|
||||
result = {"1.3.6.1.2.1.2.2.1.5.1", :gauge32, 100_000_000}
|
||||
{_oid, _type, formatted} = Format.pretty_print(result)
|
||||
|
||||
assert formatted == "100000000 (Gauge32)"
|
||||
assert formatted == "95.4 MB"
|
||||
end
|
||||
|
||||
test "formats small gauge types as integers" do
|
||||
# Gauge32 values <= 1,000,000 are formatted with type label
|
||||
# Gauge32 values <= 1,000,000 are formatted as plain integers
|
||||
result = {"1.3.6.1.2.1.2.2.1.5.1", :gauge32, 500_000}
|
||||
{_oid, _type, formatted} = Format.pretty_print(result)
|
||||
|
||||
assert formatted == "500000 (Gauge32)"
|
||||
assert formatted == "500000"
|
||||
end
|
||||
|
||||
test "formats object identifiers" do
|
||||
|
|
@ -65,8 +65,8 @@ defmodule SnmpKit.SnmpMgr.FormatTest do
|
|||
result = {"1.3.6.1.2.1.1.1.0", :unknown_type, "test value"}
|
||||
{_oid, _type, formatted} = Format.pretty_print(result)
|
||||
|
||||
# Unknown types with printable binary values are inspected
|
||||
assert formatted == "\"test value\""
|
||||
# Unknown types with printable binary values are returned as-is
|
||||
assert formatted == "test value"
|
||||
end
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ defmodule SnmpKit.SnmpMgr.MIB.ExtractionTest do
|
|||
alias SnmpKit.SnmpLib.MIB.Parser
|
||||
|
||||
# Test the actual extract_mib_mappings private function behavior
|
||||
@tag :yecc_required
|
||||
test "extract_mib_mappings resolves UBNT-MIB OIDs" do
|
||||
mib_path = Path.join(:code.priv_dir(:towerops), "mibs/ubnt/UBNT-MIB")
|
||||
{:ok, mib_content} = File.read(mib_path)
|
||||
|
|
|
|||
|
|
@ -2,6 +2,20 @@ defmodule SnmpKit.SnmpMgr.MIBTest do
|
|||
use ExUnit.Case, async: false
|
||||
|
||||
alias SnmpKit.SnmpLib.MIB.Parser
|
||||
alias SnmpKit.SnmpMgr.MIB
|
||||
|
||||
setup do
|
||||
# Start the MIB GenServer if not already running
|
||||
case GenServer.whereis(MIB) do
|
||||
nil ->
|
||||
{:ok, pid} = MIB.start_link()
|
||||
on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end)
|
||||
{:ok, mib_pid: pid}
|
||||
|
||||
pid ->
|
||||
{:ok, mib_pid: pid}
|
||||
end
|
||||
end
|
||||
|
||||
# Test helper to parse a MIB and extract name->OID map
|
||||
defp parse_and_extract(mib_path) do
|
||||
|
|
@ -90,6 +104,7 @@ defmodule SnmpKit.SnmpMgr.MIBTest do
|
|||
defp resolve_oid_def(_, _), do: {:error, :invalid_format}
|
||||
|
||||
describe "MIB parsing and OID resolution" do
|
||||
@tag :yecc_required
|
||||
test "workaround fixes missing sub_index for ubntAFLTU" do
|
||||
mib_path = Path.join(:code.priv_dir(:towerops), "mibs/ubnt/UBNT-MIB")
|
||||
oid_defs = parse_and_extract(mib_path)
|
||||
|
|
@ -99,6 +114,7 @@ defmodule SnmpKit.SnmpMgr.MIBTest do
|
|||
"ubntAFLTU should have been fixed to {:ubntMIB, <<10>>} by workaround"
|
||||
end
|
||||
|
||||
@tag :yecc_required
|
||||
test "parses UBNT-MIB and extracts OID definitions" do
|
||||
mib_path = Path.join(:code.priv_dir(:towerops), "mibs/ubnt/UBNT-MIB")
|
||||
oid_defs = parse_and_extract(mib_path)
|
||||
|
|
@ -160,7 +176,427 @@ defmodule SnmpKit.SnmpMgr.MIBTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "MIB GenServer integration" do
|
||||
# MIB loading integration tests removed - not critical for current functionality
|
||||
describe "MIB GenServer basic operations" do
|
||||
test "GenServer starts successfully" do
|
||||
assert Process.alive?(Process.whereis(MIB))
|
||||
end
|
||||
|
||||
test "load_standard_mibs returns :ok" do
|
||||
assert :ok = MIB.load_standard_mibs()
|
||||
end
|
||||
end
|
||||
|
||||
describe "resolve/1" do
|
||||
test "resolves standard MIB names" do
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 1, 1]} = MIB.resolve("sysDescr")
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 1, 5]} = MIB.resolve("sysName")
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 2, 2, 1, 2]} = MIB.resolve("ifDescr")
|
||||
end
|
||||
|
||||
test "resolves names with instance suffix" do
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 1, 1, 0]} = MIB.resolve("sysDescr.0")
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 2, 2, 1, 2, 1]} = MIB.resolve("ifDescr.1")
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 2, 2, 1, 2, 1, 2]} = MIB.resolve("ifDescr.1.2")
|
||||
end
|
||||
|
||||
test "returns error for unknown names" do
|
||||
assert {:error, :not_found} = MIB.resolve("unknownMibName")
|
||||
end
|
||||
|
||||
test "returns error for invalid input" do
|
||||
assert {:error, :invalid_name} = MIB.resolve(nil)
|
||||
assert {:error, :invalid_name} = MIB.resolve(123)
|
||||
end
|
||||
|
||||
test "returns error for names with invalid instance parts" do
|
||||
assert {:error, :invalid_instance} = MIB.resolve("sysDescr.abc")
|
||||
end
|
||||
end
|
||||
|
||||
describe "reverse_lookup/1" do
|
||||
test "reverses standard OIDs to names" do
|
||||
assert {:ok, "sysDescr"} = MIB.reverse_lookup([1, 3, 6, 1, 2, 1, 1, 1])
|
||||
assert {:ok, "sysName"} = MIB.reverse_lookup([1, 3, 6, 1, 2, 1, 1, 5])
|
||||
end
|
||||
|
||||
test "reverses OIDs with instance suffix" do
|
||||
assert {:ok, "sysDescr.0"} = MIB.reverse_lookup([1, 3, 6, 1, 2, 1, 1, 1, 0])
|
||||
assert {:ok, "ifDescr.1"} = MIB.reverse_lookup([1, 3, 6, 1, 2, 1, 2, 2, 1, 2, 1])
|
||||
end
|
||||
|
||||
test "accepts OID as string" do
|
||||
assert {:ok, "sysDescr"} = MIB.reverse_lookup("1.3.6.1.2.1.1.1")
|
||||
assert {:ok, "sysDescr.0"} = MIB.reverse_lookup("1.3.6.1.2.1.1.1.0")
|
||||
end
|
||||
|
||||
test "returns error for unknown OIDs" do
|
||||
assert {:error, :not_found} = MIB.reverse_lookup([9, 9, 9, 9, 9])
|
||||
end
|
||||
|
||||
test "returns error for empty OID" do
|
||||
assert {:error, :empty_oid} = MIB.reverse_lookup([])
|
||||
end
|
||||
|
||||
test "returns error for invalid OID format" do
|
||||
assert {:error, :invalid_oid_string} = MIB.reverse_lookup("not.a.valid.oid")
|
||||
end
|
||||
|
||||
test "returns error for binary OID (edge case)" do
|
||||
# This tests the guard in find_partial_reverse_match
|
||||
assert {:error, :invalid_oid_string} = MIB.reverse_lookup("binary_string")
|
||||
end
|
||||
end
|
||||
|
||||
describe "parent/1" do
|
||||
test "returns parent OID" do
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 1]} = MIB.parent([1, 3, 6, 1, 2, 1, 1, 1])
|
||||
assert {:ok, [1, 3, 6]} = MIB.parent([1, 3, 6, 1])
|
||||
end
|
||||
|
||||
test "returns empty list for single element" do
|
||||
assert {:ok, []} = MIB.parent([1])
|
||||
end
|
||||
|
||||
test "returns error for empty list" do
|
||||
assert {:error, :no_parent} = MIB.parent([])
|
||||
end
|
||||
|
||||
test "accepts OID as string" do
|
||||
assert {:ok, [1, 3, 6, 1, 2, 1, 1]} = MIB.parent("1.3.6.1.2.1.1.1")
|
||||
end
|
||||
|
||||
test "returns error for invalid string OID" do
|
||||
assert {:error, :invalid_oid_string} = MIB.parent("invalid")
|
||||
end
|
||||
end
|
||||
|
||||
describe "children/1" do
|
||||
test "finds direct children of an OID" do
|
||||
# system (1.3.6.1.2.1.1) should have children like sysDescr, sysObjectID, etc.
|
||||
{:ok, children} = MIB.children([1, 3, 6, 1, 2, 1, 1])
|
||||
assert is_list(children)
|
||||
assert "sysDescr" in children
|
||||
assert "sysName" in children
|
||||
end
|
||||
|
||||
test "accepts OID as string" do
|
||||
{:ok, children} = MIB.children("1.3.6.1.2.1.1")
|
||||
assert is_list(children)
|
||||
assert "sysDescr" in children
|
||||
end
|
||||
|
||||
test "accepts nil as root (returns all top-level entries)" do
|
||||
{:ok, children} = MIB.children(nil)
|
||||
assert is_list(children)
|
||||
end
|
||||
|
||||
test "returns empty list for leaf nodes" do
|
||||
# sysDescr is a leaf node
|
||||
{:ok, children} = MIB.children([1, 3, 6, 1, 2, 1, 1, 1])
|
||||
assert children == []
|
||||
end
|
||||
|
||||
test "returns error for invalid OID format" do
|
||||
assert {:error, :invalid_parent_oid} = MIB.children("invalid")
|
||||
assert {:error, :invalid_parent_oid} = MIB.children(123)
|
||||
end
|
||||
end
|
||||
|
||||
describe "walk_tree/2" do
|
||||
test "walks tree from root OID" do
|
||||
{:ok, descendants} = MIB.walk_tree([1, 3, 6, 1, 2, 1, 1])
|
||||
assert is_list(descendants)
|
||||
refute Enum.empty?(descendants)
|
||||
# Should return tuples of {name, oid}
|
||||
assert Enum.all?(descendants, fn {name, oid} -> is_binary(name) and is_list(oid) end)
|
||||
end
|
||||
|
||||
test "accepts OID as string" do
|
||||
{:ok, descendants} = MIB.walk_tree("1.3.6.1.2.1.1")
|
||||
assert is_list(descendants)
|
||||
refute Enum.empty?(descendants)
|
||||
end
|
||||
|
||||
test "accepts nil as root" do
|
||||
{:ok, descendants} = MIB.walk_tree(nil)
|
||||
assert is_list(descendants)
|
||||
# Should return all known OIDs
|
||||
refute Enum.empty?(descendants)
|
||||
end
|
||||
|
||||
test "returns empty list for unknown OID" do
|
||||
{:ok, descendants} = MIB.walk_tree([9, 9, 9, 9])
|
||||
assert descendants == []
|
||||
end
|
||||
end
|
||||
|
||||
describe "object_info/1" do
|
||||
test "returns enriched metadata for standard objects" do
|
||||
{:ok, info} = MIB.object_info("sysDescr")
|
||||
assert info.name == "sysDescr"
|
||||
assert info.oid == [1, 3, 6, 1, 2, 1, 1, 1]
|
||||
assert info.module == "SNMPv2-MIB"
|
||||
assert is_map(info.syntax)
|
||||
assert info.syntax.base == :octet_string
|
||||
assert info.syntax.textual_convention == "DisplayString"
|
||||
end
|
||||
|
||||
test "includes instance information when present" do
|
||||
{:ok, info} = MIB.object_info("sysDescr.0")
|
||||
assert info.name == "sysDescr"
|
||||
assert info.instance_index == 0
|
||||
assert info.instance_oid == [1, 3, 6, 1, 2, 1, 1, 1, 0]
|
||||
end
|
||||
|
||||
test "handles multi-part instance indices" do
|
||||
{:ok, info} = MIB.object_info("ifDescr.1.2")
|
||||
assert info.name == "ifDescr"
|
||||
assert info.instance_index == [1, 2]
|
||||
assert info.instance_oid == [1, 3, 6, 1, 2, 1, 2, 2, 1, 2, 1, 2]
|
||||
end
|
||||
|
||||
test "accepts OID as list" do
|
||||
{:ok, info} = MIB.object_info([1, 3, 6, 1, 2, 1, 1, 1])
|
||||
assert info.name == "sysDescr"
|
||||
end
|
||||
|
||||
test "accepts OID as dotted string" do
|
||||
{:ok, info} = MIB.object_info("1.3.6.1.2.1.1.1")
|
||||
assert info.name == "sysDescr"
|
||||
end
|
||||
|
||||
test "returns error for invalid input" do
|
||||
assert {:error, _} = MIB.object_info(123)
|
||||
assert {:error, _} = MIB.object_info(%{})
|
||||
end
|
||||
end
|
||||
|
||||
describe "reverse_lookup_enriched/1" do
|
||||
test "is an alias for object_info/1" do
|
||||
{:ok, info1} = MIB.object_info("sysDescr")
|
||||
{:ok, info2} = MIB.reverse_lookup_enriched("sysDescr")
|
||||
assert info1 == info2
|
||||
end
|
||||
end
|
||||
|
||||
describe "object_info_many/1" do
|
||||
test "returns list of info for multiple names" do
|
||||
{:ok, infos} = MIB.object_info_many(["sysDescr", "sysName", "sysLocation"])
|
||||
assert length(infos) == 3
|
||||
assert Enum.all?(infos, &is_map/1)
|
||||
assert Enum.map(infos, & &1.name) == ["sysDescr", "sysName", "sysLocation"]
|
||||
end
|
||||
|
||||
test "returns error if any lookup fails" do
|
||||
assert {:error, _} = MIB.object_info_many(["sysDescr", "invalidName"])
|
||||
end
|
||||
end
|
||||
|
||||
describe "resolve_enhanced/2" do
|
||||
test "resolves names to object info" do
|
||||
{:ok, info} = MIB.resolve_enhanced("sysDescr")
|
||||
assert is_map(info)
|
||||
assert info.name == "sysDescr"
|
||||
assert info.oid == [1, 3, 6, 1, 2, 1, 1, 1]
|
||||
end
|
||||
end
|
||||
|
||||
describe "compile/2" do
|
||||
test "returns error for non-existent file" do
|
||||
assert {:error, _} = MIB.compile("nonexistent.mib")
|
||||
end
|
||||
|
||||
test "handles snmp_lib_not_available gracefully" do
|
||||
# The function should return an error, not crash
|
||||
result = MIB.compile("test.mib")
|
||||
assert match?({:error, _}, result)
|
||||
end
|
||||
end
|
||||
|
||||
describe "compile_dir/2" do
|
||||
test "returns error for non-existent directory" do
|
||||
assert {:error, {:directory_error, :enoent}} = MIB.compile_dir("/nonexistent/path")
|
||||
end
|
||||
|
||||
test "processes existing directory" do
|
||||
# Create a temporary directory
|
||||
tmp_dir = Path.join(System.tmp_dir!(), "mib_test_#{:rand.uniform(10_000)}")
|
||||
File.mkdir_p!(tmp_dir)
|
||||
|
||||
try do
|
||||
# Should return ok with empty results since no .mib files
|
||||
result = MIB.compile_dir(tmp_dir)
|
||||
assert match?({:ok, _}, result) or match?({:error, _}, result)
|
||||
after
|
||||
File.rm_rf!(tmp_dir)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
describe "parse_mib_file/2" do
|
||||
test "returns error for non-existent file" do
|
||||
assert {:error, {:file_read_error, :enoent}} = MIB.parse_mib_file("nonexistent.mib")
|
||||
end
|
||||
end
|
||||
|
||||
describe "parse_mib_content/2" do
|
||||
test "parses basic MIB content" do
|
||||
content = """
|
||||
TestMIB DEFINITIONS ::= BEGIN
|
||||
testObject OBJECT-TYPE
|
||||
SYNTAX Integer32
|
||||
MAX-ACCESS read-only
|
||||
STATUS current
|
||||
DESCRIPTION "Test object"
|
||||
::= { testGroup 1 }
|
||||
END
|
||||
"""
|
||||
|
||||
{:ok, result} = MIB.parse_mib_content(content)
|
||||
assert is_map(result)
|
||||
assert Map.has_key?(result, :tokens)
|
||||
assert Map.has_key?(result, :parsed_objects)
|
||||
end
|
||||
|
||||
test "returns error for invalid MIB content" do
|
||||
# Completely invalid content that can't be tokenized
|
||||
result = MIB.parse_mib_content("not valid mib at all!!!")
|
||||
assert match?({:error, {:tokenization_failed, _}}, result)
|
||||
end
|
||||
end
|
||||
|
||||
describe "load/1" do
|
||||
test "handles non-existent compiled MIB path" do
|
||||
# Should return error, not crash
|
||||
result = MIB.load("/nonexistent/compiled.mib")
|
||||
assert match?({:error, _}, result) or match?(:ok, result)
|
||||
end
|
||||
end
|
||||
|
||||
describe "load_and_integrate_mib/2" do
|
||||
test "returns error when compilation fails" do
|
||||
result = MIB.load_and_integrate_mib("nonexistent.mib")
|
||||
assert match?({:error, _}, result)
|
||||
end
|
||||
end
|
||||
|
||||
describe "syntax parsing" do
|
||||
test "handles various syntax types in object_info" do
|
||||
# Test different syntax types through object_info
|
||||
{:ok, sys_descr_info} = MIB.object_info("sysDescr")
|
||||
assert sys_descr_info.syntax.base == :octet_string
|
||||
assert sys_descr_info.syntax.textual_convention == "DisplayString"
|
||||
|
||||
{:ok, if_speed_info} = MIB.object_info("ifSpeed")
|
||||
assert if_speed_info.syntax.base == :gauge32
|
||||
end
|
||||
|
||||
test "handles objects without curated syntax metadata" do
|
||||
# Enterprise OIDs don't have curated syntax metadata
|
||||
{:ok, enterprises_info} = MIB.object_info("enterprises")
|
||||
# Should still return syntax map even if empty
|
||||
assert is_map(enterprises_info.syntax)
|
||||
end
|
||||
end
|
||||
|
||||
describe "MIB integration with parsed content" do
|
||||
test "integrates MIB data with complex OID definitions" do
|
||||
# Simulate parsed MIB data with parent references
|
||||
mib_content = """
|
||||
TEST-MIB DEFINITIONS ::= BEGIN
|
||||
testRoot OBJECT IDENTIFIER ::= { enterprises 99999 }
|
||||
testObject OBJECT-TYPE
|
||||
SYNTAX Integer32
|
||||
MAX-ACCESS read-only
|
||||
STATUS current
|
||||
::= { testRoot 1 }
|
||||
END
|
||||
"""
|
||||
|
||||
# This will exercise the parse_mib_content path
|
||||
{:ok, result} = MIB.parse_mib_content(mib_content)
|
||||
assert is_map(result)
|
||||
assert Map.has_key?(result, :tokens)
|
||||
end
|
||||
end
|
||||
|
||||
describe "error handling edge cases" do
|
||||
test "compile handles SnmpLib.MIB compilation errors gracefully" do
|
||||
# Should return error tuple, not crash
|
||||
result = MIB.compile("invalid_mib_file.mib")
|
||||
assert match?({:error, _}, result)
|
||||
end
|
||||
|
||||
test "compile_dir handles fallback when SnmpLib is not available" do
|
||||
# Create a temp directory with a .mib file to test fallback
|
||||
tmp_dir = Path.join(System.tmp_dir!(), "mib_fallback_test_#{:rand.uniform(10_000)}")
|
||||
File.mkdir_p!(tmp_dir)
|
||||
|
||||
try do
|
||||
# Create a dummy .mib file
|
||||
test_mib = Path.join(tmp_dir, "test.mib")
|
||||
File.write!(test_mib, "TEST-MIB DEFINITIONS ::= BEGIN END")
|
||||
|
||||
# This should trigger compile_dir_fallback
|
||||
result = MIB.compile_dir(tmp_dir)
|
||||
assert match?({:ok, _}, result) or match?({:error, _}, result)
|
||||
after
|
||||
File.rm_rf!(tmp_dir)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
describe "private helper coverage" do
|
||||
test "object_info handles dotted OID strings" do
|
||||
# This exercises numeric_oid_string? and OID parsing through object_info
|
||||
{:ok, info} = MIB.object_info("1.3.6.1.2.1.1.1")
|
||||
assert info.name == "sysDescr"
|
||||
assert info.oid == [1, 3, 6, 1, 2, 1, 1, 1]
|
||||
end
|
||||
|
||||
test "object_info handles leading dot in OID string" do
|
||||
{:ok, info} = MIB.object_info(".1.3.6.1.2.1.1.1")
|
||||
assert info.name == "sysDescr"
|
||||
assert info.oid == [1, 3, 6, 1, 2, 1, 1, 1]
|
||||
end
|
||||
|
||||
test "object_info handles complex instance indices" do
|
||||
# Multi-part instance index
|
||||
{:ok, info} = MIB.object_info([1, 3, 6, 1, 2, 1, 2, 2, 1, 2, 1, 0, 0])
|
||||
assert info.name == "ifDescr"
|
||||
assert info.instance_index == [1, 0, 0]
|
||||
end
|
||||
|
||||
test "children returns sorted results" do
|
||||
{:ok, children} = MIB.children([1, 3, 6, 1, 2, 1, 1])
|
||||
# Should be sorted alphabetically
|
||||
assert children == Enum.sort(children)
|
||||
end
|
||||
|
||||
test "walk_tree returns sorted results by OID" do
|
||||
{:ok, descendants} = MIB.walk_tree([1, 3, 6, 1, 2, 1, 1])
|
||||
# Extract OIDs and verify they're sorted
|
||||
oids = Enum.map(descendants, fn {_name, oid} -> oid end)
|
||||
assert oids == Enum.sort(oids)
|
||||
end
|
||||
end
|
||||
|
||||
describe "module_for helper" do
|
||||
test "identifies correct modules for known prefixes" do
|
||||
{:ok, info} = MIB.object_info("ifHCInOctets")
|
||||
assert info.module == "IF-MIB"
|
||||
|
||||
{:ok, info2} = MIB.object_info("sysDescr")
|
||||
assert info2.module == "SNMPv2-MIB"
|
||||
|
||||
{:ok, info3} = MIB.object_info("ifDescr")
|
||||
assert info3.module == "IF-MIB"
|
||||
end
|
||||
|
||||
test "handles objects without module mapping" do
|
||||
{:ok, info} = MIB.object_info("enterprises")
|
||||
# May not have a module mapping, should handle gracefully
|
||||
assert is_nil(info.module) or is_binary(info.module)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
295
test/towerops/accounts/browser_session_test.exs
Normal file
295
test/towerops/accounts/browser_session_test.exs
Normal file
|
|
@ -0,0 +1,295 @@
|
|||
defmodule Towerops.Accounts.BrowserSessionTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias Towerops.Accounts.BrowserSession
|
||||
|
||||
describe "create_changeset/2" do
|
||||
test "valid changeset with all required fields" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with optional device metadata" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
device_name: "Chrome on macOS",
|
||||
browser_name: "Chrome",
|
||||
browser_version: "120.0.0",
|
||||
os_name: "macOS",
|
||||
os_version: "14.2",
|
||||
device_type: "desktop",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert changeset.changes.device_name == "Chrome on macOS"
|
||||
assert changeset.changes.browser_name == "Chrome"
|
||||
assert changeset.changes.device_type == "desktop"
|
||||
end
|
||||
|
||||
test "valid changeset with location data" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
country_code: "US",
|
||||
country_name: "United States",
|
||||
city_name: "San Francisco",
|
||||
subdivision_1_name: "California",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert changeset.changes.country_name == "United States"
|
||||
assert changeset.changes.city_name == "San Francisco"
|
||||
end
|
||||
|
||||
test "requires user_id" do
|
||||
attrs = %{
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{user_id: ["can't be blank"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "requires user_token_id" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{user_token_id: ["can't be blank"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "requires ip_address" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{ip_address: ["can't be blank"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "requires last_activity_at" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{last_activity_at: ["can't be blank"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "requires expires_at" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{expires_at: ["can't be blank"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "validates ip_address max length (45 chars for IPv6)" do
|
||||
# IPv6 address can be up to 45 characters with full notation
|
||||
valid_ipv6 = "2001:0db8:0000:0000:0000:8a2e:0370:7334"
|
||||
assert String.length(valid_ipv6) == 39
|
||||
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: valid_ipv6,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
assert changeset.valid?
|
||||
|
||||
# Test too long
|
||||
too_long = String.duplicate("x", 46)
|
||||
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: too_long,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{ip_address: ["should be at most 45 character(s)"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "validates device_name max length (255 chars)" do
|
||||
too_long = String.duplicate("x", 256)
|
||||
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
device_name: too_long,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{device_name: ["should be at most 255 character(s)"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "validates browser_name max length (100 chars)" do
|
||||
too_long = String.duplicate("x", 101)
|
||||
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
browser_name: too_long,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{browser_name: ["should be at most 100 character(s)"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "validates os_name max length (100 chars)" do
|
||||
too_long = String.duplicate("x", 101)
|
||||
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
os_name: too_long,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{os_name: ["should be at most 100 character(s)"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "validates device_type inclusion (desktop, mobile, tablet)" do
|
||||
# Test valid values
|
||||
for valid_type <- ["desktop", "mobile", "tablet"] do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
device_type: valid_type,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
# Test invalid value
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
device_type: "smartwatch",
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert %{device_type: ["is invalid"]} = errors_on(changeset)
|
||||
end
|
||||
|
||||
test "allows nil device_type" do
|
||||
attrs = %{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
user_token_id: Ecto.UUID.generate(),
|
||||
ip_address: "192.168.1.1",
|
||||
device_type: nil,
|
||||
last_activity_at: ~U[2026-01-29 19:00:00Z],
|
||||
expires_at: ~U[2026-02-12 19:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.create_changeset(%BrowserSession{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
end
|
||||
|
||||
describe "touch_changeset/1" do
|
||||
test "updates last_activity_at to current time" do
|
||||
session = %BrowserSession{
|
||||
last_activity_at: ~U[2026-01-29 10:00:00Z]
|
||||
}
|
||||
|
||||
changeset = BrowserSession.touch_changeset(session)
|
||||
|
||||
assert changeset.valid?
|
||||
assert changeset.changes.last_activity_at
|
||||
# Verify it's a recent timestamp (within last 5 seconds)
|
||||
assert DateTime.diff(changeset.changes.last_activity_at, DateTime.utc_now()) < 5
|
||||
end
|
||||
end
|
||||
|
||||
describe "anonymize_changeset/1" do
|
||||
test "sets user_id to nil and records anonymized_at timestamp" do
|
||||
session = %BrowserSession{
|
||||
user_id: Ecto.UUID.generate(),
|
||||
anonymized_at: nil
|
||||
}
|
||||
|
||||
changeset = BrowserSession.anonymize_changeset(session)
|
||||
|
||||
assert changeset.valid?
|
||||
assert changeset.changes.user_id == nil
|
||||
assert changeset.changes.anonymized_at
|
||||
# Verify it's a recent timestamp
|
||||
assert DateTime.diff(changeset.changes.anonymized_at, DateTime.utc_now()) < 5
|
||||
end
|
||||
end
|
||||
|
||||
# Helper function to convert changeset errors to a map
|
||||
defp errors_on(changeset) do
|
||||
Ecto.Changeset.traverse_errors(changeset, fn {message, opts} ->
|
||||
Regex.replace(~r"%{(\w+)}", message, fn _, key ->
|
||||
opts |> Keyword.get(String.to_existing_atom(key), key) |> to_string()
|
||||
end)
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
|
@ -998,4 +998,79 @@ defmodule Towerops.Agents.StatsTest do
|
|||
assert candidates == []
|
||||
end
|
||||
end
|
||||
|
||||
describe "edge cases" do
|
||||
test "get_device_assignment_breakdown/1 handles organization with SNMP-disabled devices" do
|
||||
user = user_fixture()
|
||||
org = organization_fixture(user.id)
|
||||
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
|
||||
|
||||
# Create device with SNMP disabled
|
||||
{:ok, _device} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Non-SNMP Device",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id,
|
||||
snmp_enabled: false
|
||||
})
|
||||
|
||||
breakdown = Stats.get_device_assignment_breakdown(org.id)
|
||||
|
||||
# Should return all zeros since no SNMP-enabled devices exist
|
||||
assert breakdown == %{direct: 0, site: 0, organization: 0, cloud: 0}
|
||||
end
|
||||
|
||||
test "get_high_load_agents/2 returns empty list with zero device count" do
|
||||
user = user_fixture()
|
||||
org = organization_fixture(user.id)
|
||||
{:ok, _agent_token, _token} = Agents.create_agent_token(org.id, "Zero Device Agent")
|
||||
|
||||
# No devices assigned at all
|
||||
high_load = Stats.get_high_load_agents(org.id, 1)
|
||||
|
||||
assert high_load == []
|
||||
end
|
||||
|
||||
test "find_reassignment_candidates/1 handles devices with no current agent assignment" do
|
||||
user = user_fixture()
|
||||
org = organization_fixture(user.id)
|
||||
{:ok, agent, _} = Agents.create_agent_token(org.id, "Agent")
|
||||
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
|
||||
|
||||
{:ok, device} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Unassigned Router",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id,
|
||||
snmp_enabled: true,
|
||||
snmp_community: "public",
|
||||
snmp_version: "2c"
|
||||
})
|
||||
|
||||
# Create checks from agent even though device isn't assigned
|
||||
for _ <- 1..15 do
|
||||
Monitoring.create_check(%{
|
||||
device_id: device.id,
|
||||
agent_token_id: agent.id,
|
||||
status: :success,
|
||||
response_time_ms: 50,
|
||||
checked_at: DateTime.utc_now()
|
||||
})
|
||||
end
|
||||
|
||||
# Device with no agent shouldn't be in candidates
|
||||
candidates = Stats.find_reassignment_candidates()
|
||||
|
||||
assert candidates == []
|
||||
end
|
||||
|
||||
test "get_offline_agents/1 returns empty for organization with no agents" do
|
||||
user = user_fixture()
|
||||
org = organization_fixture(user.id)
|
||||
|
||||
offline = Stats.get_offline_agents(org.id)
|
||||
|
||||
assert offline == []
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ defmodule Towerops.AgentsTest do
|
|||
|
||||
import Towerops.AccountsFixtures
|
||||
|
||||
alias Towerops.Admin.AuditLog
|
||||
alias Towerops.Agents
|
||||
alias Towerops.Agents.AgentAssignment
|
||||
alias Towerops.Agents.AgentToken
|
||||
|
|
@ -1463,4 +1464,64 @@ defmodule Towerops.AgentsTest do
|
|||
assert Enum.any?(targets, &(&1.id == device2.id))
|
||||
end
|
||||
end
|
||||
|
||||
describe "toggle_agent_debug/3" do
|
||||
test "enables remote debug and creates audit log", %{organization: org, user: user} do
|
||||
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Test Agent")
|
||||
|
||||
# Make user a superuser
|
||||
user = Repo.update!(Ecto.Changeset.change(user, is_superuser: true))
|
||||
|
||||
# Enable debug
|
||||
assert {:ok, updated_agent} = Agents.toggle_agent_debug(agent_token, true, user)
|
||||
assert updated_agent.allow_remote_debug == true
|
||||
|
||||
# Verify audit log was created
|
||||
audit_log = Repo.one(AuditLog)
|
||||
assert audit_log.superuser_id == user.id
|
||||
assert audit_log.action == "agent_debug_enable"
|
||||
assert audit_log.metadata["agent_token_id"] == agent_token.id
|
||||
assert audit_log.metadata["agent_name"] == "Test Agent"
|
||||
end
|
||||
|
||||
test "disables remote debug and creates audit log", %{organization: org, user: user} do
|
||||
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Test Agent")
|
||||
|
||||
# Make user a superuser
|
||||
user = Repo.update!(Ecto.Changeset.change(user, is_superuser: true))
|
||||
|
||||
# First enable it
|
||||
{:ok, agent_token} = Agents.toggle_agent_debug(agent_token, true, user)
|
||||
|
||||
# Now disable it
|
||||
assert {:ok, updated_agent} = Agents.toggle_agent_debug(agent_token, false, user)
|
||||
assert updated_agent.allow_remote_debug == false
|
||||
|
||||
# Verify audit log was created with correct action
|
||||
audit_logs = Repo.all(AuditLog)
|
||||
disable_log = Enum.find(audit_logs, &(&1.action == "agent_debug_disable"))
|
||||
assert disable_log.superuser_id == user.id
|
||||
assert disable_log.metadata["agent_token_id"] == agent_token.id
|
||||
end
|
||||
end
|
||||
|
||||
describe "broadcast_assignment_change/2" do
|
||||
test "broadcasts assignment change with valid agent_token_id" do
|
||||
# Subscribe to the PubSub topic
|
||||
agent_token_id = Ecto.UUID.generate()
|
||||
Phoenix.PubSub.subscribe(Towerops.PubSub, "agent:#{agent_token_id}:assignments")
|
||||
|
||||
# Broadcast the change
|
||||
assert :ok = Agents.broadcast_assignment_change(agent_token_id, :device_assigned)
|
||||
|
||||
# Verify we received the broadcast
|
||||
assert_receive {:assignments_changed, :device_assigned}
|
||||
end
|
||||
|
||||
test "handles nil agent_token_id gracefully" do
|
||||
# Should return :ok without broadcasting
|
||||
assert :ok = Agents.broadcast_assignment_change(nil, :device_unassigned)
|
||||
# No assertion needed - just verify it doesn't crash
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue