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

more tests

Browse source
This commit is contained in:
Graham McIntire 2026-01-20 16:38:53 -06:00
parent 7f338d8388
commit 23c92b2c44
No known key found for this signature in database
24 changed files with 1726 additions and 497 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

111
lib/mix/tasks/import_profiles.ex
View file

@ -182,46 +182,35 @@ defmodule Mix.Tasks.ImportProfiles do
# Look for sysDescr patterns in discovery section
case get_in(yaml_data, ["discovery"]) do
discovery when is_list(discovery) ->
sys_descr_patterns =
Enum.flat_map(discovery, fn item ->
case item do
%{"sysDescr" => patterns} when is_list(patterns) -> patterns
_ -> []
end
end)
sys_descr_patterns = Enum.flat_map(discovery, &extract_sys_descr_patterns/1)
case sys_descr_patterns do
[first | _] -> first
_ -> nil
end
List.first(sys_descr_patterns)
_ ->
nil
end
end
defp extract_sys_descr_patterns(%{"sysDescr" => patterns}) when is_list(patterns), do: patterns
defp extract_sys_descr_patterns(_), do: []
defp extract_detection_oid(yaml_data) do
# Look for sysObjectID patterns in discovery section
case get_in(yaml_data, ["discovery"]) do
discovery when is_list(discovery) ->
oid_patterns =
Enum.flat_map(discovery, fn item ->
case item do
%{"sysObjectID" => patterns} when is_list(patterns) -> patterns
_ -> []
end
end)
oid_patterns = Enum.flat_map(discovery, &extract_sys_object_id_patterns/1)
case oid_patterns do
[first | _] -> first
_ -> nil
end
List.first(oid_patterns)
_ ->
nil
end
end
defp extract_sys_object_id_patterns(%{"sysObjectID" => patterns}) when is_list(patterns), do: patterns
defp extract_sys_object_id_patterns(_), do: []
defp import_device_oids(profile, yaml_data) do
# Look for device info in various places
device_oids = extract_device_oids(yaml_data)
@ -310,46 +299,58 @@ defmodule Mix.Tasks.ImportProfiles do
end
defp parse_sensor(sensor_def, sensor_type) do
# Get the MIB name - prefer 'oid' field, fall back to 'value' or 'num_oid'
mib_name = Map.get(sensor_def, "oid") || Map.get(sensor_def, "value")
num_oid = Map.get(sensor_def, "num_oid")
# Skip table-based sensors (those with {{ $index }} templates)
if is_binary(num_oid) and String.contains?(num_oid, "{{ $index }}") do
nil
else
# Only import scalar sensors (ending with .0)
if is_binary(mib_name) and String.ends_with?(mib_name, ".0") do
# Extract description and clean up templates
descr =
case Map.get(sensor_def, "descr") do
nil ->
sensor_type
cond do
table_sensor?(num_oid) -> nil
scalar_sensor?(mib_name) -> build_sensor_map(sensor_def, sensor_type, mib_name)
true -> nil
end
end
descr_template when is_binary(descr_template) ->
# Remove template syntax like {{ MIB::name }}
descr_template
|> String.replace(~r/\{\{ [^}]+ \}\}/, "")
|> String.trim()
|> case do
"" -> sensor_type
cleaned -> cleaned
end
defp table_sensor?(num_oid) when is_binary(num_oid) do
String.contains?(num_oid, "{{ $index }}")
end
_ ->
sensor_type
end
defp table_sensor?(_), do: false
%{
sensor_type: sensor_type,
mib_name: mib_name,
sensor_descr: descr,
sensor_unit: Map.get(sensor_def, "unit"),
sensor_divisor: Map.get(sensor_def, "divisor", 1)
}
defp scalar_sensor?(mib_name) when is_binary(mib_name) do
String.ends_with?(mib_name, ".0")
end
# Skip non-scalar sensors for now
end
defp scalar_sensor?(_), do: false
defp build_sensor_map(sensor_def, sensor_type, mib_name) do
%{
sensor_type: sensor_type,
mib_name: mib_name,
sensor_descr: extract_sensor_description(sensor_def, sensor_type),
sensor_unit: Map.get(sensor_def, "unit"),
sensor_divisor: Map.get(sensor_def, "divisor", 1)
}
end
defp extract_sensor_description(sensor_def, default_type) do
case Map.get(sensor_def, "descr") do
nil ->
default_type
descr when is_binary(descr) ->
clean_description_template(descr, default_type)
_ ->
default_type
end
end
defp clean_description_template(descr, default_type) do
descr
|> String.replace(~r/\{\{ [^}]+ \}\}/, "")
|> String.trim()
|> case do
"" -> default_type
cleaned -> cleaned
end
end
end

85
lib/mix/tasks/upload_librenms.ex
View file

@ -101,50 +101,55 @@ defmodule Mix.Tasks.UploadLibrenms do
Logger.info("Found #{length(vendor_dirs)} vendor directories")
# Upload each vendor's MIBs
results =
Enum.flat_map(vendor_dirs, fn vendor_dir ->
vendor = Path.basename(vendor_dir)
Logger.info("Uploading MIBs for vendor: #{vendor}")
results = Enum.flat_map(vendor_dirs, &upload_vendor_mibs(&1, base_url, token))
# Find all MIB files in this vendor directory
mib_files =
vendor_dir
|> Path.join("*")
|> Path.wildcard()
|> Enum.filter(&File.regular?/1)
|> Enum.sort()
mib_files
|> Enum.map(fn file_path ->
filename = Path.basename(file_path)
# Double-check it's a regular file (not a directory or symlink to directory)
if File.regular?(file_path) do
Logger.debug(" Uploading: #{filename}")
case upload_mib_file(file_path, vendor, base_url, token) do
:ok ->
{:ok, filename}
{:error, reason} ->
Logger.error(" ✗ Failed to upload #{filename}: #{inspect(reason)}")
{:error, filename, reason}
end
else
Logger.debug(" Skipping non-file: #{filename}")
nil
end
end)
|> Enum.reject(&is_nil/1)
end)
successes = Enum.count(results, &match?({:ok, _}, &1))
failures = Enum.count(results, &match?({:error, _, _}, &1))
Logger.info("MIB upload complete: #{successes} succeeded, #{failures} failed")
report_upload_results(results)
end
end
defp upload_vendor_mibs(vendor_dir, base_url, token) do
vendor = Path.basename(vendor_dir)
Logger.info("Uploading MIBs for vendor: #{vendor}")
vendor_dir
|> find_mib_files()
|> Enum.map(&upload_single_mib(&1, vendor, base_url, token))
|> Enum.reject(&is_nil/1)
end
defp find_mib_files(vendor_dir) do
vendor_dir
|> Path.join("*")
|> Path.wildcard()
|> Enum.filter(&File.regular?/1)
|> Enum.sort()
end
defp upload_single_mib(file_path, vendor, base_url, token) do
if File.regular?(file_path) do
filename = Path.basename(file_path)
Logger.debug(" Uploading: #{filename}")
case upload_mib_file(file_path, vendor, base_url, token) do
:ok ->
{:ok, filename}
{:error, reason} ->
Logger.error(" ✗ Failed to upload #{filename}: #{inspect(reason)}")
{:error, filename, reason}
end
else
Logger.debug(" Skipping non-file: #{Path.basename(file_path)}")
nil
end
end
defp report_upload_results(results) do
successes = Enum.count(results, &match?({:ok, _}, &1))
failures = Enum.count(results, &match?({:error, _, _}, &1))
Logger.info("MIB upload complete: #{successes} succeeded, #{failures} failed")
end
defp upload_mib_file(file_path, vendor, base_url, token) do
url = "#{base_url}/api/v1/mibs"
filename = Path.basename(file_path)

8
lib/towerops/alerts.ex
View file

@ -71,7 +71,6 @@ defmodule Towerops.Alerts do
Accepts either an integer limit or a map of filters:
- limit (integer): Max number of alerts to return
- filters (map):
- severity: Filter by severity ("critical", "warning", "info")
- status: Filter by status ("active", "acknowledged", "resolved")
- limit: Max number of alerts to return
"""
@ -92,13 +91,6 @@ defmodule Towerops.Alerts do
preload: [device: {e, site: s}, acknowledged_by: []]
)
query =
if severity = filters["severity"] do
where(query, [a], a.severity == ^severity)
else
query
end
query =
case filters["status"] do
"active" -> where(query, [a], is_nil(a.resolved_at) and is_nil(a.acknowledged_at))

3
lib/towerops/api_tokens.ex
View file

@ -8,6 +8,7 @@ defmodule Towerops.ApiTokens do
import Ecto.Query, warn: false
alias Ecto.Adapters.SQL.Sandbox
alias Towerops.ApiTokens.ApiToken
alias Towerops.Repo
@ -202,7 +203,7 @@ defmodule Towerops.ApiTokens do
defp maybe_allow_sandbox(parent) do
if Application.get_env(:towerops, :sql_sandbox) do
Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self())
Sandbox.allow(Repo, parent, self())
end
end
end

2
lib/towerops/devices.ex
View file

@ -77,7 +77,7 @@ defmodule Towerops.Devices do
"""
def count_site_devices_down(site_id) do
Repo.aggregate(
from(e in DeviceSchema, where: e.site_id == ^site_id and e.status == "down"),
from(e in DeviceSchema, where: e.site_id == ^site_id and e.status == :down),
:count
)
end

