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

Add comprehensive test suite for Agents.Stats module

Browse source 
Implemented 18 tests covering all 6 functions in the Stats module:
- get_organization_agent_health/1 (3 tests)
- get_equipment_assignment_breakdown/1 (2 tests)
- get_offline_agents/1 (3 tests)
- get_high_load_agents/2 (3 tests)
- get_unmonitored_equipment/1 (4 tests)
- get_agent_metric_stats/1 (2 tests)

All tests verify correct behavior with proper fixtures, SNMP credentials,
and DateTime handling with :second precision.

Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Graham McIntire 2026-01-14 09:22:32 -06:00
parent 569b5cb671
commit 1abdc549c7
No known key found for this signature in database
1 changed files with 455 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

455
test/towerops/agents/stats_test.exs Normal file
View file

@ -0,0 +1,455 @@
defmodule Towerops.Agents.StatsTest do
use Towerops.DataCase
import Towerops.AccountsFixtures
import Towerops.OrganizationsFixtures
alias Towerops.Agents
alias Towerops.Agents.Stats
alias Towerops.Equipment
alias Towerops.Repo
alias Towerops.Sites
alias Towerops.Snmp.Device
alias Towerops.Snmp.Interface
alias Towerops.Snmp.InterfaceStat
alias Towerops.Snmp.Sensor
alias Towerops.Snmp.SensorReading
describe "get_organization_agent_health/1" do
test "returns empty list for organization with no agents" do
user = user_fixture()
org = organization_fixture(user.id)
assert Stats.get_organization_agent_health(org.id) == []
end
test "returns online agents with equipment counts" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Test Agent")
# Touch last_seen_at to make it online
agent_token
|> Ecto.Changeset.change(%{
last_seen_at: DateTime.truncate(DateTime.utc_now(), :second),
metadata: %{"version" => "0.1.0", "hostname" => "test-host"}
})
|> Repo.update!()
health = Stats.get_organization_agent_health(org.id)
assert length(health) == 1
agent_health = hd(health)
assert agent_health.id == agent_token.id
assert agent_health.name == "Test Agent"
assert agent_health.online == true
assert agent_health.equipment_count == 0
assert agent_health.version == "0.1.0"
assert agent_health.hostname == "test-host"
end
test "marks agents as offline after 5 minutes" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Offline Agent")
# Set last_seen_at to 6 minutes ago
six_minutes_ago = DateTime.utc_now() |> DateTime.add(-6, :minute) |> DateTime.truncate(:second)
agent_token
|> Ecto.Changeset.change(%{last_seen_at: six_minutes_ago})
|> Repo.update!()
health = Stats.get_organization_agent_health(org.id)
assert length(health) == 1
agent_health = hd(health)
assert agent_health.online == false
end
test "includes equipment count from hierarchical assignments" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Test Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, equipment} =
Equipment.create_equipment(%{
name: "Test Router",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
{:ok, _assignment} = Agents.assign_equipment_to_agent(agent_token.id, equipment.id)
agent_token
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
health = Stats.get_organization_agent_health(org.id)
assert length(health) == 1
agent_health = hd(health)
assert agent_health.equipment_count == 1
end
end
describe "get_equipment_assignment_breakdown/1" do
test "returns zero counts for organization with no equipment" do
user = user_fixture()
org = organization_fixture(user.id)
breakdown = Stats.get_equipment_assignment_breakdown(org.id)
assert breakdown == %{direct: 0, site: 0, organization: 0, cloud: 0}
end
test "counts equipment by assignment type" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Test Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
# Direct assignment
{:ok, equipment1} =
Equipment.create_equipment(%{
name: "Router 1",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
{:ok, _assignment} = Agents.assign_equipment_to_agent(agent_token.id, equipment1.id)
# Cloud polling (no agent)
{:ok, _equipment2} =
Equipment.create_equipment(%{
name: "Router 2",
ip_address: "192.168.1.2",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
breakdown = Stats.get_equipment_assignment_breakdown(org.id)
assert breakdown.direct == 1
assert breakdown.cloud == 1
end
end
describe "get_offline_agents/1" do
test "returns empty list when all agents are online" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Online Agent")
agent_token
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|> Repo.update!()
offline = Stats.get_offline_agents(org.id)
assert offline == []
end
test "returns agents not seen in last 5 minutes" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Offline Agent")
six_minutes_ago = DateTime.utc_now() |> DateTime.add(-6, :minute) |> DateTime.truncate(:second)
agent_token
|> Ecto.Changeset.change(%{last_seen_at: six_minutes_ago})
|> Repo.update!()
offline = Stats.get_offline_agents(org.id)
assert length(offline) == 1
agent = hd(offline)
assert agent.id == agent_token.id
assert agent.name == "Offline Agent"
end
test "returns agents that have never connected" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Never Connected")
offline = Stats.get_offline_agents(org.id)
assert length(offline) == 1
agent = hd(offline)
assert agent.id == agent_token.id
assert is_nil(agent.last_seen_at)
end
end
describe "get_high_load_agents/2" do
test "returns empty list when no agents exceed threshold" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, _agent_token, _token} = Agents.create_agent_token(org.id, "Low Load Agent")
high_load = Stats.get_high_load_agents(org.id, 50)
assert high_load == []
end
test "returns agents with equipment count above threshold" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "High Load Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
# Create 3 equipment and assign all to agent
for i <- 1..3 do
{:ok, equipment} =
Equipment.create_equipment(%{
name: "Router #{i}",
ip_address: "192.168.1.#{i}",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
{:ok, _assignment} = Agents.assign_equipment_to_agent(agent_token.id, equipment.id)
end
high_load = Stats.get_high_load_agents(org.id, 2)
assert length(high_load) == 1
agent = hd(high_load)
assert agent.id == agent_token.id
assert agent.name == "High Load Agent"
assert agent.equipment_count == 3
end
test "respects custom threshold parameter" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
# Create 5 equipment
for i <- 1..5 do
{:ok, equipment} =
Equipment.create_equipment(%{
name: "Router #{i}",
ip_address: "192.168.1.#{i}",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
{:ok, _assignment} = Agents.assign_equipment_to_agent(agent_token.id, equipment.id)
end
# Should appear with threshold 4
high_load_4 = Stats.get_high_load_agents(org.id, 4)
assert length(high_load_4) == 1
# Should not appear with threshold 10
high_load_10 = Stats.get_high_load_agents(org.id, 10)
assert high_load_10 == []
end
end
describe "get_unmonitored_equipment/1" do
test "returns empty list when all equipment has agents" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, equipment} =
Equipment.create_equipment(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
{:ok, _assignment} = Agents.assign_equipment_to_agent(agent_token.id, equipment.id)
unmonitored = Stats.get_unmonitored_equipment(org.id)
assert unmonitored == []
end
test "returns equipment with no agent assigned" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, equipment} =
Equipment.create_equipment(%{
name: "Unmonitored Router",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: true
})
unmonitored = Stats.get_unmonitored_equipment(org.id)
assert length(unmonitored) == 1
eq = hd(unmonitored)
assert eq.id == equipment.id
assert eq.name == "Unmonitored Router"
assert eq.ip_address == "192.168.1.1"
assert eq.site_name == "Test Site"
end
test "excludes equipment with SNMP disabled" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, _equipment} =
Equipment.create_equipment(%{
name: "SNMP Disabled",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: false,
monitoring_enabled: true
})
unmonitored = Stats.get_unmonitored_equipment(org.id)
assert unmonitored == []
end
test "excludes equipment with monitoring disabled" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, _equipment} =
Equipment.create_equipment(%{
name: "Monitoring Disabled",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c",
monitoring_enabled: false
})
unmonitored = Stats.get_unmonitored_equipment(org.id)
assert unmonitored == []
end
end
describe "get_agent_metric_stats/1" do
test "returns zero stats for agent with no metrics" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Agent")
stats = Stats.get_agent_metric_stats(agent_token.id)
assert stats.total_metrics == 0
assert stats.sensor_readings == 0
assert stats.interface_stats == 0
assert stats.avg_per_hour == 0
assert is_nil(stats.last_submission)
end
test "counts sensor readings and interface stats from last 24 hours" do
user = user_fixture()
org = organization_fixture(user.id)
{:ok, agent_token, _token} = Agents.create_agent_token(org.id, "Agent")
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: org.id})
{:ok, equipment} =
Equipment.create_equipment(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "public",
snmp_version: "2c"
})
{:ok, _assignment} = Agents.assign_equipment_to_agent(agent_token.id, equipment.id)
# Create SNMP device, sensor, and interface
device =
%Device{}
|> Device.changeset(%{
equipment_id: equipment.id,
sys_name: "Test Device",
sys_descr: "Test"
})
|> Repo.insert!()
sensor =
%Sensor{}
|> Sensor.changeset(%{
snmp_device_id: device.id,
sensor_type: "temperature",
sensor_index: "1",
sensor_oid: "1.2.3.4"
})
|> Repo.insert!()
interface =
%Interface{}
|> Interface.changeset(%{
snmp_device_id: device.id,
if_index: 1,
if_name: "GigabitEthernet0/1"
})
|> Repo.insert!()
# Insert sensor reading
%SensorReading{}
|> SensorReading.changeset(%{
sensor_id: sensor.id,
value: 45.5,
status: "ok",
checked_at: DateTime.utc_now()
})
|> Repo.insert!()
# Insert interface stat
%InterfaceStat{}
|> InterfaceStat.changeset(%{
interface_id: interface.id,
if_in_octets: 1_234_567,
if_out_octets: 9_876_543,
if_in_errors: 0,
if_out_errors: 0,
if_in_discards: 0,
if_out_discards: 0,
checked_at: DateTime.utc_now()
})
|> Repo.insert!()
stats = Stats.get_agent_metric_stats(agent_token.id)
assert stats.total_metrics == 2
assert stats.sensor_readings == 1
assert stats.interface_stats == 1
assert stats.avg_per_hour > 0
assert stats.last_submission
end
end
end