893 lines
28 KiB
Elixir
893 lines
28 KiB
Elixir
defmodule Towerops.AdminTest do
|
|
use Towerops.DataCase
|
|
|
|
import Towerops.AccountsFixtures
|
|
import Towerops.OrganizationsFixtures
|
|
|
|
alias Towerops.Accounts.User
|
|
alias Towerops.Admin
|
|
alias Towerops.Admin.AuditLog
|
|
alias Towerops.Billing.StripeClient
|
|
alias Towerops.Organizations.Organization
|
|
alias Towerops.Settings
|
|
alias Towerops.Settings.ApplicationSetting
|
|
|
|
doctest Admin, except: [delete_user: 3, delete_organization: 3, create_audit_log: 1]
|
|
|
|
describe "create_audit_log/1" do
|
|
test "creates audit log with valid attributes" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
target_user = user_fixture()
|
|
|
|
attrs = %{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "192.168.1.1"
|
|
}
|
|
|
|
assert {:ok, audit_log} = Admin.create_audit_log(attrs)
|
|
assert audit_log.action == "impersonate_start"
|
|
assert audit_log.superuser_id == superuser.id
|
|
assert audit_log.target_user_id == target_user.id
|
|
assert audit_log.ip_address.address == "192.168.1.1"
|
|
end
|
|
|
|
test "creates audit log with nil target_user_id" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
|
|
attrs = %{
|
|
action: "user_data_exported",
|
|
superuser_id: superuser.id,
|
|
target_user_id: nil,
|
|
ip_address: "10.0.0.1"
|
|
}
|
|
|
|
assert {:ok, audit_log} = Admin.create_audit_log(attrs)
|
|
assert audit_log.target_user_id == nil
|
|
end
|
|
|
|
test "creates audit log with only required field (action)" do
|
|
attrs = %{
|
|
action: "impersonate_start"
|
|
# Only action is required, superuser_id and ip_address are optional
|
|
}
|
|
|
|
assert {:ok, audit_log} = Admin.create_audit_log(attrs)
|
|
assert audit_log.action == "impersonate_start"
|
|
assert audit_log.superuser_id == nil
|
|
assert audit_log.ip_address == nil
|
|
end
|
|
|
|
test "returns error with invalid action" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
|
|
attrs = %{
|
|
action: "invalid_action",
|
|
superuser_id: superuser.id,
|
|
ip_address: "127.0.0.1"
|
|
}
|
|
|
|
assert {:error, changeset} = Admin.create_audit_log(attrs)
|
|
assert %{action: ["is invalid"]} = errors_on(changeset)
|
|
end
|
|
|
|
test "returns error with missing action" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
|
|
attrs = %{
|
|
superuser_id: superuser.id,
|
|
ip_address: "127.0.0.1"
|
|
}
|
|
|
|
assert {:error, changeset} = Admin.create_audit_log(attrs)
|
|
assert %{action: ["can't be blank"]} = errors_on(changeset)
|
|
end
|
|
end
|
|
|
|
describe "list_audit_logs/1" do
|
|
setup do
|
|
superuser = user_fixture(%{superuser: true})
|
|
target_user = user_fixture()
|
|
|
|
%{superuser: superuser, target_user: target_user}
|
|
end
|
|
|
|
test "returns all audit logs with default pagination", %{superuser: superuser, target_user: target_user} do
|
|
{:ok, _log1} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
{:ok, _log2} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_end",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs()
|
|
assert length(result) == 2
|
|
end
|
|
|
|
test "respects limit parameter", %{superuser: superuser, target_user: target_user} do
|
|
for _i <- 1..5 do
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
end
|
|
|
|
result = Admin.list_audit_logs(limit: 3)
|
|
assert length(result) == 3
|
|
end
|
|
|
|
test "orders logs by inserted_at descending", %{superuser: superuser, target_user: target_user} do
|
|
{:ok, _log1} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
{:ok, _log2} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_end",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs()
|
|
assert length(result) == 2
|
|
|
|
# Verify ordering: each record's inserted_at should be >= next record
|
|
[first, second] = result
|
|
assert DateTime.compare(first.inserted_at, second.inserted_at) in [:gt, :eq]
|
|
end
|
|
|
|
test "preloads superuser and target_user associations", %{superuser: superuser, target_user: target_user} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs()
|
|
assert length(result) == 1
|
|
|
|
log = hd(result)
|
|
assert Ecto.assoc_loaded?(log.superuser)
|
|
assert Ecto.assoc_loaded?(log.target_user)
|
|
end
|
|
|
|
test "returns empty list when no audit logs exist" do
|
|
result = Admin.list_audit_logs()
|
|
assert result == []
|
|
end
|
|
end
|
|
|
|
describe "list_all_users/1" do
|
|
test "returns all users with default pagination" do
|
|
user1 = user_fixture()
|
|
user2 = user_fixture()
|
|
user3 = user_fixture()
|
|
|
|
result = Admin.list_all_users()
|
|
assert length(result) >= 3
|
|
user_ids = Enum.map(result, & &1.id)
|
|
assert user1.id in user_ids
|
|
assert user2.id in user_ids
|
|
assert user3.id in user_ids
|
|
end
|
|
|
|
test "respects limit parameter" do
|
|
for _i <- 1..10 do
|
|
user_fixture()
|
|
end
|
|
|
|
result = Admin.list_all_users(limit: 5)
|
|
assert length(result) == 5
|
|
end
|
|
|
|
test "respects offset parameter" do
|
|
# Create users in a clean test
|
|
Repo.delete_all(User)
|
|
_users = for _i <- 1..10, do: user_fixture()
|
|
|
|
result = Admin.list_all_users(offset: 7)
|
|
assert length(result) == 3
|
|
end
|
|
|
|
test "combines limit and offset" do
|
|
Repo.delete_all(User)
|
|
|
|
for _i <- 1..20 do
|
|
user_fixture()
|
|
end
|
|
|
|
result = Admin.list_all_users(limit: 5, offset: 10)
|
|
assert length(result) == 5
|
|
end
|
|
|
|
test "orders users by inserted_at descending" do
|
|
Repo.delete_all(User)
|
|
|
|
_user1 = user_fixture()
|
|
_user2 = user_fixture()
|
|
_user3 = user_fixture()
|
|
|
|
result = Admin.list_all_users()
|
|
assert length(result) == 3
|
|
|
|
# Verify descending order: each user's inserted_at should be >= next user
|
|
[first, second, third] = result
|
|
assert DateTime.compare(first.inserted_at, second.inserted_at) in [:gt, :eq]
|
|
assert DateTime.compare(second.inserted_at, third.inserted_at) in [:gt, :eq]
|
|
end
|
|
|
|
test "returns empty list when no users exist" do
|
|
# Clean up any existing users
|
|
Repo.delete_all(User)
|
|
|
|
result = Admin.list_all_users()
|
|
assert result == []
|
|
end
|
|
|
|
test "preloads organizations association" do
|
|
user = user_fixture()
|
|
|
|
result = Admin.list_all_users()
|
|
found_user = Enum.find(result, &(&1.id == user.id))
|
|
|
|
assert Ecto.assoc_loaded?(found_user.organizations)
|
|
end
|
|
end
|
|
|
|
describe "count_users/0" do
|
|
test "returns count of all users" do
|
|
Repo.delete_all(User)
|
|
|
|
assert Admin.count_users() == 0
|
|
|
|
user_fixture()
|
|
assert Admin.count_users() == 1
|
|
|
|
user_fixture()
|
|
user_fixture()
|
|
assert Admin.count_users() == 3
|
|
end
|
|
end
|
|
|
|
describe "delete_user/3" do
|
|
test "deletes user and creates audit log" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
target_user = user_fixture()
|
|
target_user_id = target_user.id
|
|
target_user_email = target_user.email
|
|
|
|
assert {:ok, _deleted_user} = Admin.delete_user(target_user_id, superuser.id, "192.168.1.1")
|
|
|
|
# User should be deleted
|
|
refute Repo.get(User, target_user_id)
|
|
|
|
# Audit log should be created
|
|
audit_logs = Admin.list_audit_logs()
|
|
assert length(audit_logs) == 1
|
|
log = hd(audit_logs)
|
|
assert log.action == "user_delete"
|
|
assert log.superuser_id == superuser.id
|
|
# target_user_id may be nil due to foreign key constraints after deletion
|
|
assert log.ip_address.address == "192.168.1.1"
|
|
assert log.metadata["email"] == target_user_email
|
|
end
|
|
|
|
test "returns error when user does not exist" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
non_existent_id = Ecto.UUID.generate()
|
|
|
|
assert {:error, :not_found} = Admin.delete_user(non_existent_id, superuser.id, "127.0.0.1")
|
|
end
|
|
|
|
test "anonymizes login history before deletion" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
target_user = user_fixture()
|
|
|
|
# Create a login attempt for the target user
|
|
{:ok, _login} =
|
|
Towerops.Accounts.record_login_attempt(%{
|
|
user_id: target_user.id,
|
|
email: target_user.email,
|
|
success: true,
|
|
method: "password",
|
|
ip_address: "10.0.0.1"
|
|
})
|
|
|
|
assert {:ok, _deleted_user} = Admin.delete_user(target_user.id, superuser.id, "192.168.1.1")
|
|
|
|
# User should be deleted
|
|
refute Repo.get(User, target_user.id)
|
|
|
|
# Login attempt should be anonymized (user_id set to nil, anonymized_at set)
|
|
login_attempt = Repo.get_by(Towerops.Accounts.LoginAttempt, email: target_user.email)
|
|
|
|
if login_attempt do
|
|
assert is_nil(login_attempt.user_id)
|
|
assert login_attempt.anonymized_at
|
|
end
|
|
end
|
|
end
|
|
|
|
describe "list_all_organizations/1" do
|
|
test "returns all organizations with default pagination" do
|
|
# Each user can only create one free organization, so use different users
|
|
user1 = user_fixture()
|
|
user2 = user_fixture()
|
|
user3 = user_fixture()
|
|
{:ok, org1} = Towerops.Organizations.create_organization(%{name: "Org 1"}, user1.id)
|
|
{:ok, org2} = Towerops.Organizations.create_organization(%{name: "Org 2"}, user2.id)
|
|
{:ok, org3} = Towerops.Organizations.create_organization(%{name: "Org 3"}, user3.id)
|
|
|
|
result = Admin.list_all_organizations()
|
|
assert length(result) >= 3
|
|
org_ids = Enum.map(result, & &1.id)
|
|
assert org1.id in org_ids
|
|
assert org2.id in org_ids
|
|
assert org3.id in org_ids
|
|
end
|
|
|
|
test "respects limit parameter" do
|
|
for i <- 1..10 do
|
|
user = user_fixture()
|
|
Towerops.Organizations.create_organization(%{name: "Org #{i}"}, user.id)
|
|
end
|
|
|
|
result = Admin.list_all_organizations(limit: 5)
|
|
assert length(result) == 5
|
|
end
|
|
|
|
test "respects offset parameter" do
|
|
Repo.delete_all(Organization)
|
|
|
|
_orgs =
|
|
for i <- 1..10 do
|
|
user = user_fixture()
|
|
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Org #{i}"}, user.id)
|
|
org
|
|
end
|
|
|
|
result = Admin.list_all_organizations(offset: 7)
|
|
assert length(result) == 3
|
|
end
|
|
|
|
test "combines limit and offset" do
|
|
Repo.delete_all(Organization)
|
|
|
|
for i <- 1..20 do
|
|
user = user_fixture()
|
|
Towerops.Organizations.create_organization(%{name: "Org #{i}"}, user.id)
|
|
end
|
|
|
|
result = Admin.list_all_organizations(limit: 5, offset: 10)
|
|
assert length(result) == 5
|
|
end
|
|
|
|
test "orders organizations by inserted_at descending" do
|
|
Repo.delete_all(Organization)
|
|
|
|
user1 = user_fixture()
|
|
user2 = user_fixture()
|
|
user3 = user_fixture()
|
|
|
|
{:ok, _org1} = Towerops.Organizations.create_organization(%{name: "Org 1"}, user1.id)
|
|
{:ok, _org2} = Towerops.Organizations.create_organization(%{name: "Org 2"}, user2.id)
|
|
{:ok, _org3} = Towerops.Organizations.create_organization(%{name: "Org 3"}, user3.id)
|
|
|
|
result = Admin.list_all_organizations()
|
|
assert length(result) == 3
|
|
|
|
# Verify descending order: each org's inserted_at should be >= next org
|
|
[first, second, third] = result
|
|
assert DateTime.compare(first.inserted_at, second.inserted_at) in [:gt, :eq]
|
|
assert DateTime.compare(second.inserted_at, third.inserted_at) in [:gt, :eq]
|
|
end
|
|
|
|
test "returns empty list when no organizations exist" do
|
|
Repo.delete_all(Organization)
|
|
|
|
result = Admin.list_all_organizations()
|
|
assert result == []
|
|
end
|
|
|
|
test "preloads memberships with users" do
|
|
user = user_fixture()
|
|
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
|
|
result = Admin.list_all_organizations()
|
|
found_org = Enum.find(result, &(&1.id == org.id))
|
|
|
|
assert Ecto.assoc_loaded?(found_org.memberships)
|
|
assert found_org.memberships != []
|
|
assert Ecto.assoc_loaded?(hd(found_org.memberships).user)
|
|
end
|
|
end
|
|
|
|
describe "count_organizations/0" do
|
|
test "returns count of all organizations" do
|
|
Repo.delete_all(Organization)
|
|
|
|
assert Admin.count_organizations() == 0
|
|
|
|
user1 = user_fixture()
|
|
Towerops.Organizations.create_organization(%{name: "Org 1"}, user1.id)
|
|
assert Admin.count_organizations() == 1
|
|
|
|
user2 = user_fixture()
|
|
user3 = user_fixture()
|
|
Towerops.Organizations.create_organization(%{name: "Org 2"}, user2.id)
|
|
Towerops.Organizations.create_organization(%{name: "Org 3"}, user3.id)
|
|
assert Admin.count_organizations() == 3
|
|
end
|
|
end
|
|
|
|
describe "delete_organization/3" do
|
|
test "deletes organization and creates audit log" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
user = user_fixture()
|
|
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Test Org"}, user.id)
|
|
|
|
assert {:ok, _deleted_org} = Admin.delete_organization(org.id, superuser.id, "192.168.1.1")
|
|
|
|
# Organization should be deleted
|
|
refute Repo.get(Organization, org.id)
|
|
|
|
# Audit log should be created
|
|
audit_logs = Admin.list_audit_logs()
|
|
assert length(audit_logs) == 1
|
|
log = hd(audit_logs)
|
|
assert log.action == "org_delete"
|
|
assert log.superuser_id == superuser.id
|
|
assert log.ip_address.address == "192.168.1.1"
|
|
assert log.metadata["name"] == "Test Org"
|
|
assert log.metadata["slug"] == org.slug
|
|
end
|
|
|
|
test "returns error when organization does not exist" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
non_existent_id = Ecto.UUID.generate()
|
|
|
|
assert {:error, :not_found} = Admin.delete_organization(non_existent_id, superuser.id, "127.0.0.1")
|
|
end
|
|
end
|
|
|
|
describe "GDPR data access - list_audit_logs_for_user/2" do
|
|
setup do
|
|
user = user_fixture()
|
|
superuser = user_fixture(%{superuser: true})
|
|
target_user = user_fixture()
|
|
|
|
%{user: user, superuser: superuser, target_user: target_user}
|
|
end
|
|
|
|
test "returns audit logs where user is the target", %{superuser: superuser, target_user: target_user} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs_for_user(target_user.id)
|
|
assert length(result) == 1
|
|
assert hd(result).target_user_id == target_user.id
|
|
end
|
|
|
|
test "returns audit logs where user is the superuser", %{superuser: superuser, target_user: target_user} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs_for_user(superuser.id)
|
|
assert length(result) == 1
|
|
assert hd(result).superuser_id == superuser.id
|
|
end
|
|
|
|
test "returns audit logs for both superuser and target actions", %{
|
|
superuser: superuser,
|
|
target_user: target_user,
|
|
user: user
|
|
} do
|
|
# User is target in one action
|
|
{:ok, _log1} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
# User is superuser in another action
|
|
{:ok, _log2} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: user.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs_for_user(user.id)
|
|
assert length(result) == 2
|
|
end
|
|
|
|
test "does not return audit logs for other users", %{
|
|
superuser: superuser,
|
|
target_user: target_user,
|
|
user: user
|
|
} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
# User is not involved in this log
|
|
result = Admin.list_audit_logs_for_user(user.id)
|
|
assert result == []
|
|
end
|
|
|
|
test "respects limit parameter", %{superuser: superuser, target_user: target_user} do
|
|
# Create 5 audit logs
|
|
for _i <- 1..5 do
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
end
|
|
|
|
result = Admin.list_audit_logs_for_user(target_user.id, limit: 3)
|
|
assert length(result) == 3
|
|
end
|
|
|
|
test "defaults to limit 100", %{superuser: superuser, target_user: target_user} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
# Should not error with default limit
|
|
result = Admin.list_audit_logs_for_user(target_user.id)
|
|
assert length(result) == 1
|
|
end
|
|
|
|
test "orders logs by inserted_at descending", %{superuser: superuser, target_user: target_user} do
|
|
# Create logs with different timestamps by manipulating inserted_at directly
|
|
old_time = DateTime.add(DateTime.utc_now(), -2, :second)
|
|
|
|
{:ok, old_log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
# Update the old log's timestamp directly to simulate it being older
|
|
Towerops.Repo.update_all(
|
|
from(l in AuditLog, where: l.id == ^old_log.id),
|
|
set: [inserted_at: old_time]
|
|
)
|
|
|
|
{:ok, _new_log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_end",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs_for_user(target_user.id)
|
|
assert length(result) == 2
|
|
|
|
# Most recent should be first
|
|
[first, second] = result
|
|
|
|
# Verify ordering by timestamp
|
|
assert DateTime.after?(first.inserted_at, second.inserted_at)
|
|
|
|
# The newer log (impersonate_end) should come first
|
|
assert first.action == "impersonate_end"
|
|
assert second.action == "impersonate_start"
|
|
end
|
|
|
|
test "preloads superuser and target_user associations", %{
|
|
superuser: superuser,
|
|
target_user: target_user
|
|
} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "impersonate_start",
|
|
superuser_id: superuser.id,
|
|
target_user_id: target_user.id,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs_for_user(target_user.id)
|
|
assert length(result) == 1
|
|
|
|
log = hd(result)
|
|
assert Ecto.assoc_loaded?(log.superuser)
|
|
assert Ecto.assoc_loaded?(log.target_user)
|
|
assert log.superuser.id == superuser.id
|
|
assert log.target_user.id == target_user.id
|
|
end
|
|
|
|
test "returns empty list when user has no audit logs", %{user: user} do
|
|
result = Admin.list_audit_logs_for_user(user.id)
|
|
assert result == []
|
|
end
|
|
|
|
test "handles audit logs with nil target_user_id", %{superuser: superuser} do
|
|
{:ok, _log} =
|
|
Admin.create_audit_log(%{
|
|
action: "user_data_exported",
|
|
superuser_id: superuser.id,
|
|
target_user_id: nil,
|
|
ip_address: "127.0.0.1"
|
|
})
|
|
|
|
result = Admin.list_audit_logs_for_user(superuser.id)
|
|
assert length(result) == 1
|
|
assert hd(result).target_user_id == nil
|
|
end
|
|
end
|
|
|
|
describe "update_billing_overrides/4" do
|
|
setup do
|
|
superuser = user_fixture(%{superuser: true})
|
|
user = user_fixture()
|
|
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Billing Org"}, user.id)
|
|
|
|
%{superuser: superuser, organization: org}
|
|
end
|
|
|
|
test "updates custom_free_device_limit", %{superuser: superuser, organization: org} do
|
|
assert {:ok, updated_org} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_free_device_limit: 50},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
|
|
assert updated_org.custom_free_device_limit == 50
|
|
end
|
|
|
|
test "updates custom_price_per_device", %{superuser: superuser, organization: org} do
|
|
assert {:ok, updated_org} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_price_per_device: "0.50"},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
|
|
assert Decimal.equal?(updated_org.custom_price_per_device, Decimal.new("0.50"))
|
|
end
|
|
|
|
test "updates both fields at once", %{superuser: superuser, organization: org} do
|
|
assert {:ok, updated_org} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_free_device_limit: 25, custom_price_per_device: "0.75"},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
|
|
assert updated_org.custom_free_device_limit == 25
|
|
assert Decimal.equal?(updated_org.custom_price_per_device, Decimal.new("0.75"))
|
|
end
|
|
|
|
test "clears overrides with nil values", %{superuser: superuser, organization: org} do
|
|
# First set overrides
|
|
{:ok, _} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_free_device_limit: 50},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
|
|
# Then clear them
|
|
{:ok, updated_org} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_free_device_limit: nil},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
|
|
assert is_nil(updated_org.custom_free_device_limit)
|
|
end
|
|
|
|
test "creates audit log entry", %{superuser: superuser, organization: org} do
|
|
{:ok, _} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_free_device_limit: 50, custom_price_per_device: "0.50"},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
|
|
audit_logs = Admin.list_audit_logs()
|
|
log = Enum.find(audit_logs, &(&1.action == "org_billing_override_updated"))
|
|
|
|
assert %{superuser_id: _} = log
|
|
assert log.superuser_id == superuser.id
|
|
assert log.ip_address.address == "192.168.1.1"
|
|
assert log.metadata["organization_id"] == org.id
|
|
assert log.metadata["organization_name"] == org.name
|
|
assert log.metadata["changes"]["custom_free_device_limit"] == 50
|
|
end
|
|
|
|
test "returns error for invalid overrides", %{superuser: superuser, organization: org} do
|
|
assert {:error, _changeset} =
|
|
Admin.update_billing_overrides(
|
|
org.id,
|
|
%{custom_free_device_limit: -5},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
end
|
|
|
|
test "returns error for non-existent organization", %{superuser: superuser} do
|
|
assert {:error, :not_found} =
|
|
Admin.update_billing_overrides(
|
|
Ecto.UUID.generate(),
|
|
%{custom_free_device_limit: 50},
|
|
superuser.id,
|
|
"192.168.1.1"
|
|
)
|
|
end
|
|
end
|
|
|
|
describe "list_all_organizations/1 with device_count" do
|
|
test "includes device_count virtual field" do
|
|
user = user_fixture()
|
|
{:ok, org} = Towerops.Organizations.create_organization(%{name: "Count Org"}, user.id)
|
|
|
|
result = Admin.list_all_organizations()
|
|
found_org = Enum.find(result, &(&1.id == org.id))
|
|
|
|
assert found_org.device_count == 0
|
|
end
|
|
end
|
|
|
|
describe "update_global_pricing/3" do
|
|
test "updates settings, creates Stripe price, migrates subscriptions, and audits" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
user = user_fixture()
|
|
_org1 = organization_fixture(user.id, %{subscription_status: "active", stripe_subscription_id: "sub_1"})
|
|
_org2 = organization_fixture(user.id, %{subscription_status: "active", stripe_subscription_id: "sub_2"})
|
|
|
|
# Seed current settings
|
|
Repo.delete_all(from s in ApplicationSetting, where: s.key in ["default_free_devices", "default_price_per_device"])
|
|
{:ok, _} = Repo.insert(%ApplicationSetting{key: "default_free_devices", value: "10", value_type: "integer"})
|
|
{:ok, _} = Repo.insert(%ApplicationSetting{key: "default_price_per_device", value: "2.00", value_type: "string"})
|
|
|
|
# Mock Stripe
|
|
Req.Test.stub(StripeClient, fn conn ->
|
|
cond do
|
|
conn.method == "POST" and String.contains?(conn.request_path, "/v1/prices") ->
|
|
conn
|
|
|> Plug.Conn.put_resp_content_type("application/json")
|
|
|> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "price_new123"}))
|
|
|
|
conn.method == "GET" and String.contains?(conn.request_path, "/v1/subscriptions") ->
|
|
# Return subscription with items
|
|
conn
|
|
|> Plug.Conn.put_resp_content_type("application/json")
|
|
|> Plug.Conn.send_resp(
|
|
200,
|
|
Jason.encode!(%{
|
|
"id" => "sub_test",
|
|
"items" => %{"data" => [%{"id" => "si_test", "price" => %{"id" => "price_old"}}]}
|
|
})
|
|
)
|
|
|
|
String.contains?(conn.request_path, "/v1/subscriptions") ->
|
|
conn
|
|
|> Plug.Conn.put_resp_content_type("application/json")
|
|
|> Plug.Conn.send_resp(200, Jason.encode!(%{"id" => "sub_test"}))
|
|
|
|
true ->
|
|
conn
|
|
|> Plug.Conn.put_resp_content_type("application/json")
|
|
|> Plug.Conn.send_resp(404, Jason.encode!(%{}))
|
|
end
|
|
end)
|
|
|
|
attrs = %{
|
|
"default_free_devices" => "15",
|
|
"default_price_per_device" => "3.00"
|
|
}
|
|
|
|
assert {:ok, result} = Admin.update_global_pricing(attrs, superuser.id, "127.0.0.1")
|
|
|
|
# Verify settings updated
|
|
assert Settings.get_setting("default_free_devices") == 15
|
|
assert Settings.get_setting("default_price_per_device") == "3.00"
|
|
assert Settings.get_setting("stripe_price_id") == "price_new123"
|
|
|
|
# Verify migration results
|
|
assert result.total == 2
|
|
assert result.succeeded == 2
|
|
|
|
# Verify audit log
|
|
log =
|
|
Repo.one(
|
|
from al in AuditLog, where: al.action == "global_pricing_updated", order_by: [desc: al.inserted_at], limit: 1
|
|
)
|
|
|
|
assert log.superuser_id == superuser.id
|
|
assert log.ip_address.address == "127.0.0.1"
|
|
assert log.metadata["old_free_devices"] == 10
|
|
assert log.metadata["new_free_devices"] == 15
|
|
assert log.metadata["old_price"] == "2.00"
|
|
assert log.metadata["new_price"] == "3.00"
|
|
assert log.metadata["migration_succeeded"] == 2
|
|
end
|
|
|
|
test "returns ok when price unchanged and no migration needed" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
|
|
Repo.delete_all(from s in ApplicationSetting, where: s.key in ["default_free_devices", "default_price_per_device"])
|
|
{:ok, _} = Repo.insert(%ApplicationSetting{key: "default_free_devices", value: "10", value_type: "integer"})
|
|
{:ok, _} = Repo.insert(%ApplicationSetting{key: "default_price_per_device", value: "2.00", value_type: "string"})
|
|
|
|
# Only change free devices, not price
|
|
attrs = %{
|
|
"default_free_devices" => "15",
|
|
"default_price_per_device" => "2.00"
|
|
}
|
|
|
|
assert {:ok, result} = Admin.update_global_pricing(attrs, superuser.id, "127.0.0.1")
|
|
|
|
# No Stripe price created since price unchanged
|
|
assert is_nil(Settings.get_setting("stripe_price_id"))
|
|
assert result.total == 0
|
|
end
|
|
|
|
test "validates pricing values" do
|
|
superuser = user_fixture(%{superuser: true})
|
|
|
|
# Invalid free devices (negative)
|
|
assert {:error, :invalid_free_devices} =
|
|
Admin.update_global_pricing(%{"default_free_devices" => "-5"}, superuser.id, "127.0.0.1")
|
|
|
|
# Invalid price (too high)
|
|
assert {:error, :invalid_price} =
|
|
Admin.update_global_pricing(%{"default_price_per_device" => "1000.00"}, superuser.id, "127.0.0.1")
|
|
end
|
|
end
|
|
end
|