Improve test coverage: AgentLive.Index, UserAuth, Cisco profile
- Added comprehensive tests for ToweropsWeb.AgentLive.Index - Tests mount, create_agent, revoke_agent, and close_token_modal events - Tests agent status display and last_seen formatting - Coverage improved from 57.50% to 80.00% - Enhanced ToweropsWeb.UserAuth tests - Added tests for impersonation functionality - Added tests for organization loading and membership verification - Added tests for superuser authorization - Coverage improved from 51.08% to 61.87% - Expanded Towerops.Snmp.Profiles.Cisco tests - Added tests for all sensor type parsing functions - Added tests for different sensor scales, statuses, and error conditions - Tests sensor discovery with various data formats - Coverage improved from 57.30% to 100.00% Overall test coverage improved from 64.25% to 65.86%
This commit is contained in:
parent
7cf3e61709
commit
e6650833f6
3 changed files with 528 additions and 0 deletions
|
|
@ -224,5 +224,158 @@ defmodule Towerops.Snmp.Profiles.CiscoTest do
|
|||
# Should fall back to Base profile
|
||||
assert {:ok, []} = result
|
||||
end
|
||||
|
||||
test "discovers various sensor types" do
|
||||
# Test different sensor types to cover more parsing functions
|
||||
for sensor_type <- [1, 2, 3, 4, 5, 6, 7, 9, 10, 11, 12, 13, 14, 99] do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.#{sensor_type}", value: sensor_type}]}
|
||||
end)
|
||||
|
||||
# Mock get_multiple to return proper sensor data
|
||||
expect(SnmpMock, :get, 7, fn _, oid, _ ->
|
||||
cond do
|
||||
String.contains?(oid, ".1.1.1.1.1.") -> {:ok, sensor_type}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, -3}
|
||||
String.contains?(oid, ".1.1.1.1.3.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.4.") -> {:ok, 1000}
|
||||
String.contains?(oid, ".1.1.1.1.5.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, "Test Sensor"}
|
||||
true -> {:ok, ""}
|
||||
end
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
assert {:ok, _sensors} = result
|
||||
end
|
||||
end
|
||||
|
||||
test "handles different sensor scales" do
|
||||
for scale <- [-24, -12, -9, -6, -3, 0, 3, 5] do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.1000", value: 8}]}
|
||||
end)
|
||||
|
||||
expect(SnmpMock, :get, 7, fn _, oid, _ ->
|
||||
cond do
|
||||
String.contains?(oid, ".1.1.1.1.1.") -> {:ok, 8}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, scale}
|
||||
String.contains?(oid, ".1.1.1.1.3.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.4.") -> {:ok, 1000}
|
||||
String.contains?(oid, ".1.1.1.1.5.") -> {:ok, 1}
|
||||
true -> {:ok, "Test"}
|
||||
end
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
assert {:ok, _sensors} = result
|
||||
end
|
||||
end
|
||||
|
||||
test "handles different sensor statuses" do
|
||||
for status <- [1, 2, 3, 99] do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.1000", value: 8}]}
|
||||
end)
|
||||
|
||||
expect(SnmpMock, :get, 7, fn _, oid, _ ->
|
||||
cond do
|
||||
String.contains?(oid, ".1.1.1.1.1.") -> {:ok, 8}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, 0}
|
||||
String.contains?(oid, ".1.1.1.1.3.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.4.") -> {:ok, 1000}
|
||||
String.contains?(oid, ".1.1.1.1.5.") -> {:ok, status}
|
||||
true -> {:ok, "Test"}
|
||||
end
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
assert {:ok, _sensors} = result
|
||||
end
|
||||
end
|
||||
|
||||
test "handles sensor with only name (no descr)" do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.1000", value: 8}]}
|
||||
end)
|
||||
|
||||
expect(SnmpMock, :get, 7, fn _, oid, _ ->
|
||||
cond do
|
||||
String.contains?(oid, ".1.1.1.1.1.") -> {:ok, 8}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, 0}
|
||||
String.contains?(oid, ".1.1.1.1.3.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.4.") -> {:ok, 1000}
|
||||
String.contains?(oid, ".1.1.1.1.5.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, ""}
|
||||
true -> {:ok, "Sensor Name"}
|
||||
end
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
assert {:ok, _sensors} = result
|
||||
end
|
||||
|
||||
test "handles sensor get_multiple failure" do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.1000", value: 8}]}
|
||||
end)
|
||||
|
||||
# Mock get_multiple to fail
|
||||
expect(SnmpMock, :get, 7, fn _, _, _ ->
|
||||
{:error, :timeout}
|
||||
end)
|
||||
|
||||
# Falls back to Base.discover_sensors which calls walk again
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, []}
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
|
||||
# Should return empty list when sensor building fails
|
||||
assert {:ok, []} = result
|
||||
end
|
||||
|
||||
test "handles non-integer scale in calculate_divisor" do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.1000", value: 8}]}
|
||||
end)
|
||||
|
||||
expect(SnmpMock, :get, 7, fn _, oid, _ ->
|
||||
cond do
|
||||
String.contains?(oid, ".1.1.1.1.1.") -> {:ok, 8}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, "invalid"}
|
||||
String.contains?(oid, ".1.1.1.1.3.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.4.") -> {:ok, 1000}
|
||||
String.contains?(oid, ".1.1.1.1.5.") -> {:ok, 1}
|
||||
true -> {:ok, "Test"}
|
||||
end
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
assert {:ok, _sensors} = result
|
||||
end
|
||||
|
||||
test "handles invalid sensor value" do
|
||||
expect(SnmpMock, :walk, fn _, _, _ ->
|
||||
{:ok, [%{oid: "1.3.6.1.4.1.9.9.91.1.1.1.1.1.1000", value: 8}]}
|
||||
end)
|
||||
|
||||
expect(SnmpMock, :get, 7, fn _, oid, _ ->
|
||||
cond do
|
||||
String.contains?(oid, ".1.1.1.1.1.") -> {:ok, 8}
|
||||
String.contains?(oid, ".1.1.1.1.2.") -> {:ok, 0}
|
||||
String.contains?(oid, ".1.1.1.1.3.") -> {:ok, 1}
|
||||
String.contains?(oid, ".1.1.1.1.4.") -> {:ok, "invalid"}
|
||||
String.contains?(oid, ".1.1.1.1.5.") -> {:ok, 1}
|
||||
true -> {:ok, "Test"}
|
||||
end
|
||||
end)
|
||||
|
||||
result = Cisco.discover_sensors(@client_opts)
|
||||
assert {:ok, sensors} = result
|
||||
# Should still return sensor but with nil value
|
||||
assert length(sensors) == 1
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
205
test/towerops_web/live/agent_live/index_test.exs
Normal file
205
test/towerops_web/live/agent_live/index_test.exs
Normal file
|
|
@ -0,0 +1,205 @@
|
|||
defmodule ToweropsWeb.AgentLive.IndexTest do
|
||||
use ToweropsWeb.ConnCase, async: true
|
||||
|
||||
import Phoenix.LiveViewTest
|
||||
import Towerops.AccountsFixtures
|
||||
|
||||
alias Towerops.Agents
|
||||
|
||||
setup do
|
||||
user = user_fixture()
|
||||
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
%{user: user, organization: organization}
|
||||
end
|
||||
|
||||
describe "mount" do
|
||||
test "loads agent tokens for organization", %{conn: conn, user: user, organization: organization} do
|
||||
# Create some agent tokens
|
||||
{:ok, agent_token1, _token1} = Agents.create_agent_token(organization.id, "Agent 1")
|
||||
{:ok, agent_token2, _token2} = Agents.create_agent_token(organization.id, "Agent 2")
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, _view, html} = live(conn)
|
||||
|
||||
assert html =~ "Remote Agents"
|
||||
assert html =~ agent_token1.name
|
||||
assert html =~ agent_token2.name
|
||||
end
|
||||
|
||||
test "shows empty state when no agents exist", %{conn: conn, user: user, organization: organization} do
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, _view, html} = live(conn)
|
||||
|
||||
assert html =~ "Remote Agents"
|
||||
end
|
||||
|
||||
test "requires authentication", %{conn: conn, organization: organization} do
|
||||
conn = get(conn, "/orgs/#{organization.slug}/agents")
|
||||
assert redirected_to(conn) == ~p"/users/log-in"
|
||||
end
|
||||
end
|
||||
|
||||
describe "create_agent event" do
|
||||
test "creates new agent token and shows modal", %{conn: conn, user: user, organization: organization} do
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, view, _html} = live(conn)
|
||||
|
||||
# Trigger create_agent event with nested params
|
||||
html = render_submit(view, "create_agent", %{"agent_form" => %{"name" => "New Agent"}})
|
||||
|
||||
assert html =~ "Agent created successfully"
|
||||
assert html =~ "New Agent"
|
||||
|
||||
# Verify agent was actually created
|
||||
agent_tokens = Agents.list_organization_agent_tokens(organization.id)
|
||||
assert length(agent_tokens) == 1
|
||||
assert hd(agent_tokens).name == "New Agent"
|
||||
end
|
||||
|
||||
test "creates agent with flat params format", %{conn: conn, user: user, organization: organization} do
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, view, _html} = live(conn)
|
||||
|
||||
# Trigger create_agent event with flat params
|
||||
html = render_submit(view, "create_agent", %{"name" => "Flat Agent"})
|
||||
|
||||
assert html =~ "Agent created successfully"
|
||||
assert html =~ "Flat Agent"
|
||||
end
|
||||
|
||||
test "shows error on creation failure", %{conn: conn, user: user, organization: organization} do
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, view, _html} = live(conn)
|
||||
|
||||
# Trigger create_agent with invalid params (empty name)
|
||||
html = render_submit(view, "create_agent", %{"agent_form" => %{"name" => ""}})
|
||||
|
||||
assert html =~ "Failed to create agent"
|
||||
end
|
||||
end
|
||||
|
||||
describe "close_token_modal event" do
|
||||
test "closes the token modal", %{conn: conn, user: user, organization: organization} do
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, view, _html} = live(conn)
|
||||
|
||||
# First create an agent to show the modal
|
||||
render_submit(view, "create_agent", %{"name" => "Test Agent"})
|
||||
|
||||
# Now close the modal
|
||||
_html = render_click(view, "close_token_modal", %{})
|
||||
|
||||
# Modal should be closed - verify by checking that we can still interact with the view
|
||||
assert render(view) =~ "Test Agent"
|
||||
end
|
||||
end
|
||||
|
||||
describe "revoke_agent event" do
|
||||
test "revokes agent token successfully", %{conn: conn, user: user, organization: organization} do
|
||||
{:ok, agent_token, _token} = Agents.create_agent_token(organization.id, "Agent to Revoke")
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, view, _html} = live(conn)
|
||||
|
||||
html = render_click(view, "revoke_agent", %{"id" => agent_token.id})
|
||||
|
||||
assert html =~ "Agent revoked successfully"
|
||||
|
||||
# Verify agent was revoked (still in list but with enabled: false)
|
||||
agent_tokens = Agents.list_organization_agent_tokens(organization.id)
|
||||
refute Enum.empty?(agent_tokens)
|
||||
revoked_agent = hd(agent_tokens)
|
||||
refute revoked_agent.enabled
|
||||
end
|
||||
end
|
||||
|
||||
describe "agent status display" do
|
||||
test "shows Never connected for agents without last_seen_at", %{
|
||||
conn: conn,
|
||||
user: user,
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, agent_token, _token} = Agents.create_agent_token(organization.id, "Never Seen")
|
||||
|
||||
# Verify the agent has no last_seen_at
|
||||
assert is_nil(agent_token.last_seen_at)
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, _view, html} = live(conn)
|
||||
|
||||
assert html =~ "Never Seen"
|
||||
assert html =~ "Never connected" or html =~ "Never"
|
||||
end
|
||||
|
||||
test "shows agent status information in the page", %{
|
||||
conn: conn,
|
||||
user: user,
|
||||
organization: organization
|
||||
} do
|
||||
{:ok, agent_token, _token} = Agents.create_agent_token(organization.id, "Status Agent")
|
||||
|
||||
# Update the agent to have a recent last_seen_at
|
||||
agent_token
|
||||
|> Ecto.Changeset.change(%{last_seen_at: DateTime.truncate(DateTime.utc_now(), :second)})
|
||||
|> Towerops.Repo.update!()
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, _view, html} = live(conn)
|
||||
|
||||
assert html =~ "Status Agent"
|
||||
# The status should show something about timing (Online, seconds ago, etc.)
|
||||
assert html =~ "ago" or html =~ "Online"
|
||||
end
|
||||
end
|
||||
|
||||
describe "handle_params" do
|
||||
test "handles :index action", %{conn: conn, user: user, organization: organization} do
|
||||
conn =
|
||||
conn
|
||||
|> log_in_user(user)
|
||||
|> get("/orgs/#{organization.slug}/agents")
|
||||
|
||||
{:ok, view, _html} = live(conn)
|
||||
|
||||
# Verify we can navigate to the index action
|
||||
assert render(view) =~ "Remote Agents"
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -290,4 +290,174 @@ defmodule ToweropsWeb.UserAuthTest do
|
|||
refute conn.status
|
||||
end
|
||||
end
|
||||
|
||||
describe "require_superuser/2" do
|
||||
test "allows superusers", %{conn: conn} do
|
||||
superuser =
|
||||
user_fixture()
|
||||
|> Ecto.Changeset.change(%{is_superuser: true})
|
||||
|> Towerops.Repo.update!()
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(superuser))
|
||||
|> UserAuth.require_superuser([])
|
||||
|
||||
refute conn.halted
|
||||
end
|
||||
|
||||
test "redirects non-superusers", %{conn: conn, user: user} do
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(user))
|
||||
|> UserAuth.require_superuser([])
|
||||
|
||||
assert conn.halted
|
||||
assert redirected_to(conn) == ~p"/orgs"
|
||||
|
||||
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
|
||||
"You must be a superuser to access this page."
|
||||
end
|
||||
|
||||
test "redirects if not authenticated", %{conn: conn} do
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(nil))
|
||||
|> UserAuth.require_superuser([])
|
||||
|
||||
assert conn.halted
|
||||
assert redirected_to(conn) == ~p"/orgs"
|
||||
end
|
||||
end
|
||||
|
||||
describe "load_current_organization/2" do
|
||||
test "loads organization for member", %{conn: conn, user: user} do
|
||||
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(user))
|
||||
|> Map.put(:path_params, %{"org_slug" => organization.slug})
|
||||
|> UserAuth.load_current_organization([])
|
||||
|
||||
refute conn.halted
|
||||
assert conn.assigns.current_organization.id == organization.id
|
||||
assert conn.assigns.current_membership
|
||||
end
|
||||
|
||||
test "redirects for non-member", %{conn: conn, user: user} do
|
||||
other_user = user_fixture()
|
||||
{:ok, organization} = Towerops.Organizations.create_organization(%{name: "Other Org"}, other_user.id)
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(user))
|
||||
|> Map.put(:path_params, %{"org_slug" => organization.slug})
|
||||
|> UserAuth.load_current_organization([])
|
||||
|
||||
assert conn.halted
|
||||
assert redirected_to(conn) == ~p"/orgs"
|
||||
|
||||
assert Phoenix.Flash.get(conn.assigns.flash, :error) ==
|
||||
"You don't have access to this organization."
|
||||
end
|
||||
|
||||
test "redirects for non-existent organization", %{conn: conn, user: user} do
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(user))
|
||||
|> Map.put(:path_params, %{"org_slug" => "non-existent"})
|
||||
|> UserAuth.load_current_organization([])
|
||||
|
||||
assert conn.halted
|
||||
assert redirected_to(conn) == ~p"/orgs"
|
||||
assert Phoenix.Flash.get(conn.assigns.flash, :error) == "Organization not found."
|
||||
end
|
||||
|
||||
test "redirects if no org_slug", %{conn: conn, user: user} do
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(user))
|
||||
|> Map.put(:path_params, %{})
|
||||
|> UserAuth.load_current_organization([])
|
||||
|
||||
assert conn.halted
|
||||
assert redirected_to(conn) == ~p"/orgs"
|
||||
end
|
||||
|
||||
test "redirects if no user", %{conn: conn} do
|
||||
conn =
|
||||
conn
|
||||
|> fetch_flash()
|
||||
|> assign(:current_scope, Scope.for_user(nil))
|
||||
|> Map.put(:path_params, %{"org_slug" => "test"})
|
||||
|> UserAuth.load_current_organization([])
|
||||
|
||||
assert conn.halted
|
||||
assert redirected_to(conn) == ~p"/orgs"
|
||||
end
|
||||
end
|
||||
|
||||
describe "fetch_current_scope_for_user/2 with impersonation" do
|
||||
test "handles valid impersonation", %{conn: conn} do
|
||||
superuser = user_fixture(%{is_superuser: true})
|
||||
target_user = user_fixture()
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> put_session(:impersonating, true)
|
||||
|> put_session(:superuser_id, superuser.id)
|
||||
|> put_session(:target_user_id, target_user.id)
|
||||
|> UserAuth.fetch_current_scope_for_user([])
|
||||
|
||||
assert conn.assigns.current_scope.user.id == target_user.id
|
||||
assert conn.assigns.current_scope.superuser.id == superuser.id
|
||||
end
|
||||
|
||||
test "clears invalid impersonation - missing superuser", %{conn: conn} do
|
||||
target_user = user_fixture()
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> put_session(:impersonating, true)
|
||||
|> put_session(:superuser_id, Ecto.UUID.generate())
|
||||
|> put_session(:target_user_id, target_user.id)
|
||||
|> UserAuth.fetch_current_scope_for_user([])
|
||||
|
||||
refute get_session(conn, :impersonating)
|
||||
refute get_session(conn, :superuser_id)
|
||||
refute get_session(conn, :target_user_id)
|
||||
end
|
||||
|
||||
test "clears invalid impersonation - missing target", %{conn: conn} do
|
||||
superuser = user_fixture(%{is_superuser: true})
|
||||
|
||||
conn =
|
||||
conn
|
||||
|> put_session(:impersonating, true)
|
||||
|> put_session(:superuser_id, superuser.id)
|
||||
|> put_session(:target_user_id, Ecto.UUID.generate())
|
||||
|> UserAuth.fetch_current_scope_for_user([])
|
||||
|
||||
refute get_session(conn, :impersonating)
|
||||
refute get_session(conn, :superuser_id)
|
||||
refute get_session(conn, :target_user_id)
|
||||
end
|
||||
|
||||
test "clears invalid impersonation - missing IDs", %{conn: conn} do
|
||||
conn =
|
||||
conn
|
||||
|> put_session(:impersonating, true)
|
||||
|> UserAuth.fetch_current_scope_for_user([])
|
||||
|
||||
refute get_session(conn, :impersonating)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue