Add comprehensive test coverage for core context modules
Improves test coverage toward 90% target by adding tests for: - Organizations context (bulk SNMP/agent operations) - Sites context (hierarchical structure and bulk operations) - Profiles context and schemas (device/sensor OID management) - EctoTypes.JsonAny custom type - Snmp.Neighbor schema (LLDP/CDP topology) All new tests validate changeset validations, business logic, and database constraints to ensure data integrity. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
b8b3c5dce0
commit
d34b8b0f3f
8 changed files with 1667 additions and 0 deletions
184
test/towerops/ecto_types/json_any_test.exs
Normal file
184
test/towerops/ecto_types/json_any_test.exs
Normal file
|
|
@ -0,0 +1,184 @@
|
|||
defmodule Towerops.EctoTypes.JsonAnyTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias Towerops.EctoTypes.JsonAny
|
||||
|
||||
describe "type/0" do
|
||||
test "returns :jsonb" do
|
||||
assert JsonAny.type() == :jsonb
|
||||
end
|
||||
end
|
||||
|
||||
describe "cast/1" do
|
||||
test "casts nil" do
|
||||
assert JsonAny.cast(nil) == {:ok, nil}
|
||||
end
|
||||
|
||||
test "casts strings" do
|
||||
assert JsonAny.cast("hello") == {:ok, "hello"}
|
||||
assert JsonAny.cast("") == {:ok, ""}
|
||||
assert JsonAny.cast("with spaces") == {:ok, "with spaces"}
|
||||
end
|
||||
|
||||
test "casts integers" do
|
||||
assert JsonAny.cast(0) == {:ok, 0}
|
||||
assert JsonAny.cast(42) == {:ok, 42}
|
||||
assert JsonAny.cast(-100) == {:ok, -100}
|
||||
end
|
||||
|
||||
test "casts floats" do
|
||||
assert JsonAny.cast(0.0) == {:ok, 0.0}
|
||||
assert JsonAny.cast(3.14) == {:ok, 3.14}
|
||||
assert JsonAny.cast(-99.99) == {:ok, -99.99}
|
||||
end
|
||||
|
||||
test "casts booleans" do
|
||||
assert JsonAny.cast(true) == {:ok, true}
|
||||
assert JsonAny.cast(false) == {:ok, false}
|
||||
end
|
||||
|
||||
test "casts lists" do
|
||||
assert JsonAny.cast([]) == {:ok, []}
|
||||
assert JsonAny.cast([1, 2, 3]) == {:ok, [1, 2, 3]}
|
||||
assert JsonAny.cast(["a", "b"]) == {:ok, ["a", "b"]}
|
||||
assert JsonAny.cast([%{key: "value"}]) == {:ok, [%{key: "value"}]}
|
||||
end
|
||||
|
||||
test "casts maps" do
|
||||
assert JsonAny.cast(%{}) == {:ok, %{}}
|
||||
assert JsonAny.cast(%{key: "value"}) == {:ok, %{key: "value"}}
|
||||
assert JsonAny.cast(%{"nested" => %{"map" => true}}) == {:ok, %{"nested" => %{"map" => true}}}
|
||||
end
|
||||
|
||||
test "returns error for unsupported types" do
|
||||
assert JsonAny.cast({:tuple, "not supported"}) == :error
|
||||
assert JsonAny.cast(self()) == :error
|
||||
assert JsonAny.cast(make_ref()) == :error
|
||||
assert JsonAny.cast(fn -> :ok end) == :error
|
||||
end
|
||||
end
|
||||
|
||||
describe "load/1" do
|
||||
test "loads any value as-is" do
|
||||
assert JsonAny.load(nil) == {:ok, nil}
|
||||
assert JsonAny.load("string") == {:ok, "string"}
|
||||
assert JsonAny.load(42) == {:ok, 42}
|
||||
assert JsonAny.load(3.14) == {:ok, 3.14}
|
||||
assert JsonAny.load(true) == {:ok, true}
|
||||
assert JsonAny.load(false) == {:ok, false}
|
||||
assert JsonAny.load([1, 2, 3]) == {:ok, [1, 2, 3]}
|
||||
assert JsonAny.load(%{key: "value"}) == {:ok, %{key: "value"}}
|
||||
end
|
||||
|
||||
test "loads complex nested structures" do
|
||||
value = %{"users" => [%{"name" => "Alice", "age" => 30}, %{"name" => "Bob", "age" => 25}]}
|
||||
assert JsonAny.load(value) == {:ok, value}
|
||||
end
|
||||
end
|
||||
|
||||
describe "dump/1" do
|
||||
test "dumps nil" do
|
||||
assert JsonAny.dump(nil) == {:ok, nil}
|
||||
end
|
||||
|
||||
test "dumps strings" do
|
||||
assert JsonAny.dump("hello") == {:ok, "hello"}
|
||||
assert JsonAny.dump("") == {:ok, ""}
|
||||
end
|
||||
|
||||
test "dumps integers" do
|
||||
assert JsonAny.dump(0) == {:ok, 0}
|
||||
assert JsonAny.dump(42) == {:ok, 42}
|
||||
assert JsonAny.dump(-100) == {:ok, -100}
|
||||
end
|
||||
|
||||
test "dumps floats" do
|
||||
assert JsonAny.dump(0.0) == {:ok, 0.0}
|
||||
assert JsonAny.dump(3.14) == {:ok, 3.14}
|
||||
assert JsonAny.dump(-99.99) == {:ok, -99.99}
|
||||
end
|
||||
|
||||
test "dumps booleans" do
|
||||
assert JsonAny.dump(true) == {:ok, true}
|
||||
assert JsonAny.dump(false) == {:ok, false}
|
||||
end
|
||||
|
||||
test "dumps lists" do
|
||||
assert JsonAny.dump([]) == {:ok, []}
|
||||
assert JsonAny.dump([1, 2, 3]) == {:ok, [1, 2, 3]}
|
||||
assert JsonAny.dump(["a", "b"]) == {:ok, ["a", "b"]}
|
||||
end
|
||||
|
||||
test "dumps maps" do
|
||||
assert JsonAny.dump(%{}) == {:ok, %{}}
|
||||
assert JsonAny.dump(%{key: "value"}) == {:ok, %{key: "value"}}
|
||||
|
||||
assert JsonAny.dump(%{"nested" => %{"map" => true}}) ==
|
||||
{:ok, %{"nested" => %{"map" => true}}}
|
||||
end
|
||||
|
||||
test "dumps complex nested structures" do
|
||||
value = %{"users" => [%{"name" => "Alice", "age" => 30}, %{"name" => "Bob", "age" => 25}]}
|
||||
assert JsonAny.dump(value) == {:ok, value}
|
||||
end
|
||||
|
||||
test "returns error for unsupported types" do
|
||||
assert JsonAny.dump({:tuple, "not supported"}) == :error
|
||||
assert JsonAny.dump(self()) == :error
|
||||
assert JsonAny.dump(make_ref()) == :error
|
||||
assert JsonAny.dump(fn -> :ok end) == :error
|
||||
end
|
||||
end
|
||||
|
||||
describe "cast -> load -> dump roundtrip" do
|
||||
test "nil roundtrips correctly" do
|
||||
{:ok, casted} = JsonAny.cast(nil)
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == nil
|
||||
end
|
||||
|
||||
test "string roundtrips correctly" do
|
||||
{:ok, casted} = JsonAny.cast("hello")
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == "hello"
|
||||
end
|
||||
|
||||
test "number roundtrips correctly" do
|
||||
{:ok, casted} = JsonAny.cast(42)
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == 42
|
||||
end
|
||||
|
||||
test "boolean roundtrips correctly" do
|
||||
{:ok, casted} = JsonAny.cast(true)
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == true
|
||||
end
|
||||
|
||||
test "list roundtrips correctly" do
|
||||
{:ok, casted} = JsonAny.cast([1, 2, 3])
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == [1, 2, 3]
|
||||
end
|
||||
|
||||
test "map roundtrips correctly" do
|
||||
{:ok, casted} = JsonAny.cast(%{key: "value"})
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == %{key: "value"}
|
||||
end
|
||||
|
||||
test "complex structure roundtrips correctly" do
|
||||
value = %{"users" => [%{"name" => "Alice", "active" => true, "score" => 95.5}]}
|
||||
{:ok, casted} = JsonAny.cast(value)
|
||||
{:ok, loaded} = JsonAny.load(casted)
|
||||
{:ok, dumped} = JsonAny.dump(loaded)
|
||||
assert dumped == value
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -3,6 +3,7 @@ defmodule Towerops.OrganizationsTest do
|
|||
|
||||
import Towerops.AccountsFixtures
|
||||
|
||||
alias Towerops.Devices.Device
|
||||
alias Towerops.Organizations
|
||||
|
||||
describe "organizations" do
|
||||
|
|
@ -372,6 +373,23 @@ defmodule Towerops.OrganizationsTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "user_has_access?/2" do
|
||||
test "returns true when user is member of organization" do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
assert Organizations.user_has_access?(user.id, organization.id)
|
||||
end
|
||||
|
||||
test "returns false when user is not member of organization" do
|
||||
user1 = user_fixture()
|
||||
user2 = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user1.id)
|
||||
|
||||
refute Organizations.user_has_access?(user2.id, organization.id)
|
||||
end
|
||||
end
|
||||
|
||||
describe "authorization" do
|
||||
setup do
|
||||
user = user_fixture()
|
||||
|
|
@ -391,4 +409,188 @@ defmodule Towerops.OrganizationsTest do
|
|||
refute Organizations.can?(member_membership, :delete, :organization)
|
||||
end
|
||||
end
|
||||
|
||||
describe "apply_snmp_config_to_all_equipment/1" do
|
||||
test "updates SNMP config for all devices in organization" do
|
||||
user = user_fixture()
|
||||
|
||||
{:ok, organization} =
|
||||
Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, organization} =
|
||||
Organizations.update_organization(organization, %{
|
||||
snmp_version: "3",
|
||||
snmp_community: "new-community"
|
||||
})
|
||||
|
||||
{:ok, site} =
|
||||
Towerops.Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device1} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 1",
|
||||
ip_address: "192.168.1.1",
|
||||
snmp_enabled: true,
|
||||
snmp_version: "2c",
|
||||
snmp_community: "public",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{:ok, device2} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 2",
|
||||
ip_address: "192.168.1.2",
|
||||
snmp_enabled: true,
|
||||
snmp_version: "2c",
|
||||
snmp_community: "public",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{count, _} = Organizations.apply_snmp_config_to_all_equipment(organization.id)
|
||||
assert count == 2
|
||||
|
||||
updated_device1 = Repo.get!(Device, device1.id)
|
||||
assert updated_device1.snmp_version == "3"
|
||||
assert updated_device1.snmp_community == "new-community"
|
||||
|
||||
updated_device2 = Repo.get!(Device, device2.id)
|
||||
assert updated_device2.snmp_version == "3"
|
||||
assert updated_device2.snmp_community == "new-community"
|
||||
end
|
||||
|
||||
test "returns 0 when organization has no devices" do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{count, _} = Organizations.apply_snmp_config_to_all_equipment(organization.id)
|
||||
assert count == 0
|
||||
end
|
||||
end
|
||||
|
||||
describe "apply_agent_to_all_equipment/1" do
|
||||
test "creates agent assignments for all devices when default agent is set" do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, agent_token, _token} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
|
||||
|
||||
{:ok, organization} =
|
||||
Organizations.update_organization(organization, %{
|
||||
default_agent_token_id: agent_token.id
|
||||
})
|
||||
|
||||
{:ok, site} =
|
||||
Towerops.Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device1} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 1",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{:ok, device2} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 2",
|
||||
ip_address: "192.168.1.2",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{count, _} = Organizations.apply_agent_to_all_equipment(organization.id)
|
||||
assert count == 2
|
||||
|
||||
assignment1 = Towerops.Agents.get_device_assignment(device1.id)
|
||||
assert assignment1.agent_token_id == agent_token.id
|
||||
|
||||
assignment2 = Towerops.Agents.get_device_assignment(device2.id)
|
||||
assert assignment2.agent_token_id == agent_token.id
|
||||
end
|
||||
|
||||
test "deletes agent assignments when default agent is nil" do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, site} =
|
||||
Towerops.Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Test Device",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{:ok, agent_token, _token} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
|
||||
|
||||
Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
|
||||
|
||||
# Organization has no default agent set (nil)
|
||||
Organizations.apply_agent_to_all_equipment(organization.id)
|
||||
|
||||
assert Towerops.Agents.get_device_assignment(device.id) == nil
|
||||
end
|
||||
|
||||
test "replaces existing agent assignments" do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, old_agent, _token1} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Old Agent")
|
||||
|
||||
{:ok, new_agent, _token2} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "New Agent")
|
||||
|
||||
{:ok, site} =
|
||||
Towerops.Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Test Device",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
Towerops.Agents.assign_device_to_agent(old_agent.id, device.id)
|
||||
|
||||
{:ok, organization} =
|
||||
Organizations.update_organization(organization, %{
|
||||
default_agent_token_id: new_agent.id
|
||||
})
|
||||
|
||||
Organizations.apply_agent_to_all_equipment(organization.id)
|
||||
|
||||
assignment = Towerops.Agents.get_device_assignment(device.id)
|
||||
assert assignment.agent_token_id == new_agent.id
|
||||
end
|
||||
|
||||
test "returns 0 when organization has no devices" do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, agent_token, _token} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
|
||||
|
||||
{:ok, organization} =
|
||||
Organizations.update_organization(organization, %{
|
||||
default_agent_token_id: agent_token.id
|
||||
})
|
||||
|
||||
{count, _} = Organizations.apply_agent_to_all_equipment(organization.id)
|
||||
assert count == 0
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
100
test/towerops/profiles/device_oid_test.exs
Normal file
100
test/towerops/profiles/device_oid_test.exs
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
defmodule Towerops.Profiles.DeviceOidTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Profiles
|
||||
alias Towerops.Profiles.DeviceOid
|
||||
|
||||
describe "changeset/2" do
|
||||
setup do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test Profile", vendor: "Test"})
|
||||
%{profile: profile}
|
||||
end
|
||||
|
||||
test "valid changeset with all required fields", %{profile: profile} do
|
||||
attrs = %{
|
||||
field: "serial_number",
|
||||
mib_name: "TEST-MIB::serialNumber.0",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = DeviceOid.changeset(%DeviceOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "invalid changeset without field", %{profile: profile} do
|
||||
attrs = %{
|
||||
mib_name: "TEST-MIB::serialNumber.0",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = DeviceOid.changeset(%DeviceOid{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).field
|
||||
end
|
||||
|
||||
test "invalid changeset without mib_name", %{profile: profile} do
|
||||
attrs = %{
|
||||
field: "serial_number",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = DeviceOid.changeset(%DeviceOid{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).mib_name
|
||||
end
|
||||
|
||||
test "invalid changeset without profile_id" do
|
||||
attrs = %{
|
||||
field: "serial_number",
|
||||
mib_name: "TEST-MIB::serialNumber.0"
|
||||
}
|
||||
|
||||
changeset = DeviceOid.changeset(%DeviceOid{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).profile_id
|
||||
end
|
||||
|
||||
test "valid changeset with various field names", %{profile: profile} do
|
||||
fields = [
|
||||
"serial_number",
|
||||
"firmware_version",
|
||||
"hardware_version",
|
||||
"model",
|
||||
"uptime",
|
||||
"location"
|
||||
]
|
||||
|
||||
for field <- fields do
|
||||
attrs = %{
|
||||
field: field,
|
||||
mib_name: "TEST-MIB::#{field}.0",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = DeviceOid.changeset(%DeviceOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
end
|
||||
|
||||
test "valid changeset with various MIB name formats", %{profile: profile} do
|
||||
mib_names = [
|
||||
"SNMPv2-MIB::sysDescr.0",
|
||||
"MIKROTIK-MIB::mtxrSerialNumber.0",
|
||||
"CISCO-PRODUCTS-MIB::ciscoProductId",
|
||||
"1.3.6.1.2.1.1.1.0",
|
||||
".1.3.6.1.2.1.1.1.0"
|
||||
]
|
||||
|
||||
for mib_name <- mib_names do
|
||||
attrs = %{
|
||||
field: "test_field",
|
||||
mib_name: mib_name,
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = DeviceOid.changeset(%DeviceOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
130
test/towerops/profiles/device_profile_test.exs
Normal file
130
test/towerops/profiles/device_profile_test.exs
Normal file
|
|
@ -0,0 +1,130 @@
|
|||
defmodule Towerops.Profiles.DeviceProfileTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Profiles.DeviceProfile
|
||||
|
||||
describe "changeset/2" do
|
||||
test "valid changeset with all fields" do
|
||||
attrs = %{
|
||||
name: "Test Profile",
|
||||
vendor: "Test Vendor",
|
||||
detection_pattern: "Test.*Device",
|
||||
detection_oid: "1.3.6.1.4.1.12345",
|
||||
priority: 50,
|
||||
enabled: true
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with minimal required fields" do
|
||||
attrs = %{
|
||||
name: "Minimal Profile",
|
||||
vendor: "Test"
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_change(changeset, :priority) == nil
|
||||
assert get_field(changeset, :priority) == 100
|
||||
assert get_field(changeset, :enabled) == true
|
||||
end
|
||||
|
||||
test "invalid changeset without name" do
|
||||
attrs = %{vendor: "Test"}
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).name
|
||||
end
|
||||
|
||||
test "invalid changeset without vendor" do
|
||||
attrs = %{name: "Test"}
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).vendor
|
||||
end
|
||||
|
||||
test "invalid changeset with empty name" do
|
||||
attrs = %{name: "", vendor: "Test"}
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
refute changeset.valid?
|
||||
end
|
||||
|
||||
test "invalid changeset with empty vendor" do
|
||||
attrs = %{name: "Test", vendor: ""}
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
refute changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with nil optional fields" do
|
||||
attrs = %{
|
||||
name: "Test",
|
||||
vendor: "Test",
|
||||
detection_pattern: nil,
|
||||
detection_oid: nil
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with detection_pattern only" do
|
||||
attrs = %{
|
||||
name: "Test",
|
||||
vendor: "Test",
|
||||
detection_pattern: "Cisco.*"
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with detection_oid only" do
|
||||
attrs = %{
|
||||
name: "Test",
|
||||
vendor: "Test",
|
||||
detection_oid: "1.3.6.1.4.1.9"
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with both detection methods" do
|
||||
attrs = %{
|
||||
name: "Test",
|
||||
vendor: "Test",
|
||||
detection_pattern: "Cisco.*",
|
||||
detection_oid: "1.3.6.1.4.1.9"
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with custom priority" do
|
||||
attrs = %{
|
||||
name: "Test",
|
||||
vendor: "Test",
|
||||
priority: 1
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_change(changeset, :priority) == 1
|
||||
end
|
||||
|
||||
test "valid changeset with enabled false" do
|
||||
attrs = %{
|
||||
name: "Test",
|
||||
vendor: "Test",
|
||||
enabled: false
|
||||
}
|
||||
|
||||
changeset = DeviceProfile.changeset(%DeviceProfile{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_change(changeset, :enabled) == false
|
||||
end
|
||||
end
|
||||
end
|
||||
161
test/towerops/profiles/sensor_oid_test.exs
Normal file
161
test/towerops/profiles/sensor_oid_test.exs
Normal file
|
|
@ -0,0 +1,161 @@
|
|||
defmodule Towerops.Profiles.SensorOidTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Profiles
|
||||
alias Towerops.Profiles.SensorOid
|
||||
|
||||
describe "changeset/2" do
|
||||
setup do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test Profile", vendor: "Test"})
|
||||
%{profile: profile}
|
||||
end
|
||||
|
||||
test "valid changeset with all fields", %{profile: profile} do
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temperature.0",
|
||||
sensor_descr: "CPU Temperature",
|
||||
sensor_unit: "celsius",
|
||||
sensor_divisor: 10,
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with minimal required fields", %{profile: profile} do
|
||||
attrs = %{
|
||||
sensor_type: "voltage",
|
||||
mib_name: "TEST-MIB::voltage.0",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_field(changeset, :sensor_divisor) == 1
|
||||
end
|
||||
|
||||
test "invalid changeset without sensor_type", %{profile: profile} do
|
||||
attrs = %{
|
||||
mib_name: "TEST-MIB::temperature.0",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).sensor_type
|
||||
end
|
||||
|
||||
test "invalid changeset without mib_name", %{profile: profile} do
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).mib_name
|
||||
end
|
||||
|
||||
test "invalid changeset without profile_id" do
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temperature.0"
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).profile_id
|
||||
end
|
||||
|
||||
test "valid changeset with various sensor types", %{profile: profile} do
|
||||
sensor_types = [
|
||||
"temperature",
|
||||
"voltage",
|
||||
"current",
|
||||
"power",
|
||||
"fanspeed",
|
||||
"humidity",
|
||||
"pressure"
|
||||
]
|
||||
|
||||
for sensor_type <- sensor_types do
|
||||
attrs = %{
|
||||
sensor_type: sensor_type,
|
||||
mib_name: "TEST-MIB::#{sensor_type}.0",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
end
|
||||
|
||||
test "valid changeset with various sensor units", %{profile: profile} do
|
||||
units = [
|
||||
{"temperature", "celsius"},
|
||||
{"temperature", "fahrenheit"},
|
||||
{"voltage", "volts"},
|
||||
{"current", "amperes"},
|
||||
{"power", "watts"},
|
||||
{"fanspeed", "rpm"}
|
||||
]
|
||||
|
||||
for {sensor_type, unit} <- units do
|
||||
attrs = %{
|
||||
sensor_type: sensor_type,
|
||||
mib_name: "TEST-MIB::test.0",
|
||||
sensor_unit: unit,
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
end
|
||||
|
||||
test "valid changeset with various divisors", %{profile: profile} do
|
||||
divisors = [1, 10, 100, 1000]
|
||||
|
||||
for divisor <- divisors do
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temp.0",
|
||||
sensor_divisor: divisor,
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_field(changeset, :sensor_divisor) == divisor
|
||||
end
|
||||
end
|
||||
|
||||
test "valid changeset with nil optional fields", %{profile: profile} do
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temp.0",
|
||||
sensor_descr: nil,
|
||||
sensor_unit: nil,
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with sensor description", %{profile: profile} do
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temp.0",
|
||||
sensor_descr: "Board Temperature Sensor #1",
|
||||
profile_id: profile.id
|
||||
}
|
||||
|
||||
changeset = SensorOid.changeset(%SensorOid{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_field(changeset, :sensor_descr) == "Board Temperature Sensor #1"
|
||||
end
|
||||
end
|
||||
end
|
||||
346
test/towerops/profiles_test.exs
Normal file
346
test/towerops/profiles_test.exs
Normal file
|
|
@ -0,0 +1,346 @@
|
|||
defmodule Towerops.ProfilesTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
alias Towerops.Profiles
|
||||
alias Towerops.Profiles.DeviceProfile
|
||||
|
||||
describe "list_profiles/0" do
|
||||
test "returns all enabled profiles ordered by priority" do
|
||||
{:ok, profile1} =
|
||||
Profiles.create_profile(%{
|
||||
name: "High Priority",
|
||||
vendor: "Cisco",
|
||||
priority: 10,
|
||||
enabled: true
|
||||
})
|
||||
|
||||
{:ok, profile2} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Low Priority",
|
||||
vendor: "MikroTik",
|
||||
priority: 100,
|
||||
enabled: true
|
||||
})
|
||||
|
||||
{:ok, _profile3} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Disabled",
|
||||
vendor: "Ubiquiti",
|
||||
priority: 50,
|
||||
enabled: false
|
||||
})
|
||||
|
||||
profiles = Profiles.list_profiles()
|
||||
|
||||
assert length(profiles) == 2
|
||||
assert hd(profiles).id == profile1.id
|
||||
assert Enum.at(profiles, 1).id == profile2.id
|
||||
end
|
||||
|
||||
test "returns empty list when no enabled profiles" do
|
||||
{:ok, _profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Disabled",
|
||||
vendor: "Test",
|
||||
enabled: false
|
||||
})
|
||||
|
||||
assert Profiles.list_profiles() == []
|
||||
end
|
||||
end
|
||||
|
||||
describe "get_profile/1" do
|
||||
test "returns profile by name with associations" do
|
||||
{:ok, profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Test Profile",
|
||||
vendor: "Test Vendor"
|
||||
})
|
||||
|
||||
{:ok, _device_oid} =
|
||||
Profiles.add_device_oid(profile.id, "serial_number", "TEST-MIB::serialNumber.0")
|
||||
|
||||
{:ok, _sensor_oid} =
|
||||
Profiles.add_sensor_oid(profile.id, %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temperature.0"
|
||||
})
|
||||
|
||||
retrieved = Profiles.get_profile("Test Profile")
|
||||
|
||||
assert retrieved.id == profile.id
|
||||
assert length(retrieved.device_oids) == 1
|
||||
assert length(retrieved.sensor_oids) == 1
|
||||
end
|
||||
|
||||
test "returns nil when profile doesn't exist" do
|
||||
assert Profiles.get_profile("Nonexistent") == nil
|
||||
end
|
||||
end
|
||||
|
||||
describe "match_profile/1" do
|
||||
setup do
|
||||
{:ok, regex_profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Regex Match",
|
||||
vendor: "Cisco",
|
||||
detection_pattern: "Cisco IOS",
|
||||
priority: 10
|
||||
})
|
||||
|
||||
{:ok, oid_profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "OID Match",
|
||||
vendor: "MikroTik",
|
||||
detection_oid: "1.3.6.1.4.1.14988",
|
||||
priority: 20
|
||||
})
|
||||
|
||||
{:ok, both_profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Both Match",
|
||||
vendor: "Ubiquiti",
|
||||
detection_pattern: "EdgeRouter",
|
||||
detection_oid: "1.3.6.1.4.1.41112",
|
||||
priority: 30
|
||||
})
|
||||
|
||||
%{
|
||||
regex_profile: regex_profile,
|
||||
oid_profile: oid_profile,
|
||||
both_profile: both_profile
|
||||
}
|
||||
end
|
||||
|
||||
test "matches by detection_pattern regex", %{regex_profile: profile} do
|
||||
system_info = %{
|
||||
sys_descr: "Cisco IOS Software, Version 15.2",
|
||||
sys_object_id: "1.3.6.1.4.1.9"
|
||||
}
|
||||
|
||||
assert Profiles.match_profile(system_info).id == profile.id
|
||||
end
|
||||
|
||||
test "matches by detection_oid prefix", %{oid_profile: profile} do
|
||||
system_info = %{
|
||||
sys_descr: "Some description",
|
||||
sys_object_id: "1.3.6.1.4.1.14988.1.1"
|
||||
}
|
||||
|
||||
assert Profiles.match_profile(system_info).id == profile.id
|
||||
end
|
||||
|
||||
test "matches higher priority profile first", %{regex_profile: profile} do
|
||||
system_info = %{
|
||||
sys_descr: "Cisco IOS Software",
|
||||
sys_object_id: "1.3.6.1.4.1.14988"
|
||||
}
|
||||
|
||||
# Both profiles match, but regex_profile has higher priority (10 < 20)
|
||||
assert Profiles.match_profile(system_info).id == profile.id
|
||||
end
|
||||
|
||||
test "returns nil when no profile matches" do
|
||||
system_info = %{
|
||||
sys_descr: "Unknown Device",
|
||||
sys_object_id: "1.2.3.4.5"
|
||||
}
|
||||
|
||||
assert Profiles.match_profile(system_info) == nil
|
||||
end
|
||||
|
||||
test "handles missing sys_descr" do
|
||||
{:ok, profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "OID Only",
|
||||
vendor: "Test",
|
||||
detection_oid: "1.2.3.4"
|
||||
})
|
||||
|
||||
system_info = %{sys_object_id: "1.2.3.4.5"}
|
||||
|
||||
assert Profiles.match_profile(system_info).id == profile.id
|
||||
end
|
||||
|
||||
test "handles missing sys_object_id" do
|
||||
{:ok, profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Regex Only",
|
||||
vendor: "Test",
|
||||
detection_pattern: "Test Device"
|
||||
})
|
||||
|
||||
system_info = %{sys_descr: "Test Device v1.0"}
|
||||
|
||||
assert Profiles.match_profile(system_info).id == profile.id
|
||||
end
|
||||
|
||||
test "handles invalid regex pattern gracefully" do
|
||||
{:ok, _profile} =
|
||||
Profiles.create_profile(%{
|
||||
name: "Invalid Regex",
|
||||
vendor: "Test",
|
||||
detection_pattern: "[invalid(regex"
|
||||
})
|
||||
|
||||
system_info = %{sys_descr: "Some description"}
|
||||
|
||||
# Should not crash, just not match
|
||||
assert Profiles.match_profile(system_info) == nil
|
||||
end
|
||||
end
|
||||
|
||||
describe "create_profile/1" do
|
||||
test "creates profile with valid attributes" do
|
||||
attrs = %{
|
||||
name: "New Profile",
|
||||
vendor: "Test Vendor",
|
||||
detection_pattern: "Test.*Device",
|
||||
detection_oid: "1.3.6.1.4.1.12345",
|
||||
priority: 50,
|
||||
enabled: true
|
||||
}
|
||||
|
||||
assert {:ok, %DeviceProfile{} = profile} = Profiles.create_profile(attrs)
|
||||
assert profile.name == "New Profile"
|
||||
assert profile.vendor == "Test Vendor"
|
||||
assert profile.detection_pattern == "Test.*Device"
|
||||
assert profile.detection_oid == "1.3.6.1.4.1.12345"
|
||||
assert profile.priority == 50
|
||||
assert profile.enabled == true
|
||||
end
|
||||
|
||||
test "creates profile with minimal attributes" do
|
||||
attrs = %{
|
||||
name: "Minimal Profile",
|
||||
vendor: "Test"
|
||||
}
|
||||
|
||||
assert {:ok, %DeviceProfile{} = profile} = Profiles.create_profile(attrs)
|
||||
assert profile.name == "Minimal Profile"
|
||||
assert profile.vendor == "Test"
|
||||
assert profile.priority == 100
|
||||
assert profile.enabled == true
|
||||
end
|
||||
|
||||
test "returns error with missing required fields" do
|
||||
assert {:error, changeset} = Profiles.create_profile(%{})
|
||||
assert "can't be blank" in errors_on(changeset).name
|
||||
assert "can't be blank" in errors_on(changeset).vendor
|
||||
end
|
||||
|
||||
test "returns error with duplicate name" do
|
||||
{:ok, _profile} = Profiles.create_profile(%{name: "Duplicate", vendor: "Test"})
|
||||
|
||||
assert {:error, changeset} = Profiles.create_profile(%{name: "Duplicate", vendor: "Test"})
|
||||
assert "has already been taken" in errors_on(changeset).name
|
||||
end
|
||||
end
|
||||
|
||||
describe "update_profile/2" do
|
||||
test "updates profile with valid attributes" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Original", vendor: "Original Vendor"})
|
||||
|
||||
assert {:ok, updated} =
|
||||
Profiles.update_profile(profile, %{
|
||||
name: "Updated",
|
||||
vendor: "Updated Vendor",
|
||||
priority: 25
|
||||
})
|
||||
|
||||
assert updated.name == "Updated"
|
||||
assert updated.vendor == "Updated Vendor"
|
||||
assert updated.priority == 25
|
||||
end
|
||||
|
||||
test "returns error with invalid attributes" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test", vendor: "Test"})
|
||||
|
||||
assert {:error, changeset} = Profiles.update_profile(profile, %{name: nil})
|
||||
assert "can't be blank" in errors_on(changeset).name
|
||||
end
|
||||
end
|
||||
|
||||
describe "delete_profile/1" do
|
||||
test "deletes profile" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "To Delete", vendor: "Test"})
|
||||
|
||||
assert {:ok, %DeviceProfile{}} = Profiles.delete_profile(profile)
|
||||
assert Profiles.get_profile("To Delete") == nil
|
||||
end
|
||||
end
|
||||
|
||||
describe "add_device_oid/3" do
|
||||
test "adds device OID to profile" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test", vendor: "Test"})
|
||||
|
||||
assert {:ok, device_oid} =
|
||||
Profiles.add_device_oid(
|
||||
profile.id,
|
||||
"serial_number",
|
||||
"TEST-MIB::serialNumber.0"
|
||||
)
|
||||
|
||||
assert device_oid.field == "serial_number"
|
||||
assert device_oid.mib_name == "TEST-MIB::serialNumber.0"
|
||||
assert device_oid.profile_id == profile.id
|
||||
end
|
||||
|
||||
test "returns error with duplicate field for same profile" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test", vendor: "Test"})
|
||||
|
||||
{:ok, _oid1} = Profiles.add_device_oid(profile.id, "serial_number", "MIB1::sn.0")
|
||||
|
||||
assert {:error, changeset} =
|
||||
Profiles.add_device_oid(profile.id, "serial_number", "MIB2::sn.0")
|
||||
|
||||
assert "has already been taken" in errors_on(changeset).profile_id
|
||||
end
|
||||
end
|
||||
|
||||
describe "add_sensor_oid/2" do
|
||||
test "adds sensor OID to profile" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test", vendor: "Test"})
|
||||
|
||||
attrs = %{
|
||||
sensor_type: "temperature",
|
||||
mib_name: "TEST-MIB::temperature.0",
|
||||
sensor_descr: "CPU Temperature",
|
||||
sensor_unit: "celsius",
|
||||
sensor_divisor: 10
|
||||
}
|
||||
|
||||
assert {:ok, sensor_oid} = Profiles.add_sensor_oid(profile.id, attrs)
|
||||
|
||||
assert sensor_oid.sensor_type == "temperature"
|
||||
assert sensor_oid.mib_name == "TEST-MIB::temperature.0"
|
||||
assert sensor_oid.sensor_descr == "CPU Temperature"
|
||||
assert sensor_oid.sensor_unit == "celsius"
|
||||
assert sensor_oid.sensor_divisor == 10
|
||||
assert sensor_oid.profile_id == profile.id
|
||||
end
|
||||
|
||||
test "adds sensor OID with minimal attributes" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test", vendor: "Test"})
|
||||
|
||||
attrs = %{
|
||||
sensor_type: "voltage",
|
||||
mib_name: "TEST-MIB::voltage.0"
|
||||
}
|
||||
|
||||
assert {:ok, sensor_oid} = Profiles.add_sensor_oid(profile.id, attrs)
|
||||
|
||||
assert sensor_oid.sensor_type == "voltage"
|
||||
assert sensor_oid.mib_name == "TEST-MIB::voltage.0"
|
||||
assert sensor_oid.sensor_divisor == 1
|
||||
end
|
||||
|
||||
test "returns error with missing required fields" do
|
||||
{:ok, profile} = Profiles.create_profile(%{name: "Test", vendor: "Test"})
|
||||
|
||||
assert {:error, changeset} = Profiles.add_sensor_oid(profile.id, %{})
|
||||
assert "can't be blank" in errors_on(changeset).sensor_type
|
||||
assert "can't be blank" in errors_on(changeset).mib_name
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
defmodule Towerops.SitesTest do
|
||||
use Towerops.DataCase
|
||||
|
||||
alias Towerops.Devices.Device
|
||||
alias Towerops.Sites
|
||||
|
||||
describe "sites" do
|
||||
|
|
@ -27,6 +28,18 @@ defmodule Towerops.SitesTest do
|
|||
assert hd(result).id == site.id
|
||||
end
|
||||
|
||||
test "count_organization_sites/1 returns count of sites", %{organization: organization} do
|
||||
{:ok, _site1} = Sites.create_site(%{name: "Site 1", organization_id: organization.id})
|
||||
{:ok, _site2} = Sites.create_site(%{name: "Site 2", organization_id: organization.id})
|
||||
{:ok, _site3} = Sites.create_site(%{name: "Site 3", organization_id: organization.id})
|
||||
|
||||
assert Sites.count_organization_sites(organization.id) == 3
|
||||
end
|
||||
|
||||
test "count_organization_sites/1 returns 0 when no sites exist", %{organization: organization} do
|
||||
assert Sites.count_organization_sites(organization.id) == 0
|
||||
end
|
||||
|
||||
test "list_root_sites/1 returns only root sites without parents", %{organization: organization} do
|
||||
{:ok, root1} =
|
||||
Sites.create_site(%{name: "Root 1", organization_id: organization.id})
|
||||
|
|
@ -182,4 +195,178 @@ defmodule Towerops.SitesTest do
|
|||
assert length(root.children) == 2
|
||||
end
|
||||
end
|
||||
|
||||
describe "bulk operations" do
|
||||
import Towerops.AccountsFixtures
|
||||
|
||||
setup do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
%{organization: organization}
|
||||
end
|
||||
|
||||
test "apply_snmp_config_to_all_equipment/1 updates all devices at site", %{
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, site} =
|
||||
Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id,
|
||||
snmp_version: "3",
|
||||
snmp_community: "site-community"
|
||||
})
|
||||
|
||||
{:ok, device1} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 1",
|
||||
ip_address: "192.168.1.1",
|
||||
snmp_enabled: true,
|
||||
snmp_version: "2c",
|
||||
snmp_community: "public",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{:ok, device2} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 2",
|
||||
ip_address: "192.168.1.2",
|
||||
snmp_enabled: true,
|
||||
snmp_version: "2c",
|
||||
snmp_community: "public",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{count, _} = Sites.apply_snmp_config_to_all_equipment(site.id)
|
||||
assert count == 2
|
||||
|
||||
updated_device1 = Repo.get!(Device, device1.id)
|
||||
assert updated_device1.snmp_version == "3"
|
||||
assert updated_device1.snmp_community == "site-community"
|
||||
|
||||
updated_device2 = Repo.get!(Device, device2.id)
|
||||
assert updated_device2.snmp_version == "3"
|
||||
assert updated_device2.snmp_community == "site-community"
|
||||
end
|
||||
|
||||
test "apply_snmp_config_to_all_equipment/1 returns 0 when site has no devices", %{
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, site} = Sites.create_site(%{name: "Empty Site", organization_id: organization.id})
|
||||
|
||||
{count, _} = Sites.apply_snmp_config_to_all_equipment(site.id)
|
||||
assert count == 0
|
||||
end
|
||||
|
||||
test "apply_agent_to_all_equipment/1 creates agent assignments for all devices", %{
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, agent_token, _token} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
|
||||
|
||||
{:ok, site} =
|
||||
Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id,
|
||||
agent_token_id: agent_token.id
|
||||
})
|
||||
|
||||
{:ok, device1} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 1",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{:ok, device2} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Device 2",
|
||||
ip_address: "192.168.1.2",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
{count, _} = Sites.apply_agent_to_all_equipment(site.id)
|
||||
assert count == 2
|
||||
|
||||
assignment1 = Towerops.Agents.get_device_assignment(device1.id)
|
||||
assert assignment1.agent_token_id == agent_token.id
|
||||
|
||||
assignment2 = Towerops.Agents.get_device_assignment(device2.id)
|
||||
assert assignment2.agent_token_id == agent_token.id
|
||||
end
|
||||
|
||||
test "apply_agent_to_all_equipment/1 deletes assignments when agent is nil", %{
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, agent_token, _token} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
|
||||
|
||||
{:ok, site} =
|
||||
Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Test Device",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
Towerops.Agents.assign_device_to_agent(agent_token.id, device.id)
|
||||
|
||||
# Site has no agent set (nil)
|
||||
Sites.apply_agent_to_all_equipment(site.id)
|
||||
|
||||
assert Towerops.Agents.get_device_assignment(device.id) == nil
|
||||
end
|
||||
|
||||
test "apply_agent_to_all_equipment/1 replaces existing assignments", %{
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, old_agent, _token1} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Old Agent")
|
||||
|
||||
{:ok, new_agent, _token2} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "New Agent")
|
||||
|
||||
{:ok, site} =
|
||||
Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id,
|
||||
agent_token_id: new_agent.id
|
||||
})
|
||||
|
||||
{:ok, device} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Test Device",
|
||||
ip_address: "192.168.1.1",
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
Towerops.Agents.assign_device_to_agent(old_agent.id, device.id)
|
||||
|
||||
Sites.apply_agent_to_all_equipment(site.id)
|
||||
|
||||
assignment = Towerops.Agents.get_device_assignment(device.id)
|
||||
assert assignment.agent_token_id == new_agent.id
|
||||
end
|
||||
|
||||
test "apply_agent_to_all_equipment/1 returns 0 when site has no devices", %{
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, agent_token, _token} =
|
||||
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
|
||||
|
||||
{:ok, site} =
|
||||
Sites.create_site(%{
|
||||
name: "Empty Site",
|
||||
organization_id: organization.id,
|
||||
agent_token_id: agent_token.id
|
||||
})
|
||||
|
||||
{count, _} = Sites.apply_agent_to_all_equipment(site.id)
|
||||
assert count == 0
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
357
test/towerops/snmp/neighbor_test.exs
Normal file
357
test/towerops/snmp/neighbor_test.exs
Normal file
|
|
@ -0,0 +1,357 @@
|
|||
defmodule Towerops.Snmp.NeighborTest do
|
||||
use Towerops.DataCase, async: true
|
||||
|
||||
import Towerops.AccountsFixtures
|
||||
|
||||
alias Towerops.Snmp.Device
|
||||
alias Towerops.Snmp.Interface
|
||||
alias Towerops.Snmp.Neighbor
|
||||
|
||||
setup do
|
||||
user = user_fixture()
|
||||
|
||||
{:ok, organization} =
|
||||
Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
{:ok, site} =
|
||||
Towerops.Sites.create_site(%{
|
||||
name: "Test Site",
|
||||
organization_id: organization.id
|
||||
})
|
||||
|
||||
{:ok, device_schema} =
|
||||
Towerops.Devices.create_device(%{
|
||||
name: "Test Router",
|
||||
ip_address: "192.168.1.1",
|
||||
snmp_enabled: true,
|
||||
snmp_version: "2c",
|
||||
snmp_community: "public",
|
||||
snmp_port: 161,
|
||||
site_id: site.id
|
||||
})
|
||||
|
||||
snmp_device =
|
||||
%Device{}
|
||||
|> Device.changeset(%{
|
||||
device_id: device_schema.id,
|
||||
sys_name: "test-router",
|
||||
sys_descr: "Test Router"
|
||||
})
|
||||
|> Repo.insert!()
|
||||
|
||||
interface =
|
||||
%Interface{}
|
||||
|> Interface.changeset(%{
|
||||
snmp_device_id: snmp_device.id,
|
||||
if_index: 1,
|
||||
if_descr: "GigabitEthernet0/0",
|
||||
if_name: "Gi0/0",
|
||||
if_type: 6,
|
||||
if_oper_status: "up"
|
||||
})
|
||||
|> Repo.insert!()
|
||||
|
||||
%{device: device_schema, interface: interface}
|
||||
end
|
||||
|
||||
describe "changeset/2" do
|
||||
test "valid changeset with all required fields", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with all fields", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "cdp",
|
||||
remote_chassis_id: "00:1a:2b:3c:4d:5e",
|
||||
remote_system_name: "neighbor-switch",
|
||||
remote_system_description: "Cisco IOS Software",
|
||||
remote_platform: "Cisco 2960",
|
||||
remote_port_id: "GigabitEthernet1/0/1",
|
||||
remote_port_description: "Downlink to router",
|
||||
remote_address: "192.168.1.2",
|
||||
remote_capabilities: ["bridge", "router"],
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_field(changeset, :remote_chassis_id) == "00:1a:2b:3c:4d:5e"
|
||||
assert get_field(changeset, :remote_capabilities) == ["bridge", "router"]
|
||||
end
|
||||
|
||||
test "invalid changeset without device_id", %{interface: interface} do
|
||||
attrs = %{
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).device_id
|
||||
end
|
||||
|
||||
test "invalid changeset without interface_id", %{device: device} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).interface_id
|
||||
end
|
||||
|
||||
test "invalid changeset without protocol", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).protocol
|
||||
end
|
||||
|
||||
test "invalid changeset without last_discovered_at", %{
|
||||
device: device,
|
||||
interface: interface
|
||||
} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp"
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).last_discovered_at
|
||||
end
|
||||
|
||||
test "invalid changeset with invalid protocol", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "invalid",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
refute changeset.valid?
|
||||
assert "is invalid" in errors_on(changeset).protocol
|
||||
end
|
||||
|
||||
test "valid changeset with 'lldp' protocol", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with 'cdp' protocol", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "cdp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "valid changeset with nil optional fields", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
remote_chassis_id: nil,
|
||||
remote_system_name: nil,
|
||||
remote_system_description: nil,
|
||||
remote_platform: nil,
|
||||
remote_port_id: nil,
|
||||
remote_port_description: nil,
|
||||
remote_address: nil,
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "default remote_capabilities is empty list", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_field(changeset, :remote_capabilities) == []
|
||||
end
|
||||
|
||||
test "normalizes binary UUID device_id to string", %{device: device, interface: interface} do
|
||||
# Create a 16-byte binary UUID
|
||||
{:ok, binary_uuid} = Ecto.UUID.dump(device.id)
|
||||
|
||||
attrs = %{
|
||||
device_id: binary_uuid,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
# The normalized UUID should be a string format
|
||||
assert get_change(changeset, :device_id) == device.id
|
||||
end
|
||||
|
||||
test "normalizes binary UUID interface_id to string", %{device: device, interface: interface} do
|
||||
# Create a 16-byte binary UUID
|
||||
{:ok, binary_uuid} = Ecto.UUID.dump(interface.id)
|
||||
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: binary_uuid,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
# The normalized UUID should be a string format
|
||||
assert get_change(changeset, :interface_id) == interface.id
|
||||
end
|
||||
|
||||
test "keeps string UUID unchanged", %{device: device, interface: interface} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert changeset.valid?
|
||||
assert get_change(changeset, :device_id) == device.id
|
||||
assert get_change(changeset, :interface_id) == interface.id
|
||||
end
|
||||
end
|
||||
|
||||
describe "unique constraint" do
|
||||
test "prevents duplicate neighbors with same interface, chassis_id, and protocol", %{
|
||||
device: device,
|
||||
interface: interface
|
||||
} do
|
||||
attrs = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
remote_chassis_id: "00:1a:2b:3c:4d:5e",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
# Insert first neighbor
|
||||
changeset = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
assert {:ok, _neighbor} = Repo.insert(changeset)
|
||||
|
||||
# Attempt to insert duplicate
|
||||
changeset2 = Neighbor.changeset(%Neighbor{}, attrs)
|
||||
|
||||
assert {:error, changeset} = Repo.insert(changeset2)
|
||||
assert "has already been taken" in errors_on(changeset).interface_id
|
||||
end
|
||||
|
||||
test "allows same chassis_id with different interface", %{
|
||||
device: device,
|
||||
interface: interface
|
||||
} do
|
||||
# Read the SNMP device to get its ID
|
||||
snmp_device = Repo.get_by!(Device, device_id: device.id)
|
||||
|
||||
# Create a second interface
|
||||
interface2 =
|
||||
%Interface{}
|
||||
|> Interface.changeset(%{
|
||||
snmp_device_id: snmp_device.id,
|
||||
if_index: 2,
|
||||
if_descr: "GigabitEthernet0/1",
|
||||
if_name: "Gi0/1",
|
||||
if_type: 6,
|
||||
if_oper_status: "up"
|
||||
})
|
||||
|> Repo.insert!()
|
||||
|
||||
attrs1 = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
remote_chassis_id: "00:1a:2b:3c:4d:5e",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
attrs2 = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface2.id,
|
||||
protocol: "lldp",
|
||||
remote_chassis_id: "00:1a:2b:3c:4d:5e",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
# Insert neighbors on different interfaces
|
||||
changeset1 = Neighbor.changeset(%Neighbor{}, attrs1)
|
||||
assert {:ok, _neighbor1} = Repo.insert(changeset1)
|
||||
|
||||
changeset2 = Neighbor.changeset(%Neighbor{}, attrs2)
|
||||
assert {:ok, _neighbor2} = Repo.insert(changeset2)
|
||||
end
|
||||
|
||||
test "allows same chassis_id and interface with different protocol", %{
|
||||
device: device,
|
||||
interface: interface
|
||||
} do
|
||||
attrs1 = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "lldp",
|
||||
remote_chassis_id: "00:1a:2b:3c:4d:5e",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
attrs2 = %{
|
||||
device_id: device.id,
|
||||
interface_id: interface.id,
|
||||
protocol: "cdp",
|
||||
remote_chassis_id: "00:1a:2b:3c:4d:5e",
|
||||
last_discovered_at: DateTime.utc_now()
|
||||
}
|
||||
|
||||
# Insert neighbors with different protocols
|
||||
changeset1 = Neighbor.changeset(%Neighbor{}, attrs1)
|
||||
assert {:ok, _neighbor1} = Repo.insert(changeset1)
|
||||
|
||||
changeset2 = Neighbor.changeset(%Neighbor{}, attrs2)
|
||||
assert {:ok, _neighbor2} = Repo.insert(changeset2)
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue