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

test: cover Topology role/link/evidence helper branches

Browse source 
Adds a dedicated test file that drives the small private mapping helpers
inside `Towerops.Topology` via their public surface:

- `role_to_type_atom/1`: every device_role string (router, core_router,
  switch, distribution_switch, access_switch, access_point, cpe,
  backhaul, backhaul_radio, server, firewall, ups, other, unknown
  fallback) is exercised through `device_to_node` (via
  `get_topology_for_map/2`) and again through `get_node_detail/2`.
- `link_type_priority/1`: each priority pair (lldp/cdp/mac_match/
  arp_inference) is exercised via `merge_bidirectional_edges/1` by
  upserting reciprocal links of differing types and asserting the
  higher-priority `link_type` survives the merge.
- `evidence_type_to_link_type/1`: every evidence_type clause
  (lldp_neighbor, cdp_neighbor, mac_on_interface, arp_entry,
  wireless_registration, plus the catch-all) is hit via
  `upsert_link/1` → `process_evidence_for_link/2`.

Coverage: 86.28% → 86.42% (+0.14%).
This commit is contained in:
Graham McIntire 2026-05-08 17:57:47 -05:00
parent 4ea64cb550
commit fc20d60838
1 changed files with 231 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

231
test/towerops/topology_helpers_branches_test.exs Normal file
View file