26
lib/towerops/monitoring/device_monitor.ex
View file

@ -4,6 +4,7 @@ defmodule Towerops.Monitoring.DeviceMonitor do
"""
use GenServer
alias Ecto.Adapters.SQL.Sandbox
alias Towerops.Alerts
alias Towerops.Devices
alias Towerops.Monitoring
@ -221,21 +222,28 @@ defmodule Towerops.Monitoring.DeviceMonitor do
defp enqueue_check(device_id) do
if Application.get_env(:towerops, :env) == :test do
# In test, run synchronously with database sandbox access
parent = self()
Task.start(fn ->
if Application.get_env(:towerops, :sql_sandbox) do
Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self())
end
perform_check(device_id)
end)
start_check_task(device_id)
else
# In dev/prod, enqueue to Exq
{:ok, _job} = Exq.enqueue(Exq, "monitoring", Towerops.Workers.MonitorWorker, [device_id])
end
end
defp start_check_task(device_id) do
parent = self()
Task.start(fn ->
maybe_allow_sandbox(parent)
perform_check(device_id)
end)
end
defp maybe_allow_sandbox(parent) do
if Application.get_env(:towerops, :sql_sandbox) do
Sandbox.allow(Repo, parent, self())
end
end
defp via_tuple(device_id) do
{:via, Horde.Registry, {Towerops.Monitoring.Registry, device_id}}
end

15
lib/towerops/monitoring/ping.ex
View file

@ -25,13 +25,24 @@ defmodule Towerops.Monitoring.Ping do
@icmp_echo_reply 0
@doc """
Pings an IP address using raw ICMP echo request/reply.
Pings an IP address using raw ICMP echo request/reply with default timeout.
Returns {:ok, response_time_ms} on success, {:error, reason} on failure.
Response time is returned as a float with microsecond precision.
"""
@impl true
def ping(ip_address, timeout_ms \\ 5000) do
def ping(ip_address) do
ping(ip_address, 5000)
end
@doc """
Pings an IP address using raw ICMP echo request/reply with custom timeout.
Returns {:ok, response_time_ms} on success, {:error, reason} on failure.
Response time is returned as a float with microsecond precision.
"""
@impl true
def ping(ip_address, timeout_ms) do
start_time = System.monotonic_time(:microsecond)
case send_icmp_echo(ip_address, timeout_ms) do

29
lib/towerops/snmp/poller_worker.ex
View file

@ -12,7 +12,9 @@ defmodule Towerops.Snmp.PollerWorker do
"""
use GenServer
alias Ecto.Adapters.SQL.Sandbox
alias Towerops.Devices
alias Towerops.Repo
alias Towerops.Snmp
alias Towerops.Snmp.Client
alias Towerops.Snmp.NeighborDiscovery
@ -1051,7 +1053,7 @@ defmodule Towerops.Snmp.PollerWorker do
defp get_device_id_from_sensor(sensor) do
# Get the device to find device_id
case Towerops.Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do
case Repo.get(Towerops.Snmp.Device, sensor.snmp_device_id) do
nil -> nil
device -> device.device_id
end
@ -1067,21 +1069,28 @@ defmodule Towerops.Snmp.PollerWorker do
defp enqueue_poll(device_id) do
if Application.get_env(:towerops, :env) == :test do
# In test, run synchronously with database sandbox access
parent = self()
Task.start(fn ->
if Application.get_env(:towerops, :sql_sandbox) do
Ecto.Adapters.SQL.Sandbox.allow(Repo, parent, self())
end
perform_poll(device_id)
end)
start_poll_task(device_id)
else
# In dev/prod, enqueue to Exq
{:ok, _job} = Exq.enqueue(Exq, "polling", Towerops.Workers.PollWorker, [device_id])
end
end
defp start_poll_task(device_id) do
parent = self()
Task.start(fn ->
maybe_allow_sandbox(parent)
perform_poll(device_id)
end)
end
defp maybe_allow_sandbox(parent) do
if Application.get_env(:towerops, :sql_sandbox) do
Sandbox.allow(Repo, parent, self())
end
end
defp via_tuple(device_id) do
{:via, Horde.Registry, {PollerRegistry, device_id}}
end

78
lib/towerops/snmp/profiles/dynamic.ex
View file

@ -24,19 +24,7 @@ defmodule Towerops.Snmp.Profiles.Dynamic do
device_data =
profile.device_oids
|> Enum.map(fn device_oid ->
case MibTranslator.translate(device_oid.mib_name) do
{:ok, numeric_oid} ->
case Client.get(client_opts, numeric_oid) do
{:ok, value} -> {String.to_atom(device_oid.field), value}
{:error, _} -> nil
end
{:error, _} ->
Logger.warning("Failed to translate MIB name: #{device_oid.mib_name}")
nil
end
end)
|> Enum.map(&fetch_device_oid_value(&1, client_opts))
|> Enum.reject(&is_nil/1)
|> Map.new()
@ -75,30 +63,7 @@ defmodule Towerops.Snmp.Profiles.Dynamic do
sensors =
profile.sensor_oids
|> Enum.map(fn sensor_oid ->
case MibTranslator.translate(sensor_oid.mib_name) do
{:ok, numeric_oid} ->
case Client.get(client_opts, numeric_oid) do
{:ok, value} when is_integer(value) and value > 0 ->
%{
sensor_type: sensor_oid.sensor_type,
sensor_index: sensor_oid.sensor_type,
sensor_oid: numeric_oid,
sensor_descr: sensor_oid.sensor_descr || sensor_oid.sensor_type,
sensor_unit: sensor_oid.sensor_unit || "",
sensor_divisor: sensor_oid.sensor_divisor || 1,
sensor_value: value
}
_ ->
nil
end
{:error, _} ->
Logger.warning("Failed to translate sensor MIB name: #{sensor_oid.mib_name}")
nil
end
end)
|> Enum.map(&fetch_sensor_value(&1, client_opts))
|> Enum.reject(&is_nil/1)
{:ok, sensors}
@ -114,4 +79,43 @@ defmodule Towerops.Snmp.Profiles.Dynamic do
Base.discover_interfaces(client_opts)
end
# Private helper functions
defp fetch_device_oid_value(device_oid, client_opts) do
with {:ok, numeric_oid} <- MibTranslator.translate(device_oid.mib_name),
{:ok, value} <- Client.get(client_opts, numeric_oid) do
{String.to_atom(device_oid.field), value}
else
{:error, :translation_failed} ->
Logger.warning("Failed to translate MIB name: #{device_oid.mib_name}")
nil
{:error, _} ->
nil
end
end
defp fetch_sensor_value(sensor_oid, client_opts) do
with {:ok, numeric_oid} <- MibTranslator.translate(sensor_oid.mib_name),
{:ok, value} when is_integer(value) and value > 0 <-
Client.get(client_opts, numeric_oid) do
%{
sensor_type: sensor_oid.sensor_type,
sensor_index: sensor_oid.sensor_type,
sensor_oid: numeric_oid,
sensor_descr: sensor_oid.sensor_descr || sensor_oid.sensor_type,
sensor_unit: sensor_oid.sensor_unit || "",
sensor_divisor: sensor_oid.sensor_divisor || 1,
sensor_value: value
}
else
{:error, :translation_failed} ->
Logger.warning("Failed to translate sensor MIB name: #{sensor_oid.mib_name}")
nil
_ ->
nil
end
end
end

49
lib/towerops/snmp/profiles/net_snmp.ex
View file