@ -0,0 +1,231 @@
defmodule Towerops.TopologyHelpersBranchesTest do
@moduledoc """
Targeted coverage for the small helper functions inside `Towerops.Topology`
that aren't exercised by the existing happy-path topology tests:
* `role_to_type_atom/1` for every device_role string,
* `link_type_priority/1` for each link_type via merge_bidirectional_edges,
* `evidence_type_to_link_type/1` via upsert_link → process_evidence_for_link.
"""
use Towerops.DataCase, async: false
import Towerops.AccountsFixtures
alias Towerops.Snmp.Device, as: SnmpDevice
alias Towerops.Snmp.Interface
alias Towerops.Topology
setup do
user = user_fixture()
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Topo Org"}, user.id)
{:ok, site} = Towerops.Sites.create_site(%{name: "Topo Site", organization_id: org.id})
%{org: org, site: site, user: user}
end
describe "device role → type mapping (covers role_to_type_atom/1)" do
@roles_to_types [
{"router", :router},
{"core_router", :router},
{"switch", :switch},
{"distribution_switch", :switch},
{"access_switch", :switch},
{"access_point", :wireless},
{"cpe", :wireless},
{"backhaul", :wireless},
{"backhaul_radio", :wireless},
{"server", :server},
{"firewall", :firewall},
{"ups", :server},
{"other", :unknown},
{"some-future-role", :unknown}
]
test "device_to_node returns the expected type for every known role", %{org: org, site: site} do
for {role, expected_type} <- @roles_to_types do
{:ok, device} =
Towerops.Devices.create_device(
%{
name: "dev-#{role}-#{System.unique_integer([:positive])}",
ip_address: random_ip(),
site_id: site.id,
organization_id: org.id,
device_role: role
},
bypass_limits: true
)
# Drives `role_to_type_atom/1` via device_to_node/2.
result = Topology.get_topology_for_map(org.id, "added")
node = Enum.find(result.nodes, &(&1.id == device.id))
assert node, "expected node for device with role=#{role}"
assert node.type == expected_type, "role=#{role} → expected #{inspect(expected_type)}, got #{inspect(node.type)}"
end
end
test "get_node_detail also exercises role_to_type_atom for varied roles", %{org: org, site: site} do
for role <- ["router", "switch", "access_point", "server", "firewall", "ups", "other"] do
{:ok, device} =
Towerops.Devices.create_device(
%{
name: "node-#{role}-#{System.unique_integer([:positive])}",
ip_address: random_ip(),
site_id: site.id,
organization_id: org.id,
device_role: role
},
bypass_limits: true
)
detail = Topology.get_node_detail(device.id, org.id)
assert detail
assert detail.id == device.id
assert detail.device_role == role
end
end
end
describe "merge_bidirectional_edges (covers link_type_priority/1)" do
test "with both directions present, the higher-priority link_type wins", %{org: org, site: site} do
# We need two devices with interfaces, two reciprocal links of differing
# link_types, then call get_topology_for_map. After merging, the surviving
# edge should carry the better link_type.
{router_a, iface_a} = device_with_iface(org, site, "Router-A", "10.10.0.1", "aa:00:00:00:00:01")
{router_b, iface_b} = device_with_iface(org, site, "Router-B", "10.10.0.2", "aa:00:00:00:00:02")
now = DateTime.utc_now()
# link_type combinations across the priority ladder.
cases = [
{"lldp", "cdp", "lldp"},
{"cdp", "mac_match", "cdp"},
{"mac_match", "arp_inference", "mac_match"},
{"arp_inference", "mac_match", "mac_match"}
]
for {type_ab, type_ba, expected_winner} <- cases do
# Wipe device_links so each iteration is isolated.
Towerops.Repo.delete_all(Towerops.Topology.DeviceLink)
{:ok, _} =
Topology.upsert_link(%{
source_device_id: router_a.id,
target_device_id: router_b.id,
source_interface_id: iface_a.id,
target_interface_id: iface_b.id,
link_type: type_ab,
confidence: 0.9,
last_confirmed_at: now,
evidence: []
})
{:ok, _} =
Topology.upsert_link(%{
source_device_id: router_b.id,
target_device_id: router_a.id,
source_interface_id: iface_b.id,
target_interface_id: iface_a.id,
link_type: type_ba,
confidence: 0.9,
last_confirmed_at: now,
evidence: []
})
result = Topology.get_topology_for_map(org.id, "added")
edge =
Enum.find(result.edges, fn e ->
(e.source == router_a.id and e.target == router_b.id) or
(e.source == router_b.id and e.target == router_a.id)
end)
assert edge, "expected merged edge for {#{type_ab}, #{type_ba}}"
assert edge.link_type == expected_winner
end
end
end
describe "evidence_type_to_link_type (covers all clauses via upsert_link)" do
test "every evidence type gets its expected link_type", %{org: org, site: site} do
{router, iface} = device_with_iface(org, site, "EvidenceR", "10.20.0.1", "bb:00:00:00:00:01")
pairs = [
{"lldp_neighbor", "lldp"},
{"cdp_neighbor", "cdp"},
{"mac_on_interface", "mac_match"},
{"arp_entry", "arp_inference"},
{"wireless_registration", "wireless_association"},
{"unknown_evidence_type", "mac_match"}
]
now = DateTime.utc_now()
for {evidence_type, _expected_link_type} <- pairs do
# Each iteration uses a fresh discovered MAC so the link is unique.
remote_mac =
"ee:#{255 |> :rand.uniform() |> Integer.to_string(16) |> String.pad_leading(2, "0")}:00:00:00:#{255 |> :rand.uniform() |> Integer.to_string(16) |> String.pad_leading(2, "0")}"
{:ok, _link} =
Topology.upsert_link(%{
source_device_id: router.id,
target_device_id: nil,
source_interface_id: iface.id,
link_type: "lldp",
confidence: 0.9,
discovered_remote_mac: remote_mac,
last_confirmed_at: now,
evidence: [
%{
evidence_type: evidence_type,
confidence: 0.9,
observed_at: now,
metadata: %{}
}
]
})
end
# Just confirm the upserts ran without raising — this exercises every
# evidence_type_to_link_type/1 clause as a side-effect of process_evidence_for_link.
links = Topology.list_device_links(router.id)
assert length(links) >= length(pairs)
end
end
# ---------- helpers ----------
defp device_with_iface(org, site, name, ip, mac) do
{:ok, device} =
Towerops.Devices.create_device(
%{
name: name,
ip_address: ip,
site_id: site.id,
organization_id: org.id
},
bypass_limits: true
)
snmp_dev =
%SnmpDevice{}
|> SnmpDevice.changeset(%{device_id: device.id, sys_name: String.downcase(name)})
|> Repo.insert!()
iface =
%Interface{}
|> Interface.changeset(%{
snmp_device_id: snmp_dev.id,
if_index: 1,
if_name: "eth0",
if_phys_address: mac
})
|> Repo.insert!()
{device, iface}
end
defp random_ip do
"10.#{:rand.uniform(254)}.#{:rand.uniform(254)}.#{:rand.uniform(254)}"
end
end