@ -321,47 +321,22 @@ defmodule Towerops.Snmp.Profiles.NetSnmp do
Includes LM-SENSORS and UCD-SNMP tables.
"""
def collect_raw_debug_data(client_opts) do
# Walk LM-SENSORS tables
lm_temp_table =
case Client.walk(client_opts, "1.3.6.1.4.1.2021.13.16.2") do
{:ok, results} -> results
{:error, _} -> %{}
end
lm_fan_table =
case Client.walk(client_opts, "1.3.6.1.4.1.2021.13.16.3") do
{:ok, results} -> results
{:error, _} -> %{}
end
lm_voltage_table =
case Client.walk(client_opts, "1.3.6.1.4.1.2021.13.16.4") do
{:ok, results} -> results
{:error, _} -> %{}
end
# Walk UCD-SNMP tables
ucd_load_table =
case Client.walk(client_opts, "1.3.6.1.4.1.2021.10") do
{:ok, results} -> results
{:error, _} -> %{}
end
ucd_memory_table =
case Client.walk(client_opts, "1.3.6.1.4.1.2021.4") do
{:ok, results} -> results
{:error, _} -> %{}
end
%{
timestamp: DateTime.utc_now(),
lm_sensors_oids: @lm_sensors_oids,
ucd_snmp_oids: @ucd_snmp_oids,
lm_temp_table: lm_temp_table,
lm_fan_table: lm_fan_table,
lm_voltage_table: lm_voltage_table,
ucd_load_table: ucd_load_table,
ucd_memory_table: ucd_memory_table
lm_temp_table: safe_walk(client_opts, "1.3.6.1.4.1.2021.13.16.2"),
lm_fan_table: safe_walk(client_opts, "1.3.6.1.4.1.2021.13.16.3"),
lm_voltage_table: safe_walk(client_opts, "1.3.6.1.4.1.2021.13.16.4"),
ucd_load_table: safe_walk(client_opts, "1.3.6.1.4.1.2021.10"),
ucd_memory_table: safe_walk(client_opts, "1.3.6.1.4.1.2021.4")
}
end
defp safe_walk(client_opts, oid) do
case Client.walk(client_opts, oid) do
{:ok, results} -> results
{:error, _} -> %{}
end
end
end

93
lib/towerops_web/user_auth.ex
View file

@ -443,48 +443,7 @@ defmodule ToweropsWeb.UserAuth do
user = socket.assigns.current_scope && socket.assigns.current_scope.user
if user do
# Try to get organization from session, otherwise use first organization
org_id = session["current_organization_id"]
organization =
if org_id do
try do
Towerops.Organizations.get_organization!(org_id)
rescue
Ecto.NoResultsError -> nil
end
else
# Get first organization user has access to
case Towerops.Organizations.list_user_organizations(user.id) do
[first_org | _] -> first_org
[] -> nil
end
end
if organization do
membership = Towerops.Organizations.get_membership(organization.id, user.id)
if membership do
{:cont,
socket
|> Phoenix.Component.assign(:current_organization, organization)
|> Phoenix.Component.assign(:current_membership, membership)}
else
socket =
socket
|> LiveView.put_flash(:error, "You don't have access to any organizations.")
|> LiveView.redirect(to: ~p"/orgs")
{:halt, socket}
end
else
socket =
socket
|> LiveView.put_flash(:error, "Please select an organization.")
|> LiveView.redirect(to: ~p"/orgs")
{:halt, socket}
end
load_organization_for_user(socket, session, user)
else
{:cont, socket}
end
@ -521,6 +480,56 @@ defmodule ToweropsWeb.UserAuth do
end
end
defp load_organization_for_user(socket, session, user) do
org_id = session["current_organization_id"]
organization = find_user_organization(org_id, user.id)
case organization do
nil ->
halt_with_error(socket, "Please select an organization.")
org ->
assign_organization_and_membership(socket, org, user.id)
end
end
defp find_user_organization(org_id, _user_id) when is_binary(org_id) do
Towerops.Organizations.get_organization!(org_id)
rescue
Ecto.NoResultsError -> nil
end
defp find_user_organization(_org_id, user_id) do
case Towerops.Organizations.list_user_organizations(user_id) do
[first_org | _] -> first_org
[] -> nil
end
end
defp assign_organization_and_membership(socket, organization, user_id) do
membership = Towerops.Organizations.get_membership(organization.id, user_id)
case membership do
nil ->
halt_with_error(socket, "You don't have access to any organizations.")
mem ->
{:cont,
socket
|> Phoenix.Component.assign(:current_organization, organization)
|> Phoenix.Component.assign(:current_membership, mem)}
end
end
defp halt_with_error(socket, message) do
socket =
socket
|> LiveView.put_flash(:error, message)
|> LiveView.redirect(to: ~p"/orgs")
{:halt, socket}
end
defp mount_current_scope(socket, session) do
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn ->

56
test/towerops/alerts_test.exs
View file

@ -327,5 +327,61 @@ defmodule Towerops.AlertsTest do
Alerts.get_alert!(Ecto.UUID.generate())
end
end
test "send_alert_notification/1 sends notification and marks email as sent", %{
device: device
} do
attrs = Map.put(@valid_attrs, :device_id, device.id)
{:ok, alert} = Alerts.create_alert(attrs)
# Reload to get associations
alert = Alerts.get_alert!(alert.id)
refute alert.email_sent_at
{:ok, updated_alert} = Alerts.send_alert_notification(alert)
assert updated_alert.email_sent_at
assert %DateTime{} = updated_alert.email_sent_at
end
test "get_active_alert/2 returns most recent when multiple active alerts", %{
device: device
} do
attrs = Map.put(@valid_attrs, :device_id, device.id)
{:ok, old_alert} =
Alerts.create_alert(Map.put(attrs, :triggered_at, ~U[2025-12-21 10:00:00Z]))
{:ok, new_alert} =
Alerts.create_alert(Map.put(attrs, :triggered_at, ~U[2025-12-21 12:00:00Z]))
found = Alerts.get_active_alert(device.id, :device_down)
assert found.id == new_alert.id
refute found.id == old_alert.id
end
test "count_active_alerts/1 only counts device_down alerts", %{
device: device,
organization: organization
} do
# Create device_down alert
{:ok, _} =
Alerts.create_alert(%{
device_id: device.id,
alert_type: :device_down,
triggered_at: DateTime.utc_now()
})
# Create device_up alert (should not be counted)
{:ok, _} =
Alerts.create_alert(%{
device_id: device.id,
alert_type: :device_up,
triggered_at: DateTime.utc_now()
})
assert Alerts.count_active_alerts(organization.id) == 1
end
end
end

168
test/towerops/api_tokens_test.exs
View file

@ -1,5 +1,6 @@
defmodule Towerops.ApiTokensTest do
use Towerops.DataCase
use ExUnitProperties
import Towerops.AccountsFixtures
import Towerops.OrganizationsFixtures
@ -468,4 +469,171 @@ defmodule Towerops.ApiTokensTest do
assert "should be at most 255 character(s)" in errors_on(changeset).name
end
end
describe "property-based tests" do
property "generated tokens always start with towerops_" do
check all(_ <- constant(nil), max_runs: 50) do
user = user_fixture()
organization = organization_fixture(user.id)
{:ok, {_token, raw_token}} =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: "Test Token"
})
assert String.starts_with?(raw_token, "towerops_")
# Token should be sufficiently long (prefix + base64 encoded 32 bytes)
assert String.length(raw_token) > 40
end
end
property "tokens with valid names can always be created" do
check all(name <- string(:alphanumeric, min_length: 1, max_length: 255)) do
user = user_fixture()
organization = organization_fixture(user.id)
result =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: name
})
case result do
{:ok, {token, raw_token}} ->
assert token.name == name
assert String.starts_with?(raw_token, "towerops_")
{:error, _changeset} ->
flunk("Valid name '#{name}' was rejected")
end
end
end
property "verify_token roundtrip always works for non-expired tokens" do
check all(_ <- constant(nil), max_runs: 20) do
user = user_fixture()
organization = organization_fixture(user.id)
{:ok, {_token, raw_token}} =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: "Test Token"
})
# Verify the token we just created
assert {:ok, org_id, returned_user} = ApiTokens.verify_token(raw_token)
assert org_id == organization.id
assert returned_user.id == user.id
end
end
property "expired tokens always return error" do
check all(days_ago <- integer(1..365)) do
user = user_fixture()
organization = organization_fixture(user.id)
expires_at = DateTime.add(DateTime.utc_now(), -days_ago, :day)
{:ok, {_token, raw_token}} =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: "Expired Token",
expires_at: expires_at
})
assert {:error, :invalid_token} = ApiTokens.verify_token(raw_token)
end
end
property "list operations return consistent counts" do
check all(token_count <- integer(0..5)) do
user = user_fixture()
organization = organization_fixture(user.id)
# Create specified number of tokens
if token_count > 0 do
for i <- 1..token_count do
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: "Token #{i}"
})
end
end
org_tokens = ApiTokens.list_organization_api_tokens(organization.id)
user_tokens = ApiTokens.list_user_api_tokens(user.id)
assert length(org_tokens) == token_count
assert length(user_tokens) == token_count
assert length(org_tokens) == length(user_tokens)
end
end
property "deleting tokens reduces count by one" do
check all(_ <- constant(nil), max_runs: 20) do
user = user_fixture()
organization = organization_fixture(user.id)
# Create two tokens
{:ok, {token1, _}} =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: "Token 1"
})
{:ok, {_token2, _}} =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: "Token 2"
})
initial_count = length(ApiTokens.list_organization_api_tokens(organization.id))
assert initial_count == 2
# Delete one token
{:ok, _} = ApiTokens.delete_api_token(token1)
final_count = length(ApiTokens.list_organization_api_tokens(organization.id))
assert final_count == initial_count - 1
end
end
property "updating token name preserves token hash" do
check all(
original_name <- string(:alphanumeric, min_length: 1, max_length: 50),
new_name <- string(:alphanumeric, min_length: 1, max_length: 50)
) do
user = user_fixture()
organization = organization_fixture(user.id)
{:ok, {token, raw_token}} =
ApiTokens.create_api_token(%{
organization_id: organization.id,
user_id: user.id,
name: original_name
})
original_hash = token.token_hash
# Update the name
{:ok, updated_token} = ApiTokens.update_api_token(token, %{name: new_name})
# Token hash should remain the same (name changes don't affect token)
assert updated_token.token_hash == original_hash
assert updated_token.name == new_name
# Original raw token should still work
assert {:ok, org_id, _user} = ApiTokens.verify_token(raw_token)
assert org_id == organization.id
end
end
end
end

331
test/towerops/devices_test.exs
View file

@ -1,5 +1,6 @@
defmodule Towerops.EquipmentTest do
use Towerops.DataCase
use ExUnitProperties
alias Towerops.Devices
alias Towerops.Devices.Device
@ -455,4 +456,334 @@ defmodule Towerops.EquipmentTest do
assert length(events) == 5
end
end
describe "additional coverage tests" do
import Towerops.AccountsFixtures
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
})
%{organization: organization, site: site, user: user}
end
test "get_device/1 returns device when exists", %{site: site} do
{:ok, device} =
Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id
})
found = Devices.get_device(device.id)
assert found
assert found.id == device.id
end
test "get_device/1 returns nil when not exists" do
assert Devices.get_device(Ecto.UUID.generate()) == nil
end
test "get_device_with_details/1 preloads associations", %{site: site} do
{:ok, device} =
Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id
})
result = Devices.get_device_with_details(device.id)
assert result.id == device.id
assert Ecto.assoc_loaded?(result.site)
end
@tag :skip
test "delete_device/1 deletes the device", %{site: site} do
# Skipped: Requires Monitoring.Supervisor to be running
{:ok, device} =
Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id
})
assert {:ok, _deleted} = Devices.delete_device(device)
assert Devices.get_device(device.id) == nil
end
test "count_organization_devices/1 returns correct count", %{organization: organization, site: site} do
{:ok, _device1} =
Devices.create_device(%{
name: "Router 1",
ip_address: "192.168.1.1",
site_id: site.id
})
{:ok, _device2} =
Devices.create_device(%{
name: "Router 2",
ip_address: "192.168.1.2",
site_id: site.id
})
assert Devices.count_organization_devices(organization.id) == 2
end
test "count_site_devices/1 returns correct count", %{site: site} do
{:ok, _device1} =
Devices.create_device(%{
name: "Router 1",
ip_address: "192.168.1.1",
site_id: site.id
})
{:ok, _device2} =
Devices.create_device(%{
name: "Router 2",
ip_address: "192.168.1.2",
site_id: site.id
})
assert Devices.count_site_devices(site.id) == 2
end
test "count_site_devices_down/1 returns correct count", %{site: site} do
{:ok, device1} =
Devices.create_device(%{
name: "Router 1",
ip_address: "192.168.1.1",
site_id: site.id
})
{:ok, _device2} =
Devices.create_device(%{
name: "Router 2",
ip_address: "192.168.1.2",
site_id: site.id
})
# Mark one as down
{:ok, _} = Devices.update_device_status(device1, :down)
assert Devices.count_site_devices_down(site.id) == 1
end
test "list_organization_devices/2 filters by site_id", %{organization: organization, site: site} do
{:ok, site2} =
Towerops.Sites.create_site(%{
name: "Site 2",
organization_id: organization.id
})
{:ok, device1} =
Devices.create_device(%{
name: "Router 1",
ip_address: "192.168.1.1",
site_id: site.id
})
{:ok, _device2} =
Devices.create_device(%{
name: "Router 2",
ip_address: "192.168.1.2",
site_id: site2.id
})
result = Devices.list_organization_devices(organization.id, %{"site_id" => site.id})
assert length(result) == 1
assert hd(result).id == device1.id
end
test "list_organization_devices/2 filters by status", %{organization: organization, site: site} do
{:ok, device1} =
Devices.create_device(%{
name: "Router 1",
ip_address: "192.168.1.1",
site_id: site.id
})
{:ok, _device2} =
Devices.create_device(%{
name: "Router 2",
ip_address: "192.168.1.2",
site_id: site.id
})
# Mark one as up
Devices.update_device_status(device1, :up)
result = Devices.list_organization_devices(organization.id, %{"status" => "up"})
assert length(result) == 1
assert hd(result).id == device1.id
end
test "get_snmp_config/1 with device ID string", %{site: site} do
{:ok, device} =
Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: "test-community"
})
config = Devices.get_snmp_config(device.id)
assert config.community == "test-community"
assert config.source == :device
end
end
describe "property-based tests" do
import Towerops.AccountsFixtures
property "creating devices with valid IPs always succeeds" do
check all(
name <- string(:alphanumeric, min_length: 2, max_length: 20),
octet1 <- integer(0..255),
octet2 <- integer(0..255),
octet3 <- integer(0..255),
octet4 <- integer(1..254)
) do
# Create fresh org and site for each property test run
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
})
ip_address = "#{octet1}.#{octet2}.#{octet3}.#{octet4}"
attrs = %{
name: name,
ip_address: ip_address,
site_id: site.id
}
assert {:ok, %Device{} = device} = Devices.create_device(attrs)
assert device.ip_address == ip_address
assert device.status == :unknown
end
end
property "device status changes are idempotent" do
check all(status <- member_of([:up, :down, :unknown])) 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} =
Devices.create_device(%{
name: "Test Device",
ip_address: "192.168.1.100",
site_id: site.id
})
# Update status twice
{:ok, updated1} = Devices.update_device_status(device, status)
{:ok, updated2} = Devices.update_device_status(updated1, status)
assert updated2.status == status
# Second update should not change last_status_change_at
assert updated2.last_status_change_at == updated1.last_status_change_at
end
end
property "device counters are always non-negative" do
check all(device_count <- integer(0..5)) 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
})
# Create specified number of devices
if device_count > 0 do
for i <- 1..device_count do
Devices.create_device(%{
name: "Device #{i}",
ip_address: "192.168.1.#{i}",
site_id: site.id
})
end
end
org_count = Devices.count_organization_devices(organization.id)
site_count = Devices.count_site_devices(site.id)
down_count = Devices.count_site_devices_down(site.id)
assert org_count >= 0
assert site_count >= 0
assert down_count >= 0
assert down_count <= site_count
assert org_count == site_count
end
end
property "SNMP config prioritization is consistent" do
check all(
device_community <- one_of([constant(nil), string(:alphanumeric, min_length: 1, max_length: 20)]),
site_community <- one_of([constant(nil), string(:alphanumeric, min_length: 1, max_length: 20)])
) 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
})
# Update site with community if provided
site =
if site_community do
{:ok, updated_site} = Towerops.Sites.update_site(site, %{snmp_community: site_community})
updated_site
else
site
end
{:ok, device} =
Devices.create_device(%{
name: "Router",
ip_address: "192.168.1.1",
site_id: site.id,
snmp_enabled: true,
snmp_community: device_community
})
device = Towerops.Repo.preload(device, site: :organization)
config = Devices.get_snmp_config(device)
# Device-level always takes precedence if set
if device_community do
assert config.community == device_community
assert config.source == :device
else
if site_community do
assert config.community == site_community
assert config.source == :site
else
assert config.source in [:organization, :default]
end
end
end
end
end
end

4
test/towerops/monitoring/device_monitor_test.exs
View file

@ -76,7 +76,7 @@ defmodule Towerops.Monitoring.DeviceMonitorTest do
# Verify check was created
checks = Monitoring.list_devices_checks(device.id)
assert length(checks) >= 1
assert checks != [], "Expected at least 1 check"
assert hd(checks).status == :success
assert hd(checks).response_time_ms == 10
@ -292,7 +292,7 @@ defmodule Towerops.Monitoring.DeviceMonitorTest do
# Check recovery alert was created
alerts = Alerts.list_devices_alerts(device.id)
up_alerts = Enum.filter(alerts, &(&1.alert_type == :device_up))
assert length(up_alerts) >= 1
assert up_alerts != [], "Expected at least 1 up alert"
up_alert = hd(up_alerts)
assert up_alert.message =~ "now responding"

108
test/towerops/monitoring/ping_test.exs
View file

@ -1,5 +1,6 @@
defmodule Towerops.Monitoring.PingTest do
use ExUnit.Case, async: true
use ExUnitProperties
alias Towerops.Monitoring.Ping
@ -161,11 +162,15 @@ defmodule Towerops.Monitoring.PingTest do
describe "PingBehaviour implementation" do
test "implements ping/1 callback" do
assert function_exported?(Ping, :ping, 1)
# Verify the function exists by checking module exports
exports = Ping.module_info(:exports)
assert {:ping, 1} in exports
end
test "implements ping/2 callback" do
assert function_exported?(Ping, :ping, 2)
# Verify the function exists by checking module exports
exports = Ping.module_info(:exports)
assert {:ping, 2} in exports
end
@tag :integration
@ -196,4 +201,103 @@ defmodule Towerops.Monitoring.PingTest do
end
end
end
describe "property-based tests" do
property "invalid IP strings always return error" do
check all(
invalid_ip <-
one_of([
string(:alphanumeric, min_length: 1, max_length: 20),
:ascii
|> string(min_length: 0, max_length: 50)
|> filter(&(!valid_ip_format?(&1)))
])
) do
result = Ping.ping(invalid_ip, 100)
assert match?({:error, _}, result)
end
end
property "valid IPv4 addresses are accepted by parser" do
check all(
octet1 <- integer(0..255),
octet2 <- integer(0..255),
octet3 <- integer(0..255),
octet4 <- integer(0..255)
) do
ip = "#{octet1}.#{octet2}.#{octet3}.#{octet4}"
result = Ping.ping(ip, 10)
# Should either succeed or fail with known errors (not invalid_ip)
case result do
{:ok, response_time} ->
assert is_float(response_time)
assert response_time >= 0
{:error, :invalid_ip} ->
flunk("Valid IP #{ip} was rejected as invalid")
{:error, _reason} ->
# Timeout, insufficient privileges, or other network errors are acceptable
assert true
end
end
end
property "timeout values affect maximum execution time" do
check all(timeout <- integer(1..100)) do
start_time = System.monotonic_time(:millisecond)
_result = Ping.ping("192.0.2.1", timeout)
end_time = System.monotonic_time(:millisecond)
elapsed = end_time - start_time
# Should complete within reasonable margin of timeout (allow 3x for system overhead)
assert elapsed < timeout * 3
end
end
property "response times are always non-negative when successful" do
check all(_ <- constant(nil), max_runs: 10) do
case Ping.ping("127.0.0.1", 100) do
{:ok, response_time} ->
assert is_float(response_time)
assert response_time >= 0.0
# Should be reasonable (less than timeout)
assert response_time < 100.0
{:error, _} ->
# Expected in test environment without privileges or network issues
assert true
end
end
end
property "ping always returns consistent tuple format" do
check all(
timeout <- integer(1..1000),
max_runs: 20
) do
result = Ping.ping("127.0.0.1", timeout)
case result do
{:ok, response_time} ->
assert is_float(response_time)
{:error, reason} ->
assert is_atom(reason) or is_tuple(reason)
other ->
flunk("Unexpected return value: #{inspect(other)}")
end
end
end
end
# Helper function for property test
defp valid_ip_format?(str) do
case :inet.parse_address(String.to_charlist(str)) do
{:ok, _} -> true
{:error, _} -> false
end
end
end

47
test/towerops/redis_health_check_test.exs
View file

@ -1,6 +1,8 @@
defmodule Towerops.RedisHealthCheckTest do
use ExUnit.Case, async: false
import ExUnit.CaptureLog
alias Towerops.RedisHealthCheck
describe "wait_for_redis/2" do
@ -14,8 +16,16 @@ defmodule Towerops.RedisHealthCheckTest do
test "returns error after max attempts with unreachable Redis" do
redis_config = [host: "127.0.0.1", port: 9999]
assert {:error, :max_attempts_exceeded} =
RedisHealthCheck.wait_for_redis(redis_config, max_attempts: 2, backoff: 50, timeout: 100)
# Suppress Redis connection warnings during test
_logs =
capture_log(fn ->
assert {:error, :max_attempts_exceeded} =
RedisHealthCheck.wait_for_redis(redis_config,
max_attempts: 2,
backoff: 50,
timeout: 100
)
end)
end
test "retries with exponential backoff" do
@ -23,7 +33,15 @@ defmodule Towerops.RedisHealthCheckTest do
start_time = System.monotonic_time(:millisecond)
RedisHealthCheck.wait_for_redis(redis_config, max_attempts: 3, backoff: 100, timeout: 100)
# Suppress Redis connection warnings during test
_logs =
capture_log(fn ->
RedisHealthCheck.wait_for_redis(redis_config,
max_attempts: 3,
backoff: 100,
timeout: 100
)
end)
end_time = System.monotonic_time(:millisecond)
elapsed = end_time - start_time
@ -49,22 +67,33 @@ defmodule Towerops.RedisHealthCheckTest do
test "returns error for unavailable Redis" do
redis_config = [host: "127.0.0.1", port: 9999]
assert {:error, _reason} = RedisHealthCheck.check_health(redis_config, 100)
# Suppress Redis connection warnings during test
_logs =
capture_log(fn ->
assert {:error, _reason} = RedisHealthCheck.check_health(redis_config, 100)
end)
end
test "handles malformed Redis config" do
redis_config = [host: "not-a-valid-hostname-at-all-12345", port: 6379]
assert {:error, _reason} = RedisHealthCheck.check_health(redis_config, 100)
# Suppress Redis connection warnings during test
_logs =
capture_log(fn ->
assert {:error, _reason} = RedisHealthCheck.check_health(redis_config, 100)
end)
end
test "accepts custom timeout" do
redis_config = [host: "127.0.0.1", port: 9999]
result = RedisHealthCheck.check_health(redis_config, 200)
# Should return error (either immediate connection failure or timeout)
assert match?({:error, _}, result)
# Suppress Redis connection warnings during test
_logs =
capture_log(fn ->
result = RedisHealthCheck.check_health(redis_config, 200)
# Should return error (either immediate connection failure or timeout)
assert match?({:error, _}, result)
end)
end
end

425
test/towerops/sites_test.exs
View file

@ -1,81 +1,70 @@
defmodule Towerops.SitesTest do
use Towerops.DataCase
alias Towerops.Devices.Device
alias Towerops.Sites
alias Towerops.Sites.Site
describe "sites" do
import Towerops.AccountsFixtures
alias Towerops.Sites.Site
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
%{organization: organization, user: user}
end
@valid_attrs %{name: "Test Site", location: "New York", description: "A test site"}
@update_attrs %{name: "Updated Site", location: "Boston", description: "Updated desc"}
@invalid_attrs %{name: nil}
@valid_attrs %{
name: "Main Office",
description: "Primary location",
location: "123 Main St",
snmp_version: "2c",
snmp_community: "public"
}
test "list_organization_sites/1 returns all sites for an organization", %{
organization: organization
} do
{:ok, site} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
result = Sites.list_organization_sites(organization.id)
assert length(result) == 1
assert hd(result).id == site.id
@invalid_attrs %{name: nil, organization_id: nil}
test "list_organization_sites/1 returns all sites for an organization", %{organization: organization} do
{:ok, site1} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
{:ok, site2} = Sites.create_site(%{name: "Branch Office", organization_id: organization.id})
sites = Sites.list_organization_sites(organization.id)
assert length(sites) == 2
assert Enum.any?(sites, &(&1.id == site1.id))
assert Enum.any?(sites, &(&1.id == site2.id))
end
test "list_organization_sites/1 orders sites alphabetically", %{organization: organization} do
{:ok, _zebra} = Sites.create_site(%{name: "Zebra Site", organization_id: organization.id})
{:ok, _alpha} = Sites.create_site(%{name: "Alpha Site", organization_id: organization.id})
sites = Sites.list_organization_sites(organization.id)
assert List.first(sites).name == "Alpha Site"
assert List.last(sites).name == "Zebra Site"
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
{:ok, _} = Sites.create_site(%{name: "Site 1", organization_id: organization.id})
{:ok, _} = Sites.create_site(%{name: "Site 2", organization_id: organization.id})
assert Sites.count_organization_sites(organization.id) == 2
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})
{:ok, root2} =
Sites.create_site(%{name: "Root 2", organization_id: organization.id})
{:ok, _child} =
Sites.create_site(%{
name: "Child",
organization_id: organization.id,
parent_site_id: root1.id
})
test "list_root_sites/1 returns only sites without parents", %{organization: organization} do
{:ok, root} = Sites.create_site(%{name: "Root Site", organization_id: organization.id})
{:ok, _child} = Sites.create_site(%{name: "Child Site", organization_id: organization.id, parent_site_id: root.id})
roots = Sites.list_root_sites(organization.id)
assert length(roots) == 2
assert Enum.any?(roots, &(&1.id == root1.id))
assert Enum.any?(roots, &(&1.id == root2.id))
assert length(roots) == 1
assert hd(roots).id == root.id
end
test "list_child_sites/1 returns child sites for a parent", %{organization: organization} do
test "list_child_sites/1 returns child sites of a parent", %{organization: organization} do
{:ok, parent} = Sites.create_site(%{name: "Parent", organization_id: organization.id})
{:ok, child1} =
Sites.create_site(%{
name: "Child 1",
organization_id: organization.id,
parent_site_id: parent.id
})
{:ok, child2} =
Sites.create_site(%{
name: "Child 2",
organization_id: organization.id,
parent_site_id: parent.id
})
{:ok, child1} = Sites.create_site(%{name: "Child 1", organization_id: organization.id, parent_site_id: parent.id})
{:ok, child2} = Sites.create_site(%{name: "Child 2", organization_id: organization.id, parent_site_id: parent.id})
children = Sites.list_child_sites(parent.id)
assert length(children) == 2
@ -83,185 +72,201 @@ defmodule Towerops.SitesTest do
assert Enum.any?(children, &(&1.id == child2.id))
end
test "list_child_sites/1 returns empty list when no children exist", %{
organization: organization
} do
{:ok, parent} = Sites.create_site(%{name: "Parent", organization_id: organization.id})
assert Sites.list_child_sites(parent.id) == []
test "get_site!/1 returns the site with given id", %{organization: organization} do
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: organization.id})
fetched = Sites.get_site!(site.id)
assert fetched.id == site.id
assert fetched.name == "Test Site"
end
test "get_site!/1 returns the site with preloads", %{organization: organization} do
test "get_site!/1 preloads associations", %{organization: organization} do
{:ok, parent} = Sites.create_site(%{name: "Parent", organization_id: organization.id})
{:ok, site} = Sites.create_site(%{name: "Child", organization_id: organization.id, parent_site_id: parent.id})
{:ok, child} =
Sites.create_site(%{
name: "Child",
organization_id: organization.id,
parent_site_id: parent.id
})
retrieved_parent = Sites.get_site!(parent.id)
assert retrieved_parent.id == parent.id
assert Ecto.assoc_loaded?(retrieved_parent.child_sites)
assert length(retrieved_parent.child_sites) == 1
retrieved_child = Sites.get_site!(child.id)
assert retrieved_child.id == child.id
assert Ecto.assoc_loaded?(retrieved_child.parent_site)
assert retrieved_child.parent_site.id == parent.id
fetched = Sites.get_site!(site.id)
assert Ecto.assoc_loaded?(fetched.parent_site)
assert Ecto.assoc_loaded?(fetched.child_sites)
assert Ecto.assoc_loaded?(fetched.device)
end
test "get_organization_site!/2 returns the site with given id", %{organization: organization} do
{:ok, site} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
test "get_site!/1 raises when site doesn't exist" do
assert_raise Ecto.NoResultsError, fn ->
Sites.get_site!(Ecto.UUID.generate())
end
end
assert Sites.get_organization_site!(organization.id, site.id).id == site.id
test "get_organization_site!/2 returns site for specific organization", %{organization: organization} do
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: organization.id})
fetched = Sites.get_organization_site!(organization.id, site.id)
assert fetched.id == site.id
end
test "get_organization_site!/2 preloads associations", %{organization: organization} do
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: organization.id})
fetched = Sites.get_organization_site!(organization.id, site.id)
assert Ecto.assoc_loaded?(fetched.parent_site)
assert Ecto.assoc_loaded?(fetched.child_sites)
assert Ecto.assoc_loaded?(fetched.device)
end
test "get_organization_site!/2 raises when site doesn't exist" do
user = user_fixture()
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Org"}, user.id)
assert_raise Ecto.NoResultsError, fn ->
Sites.get_organization_site!(org.id, Ecto.UUID.generate())
end
end
test "create_site/1 with valid data creates a site", %{organization: organization} do
attrs = Map.put(@valid_attrs, :organization_id, organization.id)
assert {:ok, %Site{} = site} = Sites.create_site(attrs)
assert site.name == "Test Site"
assert site.location == "New York"
assert site.description == "A test site"
end
test "create_site/1 with parent_site_id creates hierarchical site", %{
organization: organization
} do
{:ok, parent} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
attrs =
@valid_attrs
|> Map.put(:organization_id, organization.id)
|> Map.put(:parent_site_id, parent.id)
|> Map.put(:name, "Child Site")
assert {:ok, %Site{} = child} = Sites.create_site(attrs)
assert child.parent_site_id == parent.id
assert site.name == "Main Office"
assert site.description == "Primary location"
assert site.location == "123 Main St"
assert site.snmp_version == "2c"
assert site.snmp_community == "public"
end
test "create_site/1 with invalid data returns error changeset" do
assert {:error, %Ecto.Changeset{}} = Sites.create_site(@invalid_attrs)
end
test "create_site/1 validates name length", %{organization: organization} do
# Too short
attrs = %{name: "A", organization_id: organization.id}
assert {:error, changeset} = Sites.create_site(attrs)
assert "should be at least 2 character(s)" in errors_on(changeset).name
# Too long
attrs = %{name: String.duplicate("a", 201), organization_id: organization.id}
assert {:error, changeset} = Sites.create_site(attrs)
assert "should be at most 200 character(s)" in errors_on(changeset).name
end
test "create_site/1 validates description length", %{organization: organization} do
attrs = %{name: "Test Site", description: String.duplicate("a", 1001), organization_id: organization.id}
assert {:error, changeset} = Sites.create_site(attrs)
assert "should be at most 1000 character(s)" in errors_on(changeset).description
end
test "create_site/1 validates location length", %{organization: organization} do
attrs = %{name: "Test Site", location: String.duplicate("a", 201), organization_id: organization.id}
assert {:error, changeset} = Sites.create_site(attrs)
assert "should be at most 200 character(s)" in errors_on(changeset).location
end
test "create_site/1 validates snmp_version inclusion", %{organization: organization} do
attrs = %{name: "Test Site", snmp_version: "invalid", organization_id: organization.id}
assert {:error, changeset} = Sites.create_site(attrs)
assert "must be 1, 2c, or 3" in errors_on(changeset).snmp_version
end
test "create_site/1 prevents circular parent reference", %{organization: organization} do
{:ok, site} = Sites.create_site(%{name: "Test Site", organization_id: organization.id})
# Try to make site its own parent
assert {:error, changeset} = Sites.update_site(site, %{parent_site_id: site.id})
assert "cannot be the same as the site itself" in errors_on(changeset).parent_site_id
end
test "update_site/2 with valid data updates the site", %{organization: organization} do
{:ok, site} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
assert {:ok, %Site{} = site} = Sites.update_site(site, @update_attrs)
assert site.name == "Updated Site"
assert site.location == "Boston"
{:ok, site} = Sites.create_site(%{name: "Original", organization_id: organization.id})
assert {:ok, %Site{} = updated} = Sites.update_site(site, %{name: "Updated", description: "New description"})
assert updated.name == "Updated"
assert updated.description == "New description"
end
test "update_site/2 with invalid data returns error changeset", %{organization: organization} do
{:ok, site} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
assert {:error, %Ecto.Changeset{}} = Sites.update_site(site, @invalid_attrs)
assert Sites.get_organization_site!(organization.id, site.id).name == site.name
{:ok, site} = Sites.create_site(%{name: "Test", organization_id: organization.id})
assert {:error, %Ecto.Changeset{}} = Sites.update_site(site, %{name: nil})
end
test "delete_site/1 deletes the site", %{organization: organization} do
{:ok, site} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
{:ok, site} = Sites.create_site(%{name: "Test", organization_id: organization.id})
assert {:ok, %Site{}} = Sites.delete_site(site)
assert_raise Ecto.NoResultsError, fn ->
Sites.get_organization_site!(organization.id, site.id)
end
assert_raise Ecto.NoResultsError, fn -> Sites.get_site!(site.id) end
end
test "change_site/1 returns a site changeset", %{organization: organization} do
{:ok, site} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
{:ok, site} = Sites.create_site(%{name: "Test", organization_id: organization.id})
assert %Ecto.Changeset{} = Sites.change_site(site)
end
test "build_site_tree/1 builds hierarchical tree structure", %{organization: organization} do
{:ok, parent} = Sites.create_site(Map.put(@valid_attrs, :organization_id, organization.id))
test "change_site/2 returns a site changeset with changes", %{organization: organization} do
{:ok, site} = Sites.create_site(%{name: "Test", organization_id: organization.id})
changeset = Sites.change_site(site, %{name: "New Name"})
assert %Ecto.Changeset{} = changeset
assert changeset.changes.name == "New Name"
end
{:ok, _child1} =
Sites.create_site(%{
name: "Child 1",
organization_id: organization.id,
parent_site_id: parent.id
})
test "build_site_tree/1 builds hierarchical structure", %{organization: organization} do
{:ok, root1} = Sites.create_site(%{name: "Root 1", organization_id: organization.id})
{:ok, root2} = Sites.create_site(%{name: "Root 2", organization_id: organization.id})
{:ok, child1} = Sites.create_site(%{name: "Child 1", organization_id: organization.id, parent_site_id: root1.id})
{:ok, _child2} =
Sites.create_site(%{
name: "Child 2",
organization_id: organization.id,
parent_site_id: parent.id
})
{:ok, grandchild} =
Sites.create_site(%{name: "Grandchild", organization_id: organization.id, parent_site_id: child1.id})
tree = Sites.build_site_tree(organization.id)
assert length(tree) == 1
[root] = tree
assert length(root.children) == 2
end
end
assert length(tree) == 2
describe "bulk operations" do
import Towerops.AccountsFixtures
root1_node = Enum.find(tree, &(&1.site.id == root1.id))
assert length(root1_node.children) == 1
assert hd(root1_node.children).site.id == child1.id
assert length(hd(root1_node.children).children) == 1
assert hd(hd(root1_node.children).children).site.id == grandchild.id
setup do
user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
%{organization: organization}
root2_node = Enum.find(tree, &(&1.site.id == root2.id))
assert length(root2_node.children) == 0
end
test "apply_snmp_config_to_all_equipment/1 updates all devices at site", %{
organization: organization
} do
test "build_site_tree/1 returns empty list for organization with no sites", %{organization: organization} do
tree = Sites.build_site_tree(organization.id)
assert tree == []
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"
snmp_version: "2c",
snmp_community: "test-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
site_id: site.id,
snmp_version: "1",
snmp_community: "old-community"
})
{: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
site_id: site.id,
snmp_version: "1",
snmp_community: "old-community"
})
{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"
updated1 = Towerops.Devices.get_device!(device1.id)
assert updated1.snmp_version == "2c"
assert updated1.snmp_community == "test-community"
updated_device2 = Repo.get!(Device, device2.id)
assert updated_device2.snmp_version == "3"
assert updated_device2.snmp_community == "site-community"
updated2 = Towerops.Devices.get_device!(device2.id)
assert updated2.snmp_version == "2c"
assert updated2.snmp_community == "test-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")
test "apply_agent_to_all_equipment/1 creates agent assignments", %{organization: organization, user: _user} do
{:ok, agent_token, _token} = Towerops.Agents.create_agent_token(organization.id, "Test Agent")
{:ok, site} =
Sites.create_site(%{
@ -294,79 +299,39 @@ defmodule Towerops.SitesTest do
assert assignment2.agent_token_id == agent_token.id
end
test "apply_agent_to_all_equipment/1 deletes assignments when agent is nil", %{
organization: organization
test "apply_agent_to_all_equipment/1 deletes assignments when no agent set", %{
organization: organization,
user: _user
} do
{:ok, agent_token, _token} =
Towerops.Agents.create_agent_token(organization.id, "Test Agent")
{: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
})
{:ok, device} =
Towerops.Devices.create_device(%{
name: "Device 1",
ip_address: "192.168.1.1",
site_id: site.id
})
# Create initial assignment
Sites.apply_agent_to_all_equipment(site.id)
assert Towerops.Agents.get_device_assignment(device.id)
# Remove agent from site
{:ok, site} = Sites.update_site(site, %{agent_token_id: nil})
# Apply again - should delete assignments
{count, _} = Sites.apply_agent_to_all_equipment(site.id)
assert count == 0
assert count == 1
# Verify assignment was deleted
assert is_nil(Towerops.Agents.get_device_assignment(device.id))
end
end
end

3
test/towerops/snmp/discovery_test.exs
View file

@ -4,6 +4,7 @@ defmodule Towerops.Snmp.DiscoveryTest do
import Mox
import Towerops.AccountsFixtures
alias Ecto.Adapters.SQL.Sandbox
alias Towerops.Devices.Device
alias Towerops.Repo
alias Towerops.Snmp.Device
@ -20,7 +21,7 @@ defmodule Towerops.Snmp.DiscoveryTest do
event_logger_pid = Process.whereis(Towerops.Devices.EventLogger)
if event_logger_pid do
Ecto.Adapters.SQL.Sandbox.allow(Repo, self(), event_logger_pid)
Sandbox.allow(Repo, self(), event_logger_pid)
end
user = user_fixture()

6
test/towerops/snmp/mib_parser_test.exs
View file

@ -248,7 +248,7 @@ defmodule Towerops.Snmp.MibParserTest do
test "returns absolute paths" do
files = MibParser.list_mib_files()
if length(files) > 0 do
if files != [] do
Enum.each(files, fn path ->
assert String.starts_with?(path, "/")
end)
@ -267,7 +267,7 @@ defmodule Towerops.Snmp.MibParserTest do
files = MibParser.list_mib_files()
# Should find MIBs in different subdirectories
if length(files) > 0 do
if files != [] do
# At least some files should be in subdirectories
subdirs =
files
@ -275,7 +275,7 @@ defmodule Towerops.Snmp.MibParserTest do
|> Enum.uniq()
# Should have multiple subdirectories
assert length(subdirs) >= 1
assert subdirs != [], "Expected at least 1 subdirectory"
end
end
end

3
test/towerops/snmp/neighbor_discovery_test.exs
View file

@ -8,9 +8,6 @@ defmodule Towerops.Snmp.NeighborDiscoveryTest do
setup :verify_on_exit!
@lldp_rem_table_oid "1.0.8802.1.1.2.1.4.1.1"
@cdp_cache_table_oid "1.3.6.1.4.1.9.9.23.1.2.1.1"
setup do
# Create mock interface data
interfaces = [

6
test/towerops/workers/monitor_worker_test.exs
View file

@ -63,7 +63,7 @@ defmodule Towerops.Workers.MonitorWorkerTest do
# Verify no check was created
checks = Monitoring.list_devices_checks(device.id, 10)
assert length(checks) == 0
assert checks == []
end
test "returns error when device not found" do
@ -132,7 +132,7 @@ defmodule Towerops.Workers.MonitorWorkerTest do
# Verify a successful check was created
checks = Monitoring.list_devices_checks(device.id, 10)
assert length(checks) >= 1
assert checks != [], "Expected at least 1 check"
check = hd(checks)
assert check.device_id == device.id
@ -148,7 +148,7 @@ defmodule Towerops.Workers.MonitorWorkerTest do
# Verify no check was created
checks = Monitoring.list_devices_checks(device.id, 10)
assert length(checks) == 0
assert checks == []
end
end
end

4
test/towerops/workers/poll_worker_test.exs
View file

@ -145,11 +145,11 @@ defmodule Towerops.Workers.PollWorkerTest do
# Verify SNMP device exists and has sensors
snmp_device = Snmp.get_device_with_associations(device.id)
assert snmp_device
assert length(snmp_device.sensors) >= 1
assert snmp_device.sensors != [], "Expected at least 1 sensor"
# Verify error reading was created
readings = Snmp.get_sensor_readings(sensor.id, limit: 10)
assert length(readings) >= 1, "Expected at least 1 sensor reading but got #{length(readings)}"
assert readings != [], "Expected at least 1 sensor reading"
reading = hd(readings)
assert reading.status == "error"
assert reading.value == nil

563
test/towerops_web/user_auth_test.exs
View file

@ -1,5 +1,6 @@
defmodule ToweropsWeb.UserAuthTest do
use ToweropsWeb.ConnCase, async: true
use ExUnitProperties
import Towerops.AccountsFixtures
@ -218,6 +219,24 @@ defmodule ToweropsWeb.UserAuthTest do
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
"You must re-authenticate to access this page."
end
test "stores return path on GET requests when redirecting", %{conn: conn, user: user} do
eleven_minutes_ago = DateTime.add(DateTime.utc_now(), -11, :minute)
user =
user
|> Ecto.Changeset.change(authenticated_at: eleven_minutes_ago)
|> Towerops.Repo.update!()
conn =
%{conn | path_info: ["admin", "users"], query_string: "", method: "GET"}
|> fetch_flash()
|> assign(:current_scope, Scope.for_user(user))
|> UserAuth.require_sudo_mode([])
assert redirected_to(conn) == ~p"/users/log-in"
assert get_session(conn, :user_return_to) == "/admin/users"
end
end
describe "redirect_if_user_is_authenticated/2" do
@ -600,6 +619,21 @@ defmodule ToweropsWeb.UserAuthTest do
assert socket.redirected
end
test "redirects to /devices when user has organizations", %{user: user} do
{:ok, _organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
user_token = Accounts.generate_user_session_token(user)
{:halt, socket} =
UserAuth.on_mount(
:redirect_if_user_is_authenticated,
%{},
%{"user_token" => user_token},
%Socket{}
)
assert socket.redirected
end
test "continues for unauthenticated users" do
{:cont, socket} =
UserAuth.on_mount(
@ -723,6 +757,23 @@ defmodule ToweropsWeb.UserAuthTest do
refute socket.redirected
end
test "redirects when org_slug exists but no user" do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(nil) end)
{:halt, socket} =
UserAuth.on_mount(
:load_current_organization,
%{"org_slug" => "test-org"},
%{},
socket
)
assert socket.redirected
end
end
describe "on_mount/4 - :require_superuser" do
@ -777,4 +828,516 @@ defmodule ToweropsWeb.UserAuthTest do
assert socket.redirected
end
end
describe "on_mount/4 - :require_sudo_mode" do
test "allows users in sudo mode", %{user: user} do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:cont, result_socket} =
UserAuth.on_mount(
:require_sudo_mode,
%{},
%{},
socket
)
refute result_socket.redirected
end
test "redirects users not in sudo mode" do
eleven_minutes_ago = DateTime.add(DateTime.utc_now(), -11, :minute)
user =
user_fixture()
|> Ecto.Changeset.change(authenticated_at: eleven_minutes_ago)
|> Towerops.Repo.update!()
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:halt, result_socket} =
UserAuth.on_mount(
:require_sudo_mode,
%{},
%{},
socket
)
assert result_socket.redirected
end
test "redirects unauthenticated users" do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(nil) end)
{:halt, result_socket} =
UserAuth.on_mount(
:require_sudo_mode,
%{},
%{},
socket
)
assert result_socket.redirected
end
end
describe "on_mount/4 - :load_default_organization" do
test "loads organization from session for authenticated user", %{user: user} do
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:cont, result_socket} =
UserAuth.on_mount(
:load_default_organization,
%{},
%{"current_organization_id" => organization.id},
socket
)
assert result_socket.assigns.current_organization.id == organization.id
assert result_socket.assigns.current_membership
end
test "loads first organization when no org_id in session", %{user: user} do
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:cont, result_socket} =
UserAuth.on_mount(
:load_default_organization,
%{},
%{},
socket
)
assert result_socket.assigns.current_organization.id == organization.id
assert result_socket.assigns.current_membership
end
test "halts when organization not found", %{user: user} do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:halt, result_socket} =
UserAuth.on_mount(
:load_default_organization,
%{},
%{"current_organization_id" => Ecto.UUID.generate()},
socket
)
assert result_socket.redirected
end
test "halts when user has no organizations", %{user: user} do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:halt, result_socket} =
UserAuth.on_mount(
:load_default_organization,
%{},
%{},
socket
)
assert result_socket.redirected
end
test "halts when user is not a member of organization", %{user: user} do
other_user = user_fixture()
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Other Org"}, other_user.id)
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(user) end)
{:halt, result_socket} =
UserAuth.on_mount(
:load_default_organization,
%{},
%{"current_organization_id" => organization.id},
socket
)
assert result_socket.redirected
end
test "continues when user is not authenticated" do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
socket =
Phoenix.Component.assign_new(socket, :current_scope, fn -> Scope.for_user(nil) end)
{:cont, result_socket} =
UserAuth.on_mount(
:load_default_organization,
%{},
%{},
socket
)
refute result_socket.redirected
end
end
describe "store_return_to_for_liveview/2" do
test "stores GET path when no existing return_to", %{conn: conn} do
conn =
UserAuth.store_return_to_for_liveview(
%{conn | request_path: "/devices", method: "GET", path_info: ["devices"], query_string: ""},
[]
)
assert get_session(conn, :user_return_to) == "/devices"
end
test "does not store if return_to already exists", %{conn: conn} do
conn =
conn
|> put_session(:user_return_to, "/original")
|> Map.put(:request_path, "/new")
|> Map.put(:method, "GET")
|> UserAuth.store_return_to_for_liveview([])
assert get_session(conn, :user_return_to) == "/original"
end
test "does not store POST requests", %{conn: conn} do
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/devices", method: "POST"}, [])
refute get_session(conn, :user_return_to)
end
test "does not store login path", %{conn: conn} do
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/log-in", method: "GET"}, [])
refute get_session(conn, :user_return_to)
end
test "does not store register path", %{conn: conn} do
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/register", method: "GET"}, [])
refute get_session(conn, :user_return_to)
end
test "does not store reset password path", %{conn: conn} do
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/reset-password", method: "GET"}, [])
refute get_session(conn, :user_return_to)
end
test "does not store confirm path", %{conn: conn} do
conn = UserAuth.store_return_to_for_liveview(%{conn | request_path: "/users/confirm", method: "GET"}, [])
refute get_session(conn, :user_return_to)
end
end
describe "log_out_user/1 with live_socket_id" do
test "broadcasts disconnect when live_socket_id exists", %{conn: conn, user: user} do
user_token = Accounts.generate_user_session_token(user)
live_socket_id = "users_sessions:#{user.id}"
conn =
conn
|> put_session(:user_token, user_token)
|> put_session(:live_socket_id, live_socket_id)
|> fetch_cookies()
# Subscribe to the topic to verify broadcast
ToweropsWeb.Endpoint.subscribe(live_socket_id)
conn = UserAuth.log_out_user(conn)
# Verify broadcast was sent
assert_receive %Phoenix.Socket.Broadcast{
topic: ^live_socket_id,
event: "disconnect"
}
refute get_session(conn, :user_token)
assert redirected_to(conn) == ~p"/users/log-in"
end
end
describe "redirect_if_user_is_authenticated/2 with organizations" do
test "redirects to /devices when user has organizations", %{conn: conn, user: user} do
{:ok, _organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
conn =
conn
|> assign(:current_scope, Scope.for_user(user))
|> UserAuth.redirect_if_user_is_authenticated([])
assert conn.halted
assert redirected_to(conn) == ~p"/devices"
end
end
describe "mount_current_scope/2 timezone handling" do
test "uses user's timezone when set", %{user: user} do
user =
user
|> Ecto.Changeset.change(timezone: "America/New_York")
|> Towerops.Repo.update!()
user_token = Accounts.generate_user_session_token(user)
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:cont, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{"user_token" => user_token},
socket
)
assert result_socket.assigns.timezone == "America/New_York"
end
test "defaults to UTC when user has no timezone", %{user: user} do
user_token = Accounts.generate_user_session_token(user)
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:cont, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{"user_token" => user_token},
socket
)
# When user has no timezone and no connect_params, defaults to UTC
assert result_socket.assigns.timezone == "UTC"
end
test "defaults to UTC when not authenticated" do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:halt, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{},
socket
)
# Unauthenticated users also get UTC as default
assert result_socket.assigns.timezone == "UTC"
end
end
describe "fetch_current_scope_for_user/2 edge cases" do
test "assigns nil scope when no token in session or cookies", %{conn: conn} do
conn =
conn
|> fetch_cookies()
|> UserAuth.fetch_current_scope_for_user([])
assert Map.has_key?(conn.assigns, :current_scope)
assert conn.assigns[:current_scope] == nil || conn.assigns.current_scope.user == nil
end
end
describe "on_mount/4 impersonation session building" do
test "builds impersonation scope from session" do
superuser = user_fixture(%{is_superuser: true})
target_user = user_fixture()
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:cont, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{
"impersonating" => true,
"superuser_id" => superuser.id,
"target_user_id" => target_user.id
},
socket
)
assert result_socket.assigns.current_scope.user.id == target_user.id
assert result_socket.assigns.current_scope.superuser.id == superuser.id
end
test "clears impersonation when superuser not found" do
target_user = user_fixture()
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:halt, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{
"impersonating" => true,
"superuser_id" => Ecto.UUID.generate(),
"target_user_id" => target_user.id
},
socket
)
assert result_socket.redirected
end
test "clears impersonation when target user not found" do
superuser = user_fixture(%{is_superuser: true})
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:halt, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{
"impersonating" => true,
"superuser_id" => superuser.id,
"target_user_id" => Ecto.UUID.generate()
},
socket
)
assert result_socket.redirected
end
test "clears impersonation when missing IDs" do
socket = %Socket{assigns: %{__changed__: %{}, flash: %{}}}
{:halt, result_socket} =
UserAuth.on_mount(
:require_authenticated_user,
%{},
%{"impersonating" => true},
socket
)
assert result_socket.redirected
end
end
describe "log_in_user/3 with return_to" do
test "restores user_return_to after login and keeps it in session", %{conn: conn, user: user} do
conn =
conn
|> put_session(:user_return_to, "/devices")
|> UserAuth.log_in_user(user)
assert redirected_to(conn) == "/devices"
# The return_to is restored in the session by renew_session
assert get_session(conn, :user_return_to) == "/devices"
end
test "does not clear session when already logged in with same user", %{conn: conn, user: user} do
# First login
conn = UserAuth.log_in_user(conn, user)
token = get_session(conn, :user_token)
# Re-login with same user (simulating session renewal)
conn =
conn
|> recycle()
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|> init_test_session(%{user_token: token})
|> assign(:current_scope, Scope.for_user(user))
|> put_session(:test_data, "should_persist")
|> UserAuth.log_in_user(user)
# Session data should persist when re-authenticating with same user
assert get_session(conn, :test_data) == "should_persist"
end
end
describe "property-based tests" do
property "store_return_to_for_liveview always stores valid paths for GET requests", %{conn: conn} do
check all(
path <- string(:alphanumeric, min_length: 1, max_length: 50),
query <- one_of([constant(""), string(:alphanumeric, min_length: 1, max_length: 20)])
) do
path = "/" <> path
path_info = String.split(path, "/", trim: true)
conn =
UserAuth.store_return_to_for_liveview(
%{conn | request_path: path, method: "GET", path_info: path_info, query_string: query},
[]
)
stored_path = get_session(conn, :user_return_to)
if String.starts_with?(path, "/users/log-in") or String.starts_with?(path, "/users/register") or
String.starts_with?(path, "/users/reset-password") or
String.starts_with?(path, "/users/confirm") do
refute stored_path
else
assert stored_path
assert String.starts_with?(stored_path, "/")
end
end
end
property "fetch_current_scope_for_user always assigns a scope", %{conn: conn} do
check all(has_token <- boolean()) do
conn =
if has_token do
user = user_fixture(%{authenticated_at: DateTime.utc_now(:second)})
token = Accounts.generate_user_session_token(user)
put_session(conn, :user_token, token)
else
conn
end
conn = UserAuth.fetch_current_scope_for_user(conn, [])
assert Map.has_key?(conn.assigns, :current_scope)
end
end
property "redirect paths are always valid for authenticated users", %{user: user} do
check all(has_org <- boolean()) do
if has_org do
{:ok, _org} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
end
# Reload user to get fresh org data
user = Towerops.Repo.reload(user)
conn =
build_conn()
|> Map.replace!(:secret_key_base, ToweropsWeb.Endpoint.config(:secret_key_base))
|> init_test_session(%{})
|> assign(:current_scope, Scope.for_user(user))
|> UserAuth.redirect_if_user_is_authenticated([])
assert conn.halted
redirect_path = redirected_to(conn)
assert redirect_path in [~p"/orgs", ~p"/devices"]
end
end
end
end