470 lines
18 KiB
Python
470 lines
18 KiB
Python
"""GET and PUT /api/org/modules/: who may read and change an org's modules,
|
|
the 400 envelope, the refusals, and what a change leaves alone."""
|
|
|
|
import pytest
|
|
|
|
from accounts.models import Account
|
|
from cases.models import Case, InboundMailbox
|
|
from common.audit_log import SecurityAuditLog
|
|
from common.models import User
|
|
from common.modules import KEYS
|
|
from common.testing import (
|
|
_make_authenticated_client,
|
|
clear_rls_context,
|
|
set_modules_off,
|
|
)
|
|
from invoices.models import Invoice
|
|
from leads.models import Lead
|
|
from tasks.models import Task
|
|
from webforms.models import WebForm
|
|
|
|
URL = "/api/org/modules/"
|
|
ROW_KEYS = {"key", "label", "group", "covers", "stops", "enabled", "available", "needs"}
|
|
DEFAULT_ORDER = [
|
|
"deals", "leads", "goals", "tasks", "tickets", "knowledge_base", "documents",
|
|
"invoices", "timesheet",
|
|
] # fmt: skip
|
|
|
|
|
|
def put(client, off, order=()):
|
|
return client.put(URL, {"off": list(off), "order": list(order)}, format="json")
|
|
|
|
|
|
def rows(response):
|
|
return {r["key"]: r for r in response.json()["modules"]}
|
|
|
|
|
|
def form(org, name, target="lead", published=True):
|
|
return WebForm.objects.create(
|
|
org=org, name=name, target=target, is_published=published
|
|
)
|
|
|
|
|
|
class TestRead:
|
|
def test_any_member_reads_the_modules(self, user_client, admin_client):
|
|
response = user_client.get(URL)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert [r["key"] for r in body["modules"]] == DEFAULT_ORDER
|
|
assert all(set(r) == ROW_KEYS for r in body["modules"])
|
|
assert [r["group"] for r in body["modules"]] == (
|
|
["sell"] * 3 + ["serve"] * 4 + ["bill"] * 2
|
|
)
|
|
assert body["can_edit"] is False
|
|
assert admin_client.get(URL).json()["can_edit"] is True
|
|
|
|
def test_the_rows_carry_the_contract_values(self, user_client):
|
|
goals = rows(user_client.get(URL))["goals"]
|
|
assert goals == {
|
|
"key": "goals",
|
|
"label": "Goals",
|
|
"group": "sell",
|
|
"covers": "Sales goals, the leaderboard and goal history.",
|
|
"stops": "Goals leave every menu, and goal milestone emails stop.",
|
|
"enabled": True,
|
|
"available": True,
|
|
"needs": "deals",
|
|
}
|
|
|
|
def test_no_session_is_403(self, unauthenticated_client):
|
|
for response in (
|
|
unauthenticated_client.get(URL),
|
|
put(unauthenticated_client, []),
|
|
):
|
|
assert response.status_code == 403
|
|
assert response.json() == {
|
|
"detail": "Authentication credentials were not provided."
|
|
}
|
|
|
|
def test_a_session_with_no_org_is_403(self, db):
|
|
user = User.objects.create_user(email="lone@test.com", password="x")
|
|
client = _make_authenticated_client(user, None, None)
|
|
assert client.get(URL).status_code == 403
|
|
|
|
def test_patch_is_405(self, admin_client):
|
|
assert admin_client.patch(URL, {"off": []}, format="json").status_code == 405
|
|
|
|
def test_labels_and_stops_follow_terminology_and_never_leak_the_template(
|
|
self, admin_client, org_a
|
|
):
|
|
for terminology in (
|
|
{},
|
|
{"lead.plural": "Enquiries", "opportunity.plural": "Jobs"},
|
|
):
|
|
org_a.terminology = terminology
|
|
org_a.save(update_fields=["terminology"])
|
|
for response in (admin_client.get(URL), put(admin_client, [], KEYS)):
|
|
body = response.json()
|
|
assert all("{" not in r["stops"] for r in body["modules"])
|
|
assert all("{label}" not in r["stops"] for r in body["modules"])
|
|
by_key = rows(admin_client.get(URL))
|
|
assert by_key["leads"]["label"] == "Enquiries"
|
|
assert (
|
|
by_key["leads"]["stops"] == "Enquiries leave every menu, search and Today."
|
|
)
|
|
assert by_key["deals"]["label"] == "Jobs"
|
|
assert by_key["deals"]["stops"].startswith("Everything in Jobs leaves")
|
|
assert by_key["invoices"]["stops"].endswith(
|
|
"Payment reminders that fall due while Invoices is off are not sent later."
|
|
)
|
|
assert by_key["invoices"]["stops"].count("Invoices") == 2
|
|
|
|
|
|
class TestWho:
|
|
def test_a_member_cannot_switch_modules(self, user_client, org_a):
|
|
response = put(user_client, ["tickets"])
|
|
assert response.status_code == 403
|
|
assert response.json() == {
|
|
"error": True,
|
|
"errors": "Only admins can switch modules on or off.",
|
|
}
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == []
|
|
assert not SecurityAuditLog.objects.filter(event_type="MODULES_CHANGED")
|
|
|
|
def test_a_member_is_refused_before_the_body_is_read(self, user_client):
|
|
response = user_client.put(URL, {"off": "nonsense"}, format="json")
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestSwitching:
|
|
def test_an_admin_switches_tickets_off_and_on(self, admin_client, org_a):
|
|
response = put(admin_client, ["tickets"])
|
|
assert response.status_code == 200
|
|
assert response.json() == admin_client.get(URL).json()
|
|
by_key = rows(response)
|
|
assert by_key["tickets"]["enabled"] is False
|
|
assert by_key["timesheet"]["enabled"] is True
|
|
assert by_key["timesheet"]["available"] is False
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == ["tickets"]
|
|
assert org_a.modules_on_at == {}
|
|
|
|
by_key = rows(put(admin_client, []))
|
|
assert by_key["tickets"]["available"] and by_key["timesheet"]["available"]
|
|
org_a.refresh_from_db()
|
|
assert set(org_a.modules_on_at) == {"tickets", "timesheet"}
|
|
stamps = dict(org_a.modules_on_at)
|
|
|
|
assert put(admin_client, []).status_code == 200
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_on_at == stamps
|
|
|
|
def test_switching_deals_off_keeps_the_goals_setting(self, admin_client):
|
|
response = put(admin_client, ["deals"])
|
|
assert response.status_code == 200
|
|
goals = rows(response)["goals"]
|
|
assert goals["enabled"] is True
|
|
assert goals["available"] is False
|
|
|
|
def test_goals_cannot_switch_on_while_deals_is_off(self, admin_client, org_a):
|
|
set_modules_off(org_a, "deals", "goals")
|
|
response = put(admin_client, ["deals"])
|
|
assert response.status_code == 400
|
|
assert response.json() == {
|
|
"off": ["Goals needs Pipeline. Switch Pipeline on first."]
|
|
}
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == ["deals", "goals"]
|
|
response = put(admin_client, [])
|
|
assert response.status_code == 200
|
|
assert rows(response)["goals"]["available"] is True
|
|
|
|
NEEDS = {"off": ["Goals needs Pipeline. Switch Pipeline on first."]}
|
|
|
|
def test_goals_cannot_switch_on_as_pipeline_switches_off(self, admin_client, org_a):
|
|
set_modules_off(org_a, "goals")
|
|
response = put(admin_client, ["deals"])
|
|
assert response.status_code == 400
|
|
assert response.json() == self.NEEDS
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == ["goals"]
|
|
|
|
@pytest.mark.parametrize(
|
|
"stored, sent, status",
|
|
[
|
|
([], ["deals"], 200),
|
|
(["goals", "deals"], ["goals"], 200),
|
|
(["goals", "deals"], [], 200),
|
|
(["goals", "deals"], ["deals"], 400),
|
|
],
|
|
)
|
|
def test_goals_needs_pipeline_on_the_new_off_list(
|
|
self, admin_client, org_a, stored, sent, status
|
|
):
|
|
set_modules_off(org_a, *stored)
|
|
response = put(admin_client, sent)
|
|
assert response.status_code == status
|
|
if status == 400:
|
|
assert response.json() == self.NEEDS
|
|
else:
|
|
goals = rows(response)["goals"]
|
|
assert (goals["enabled"], goals["available"]) == (
|
|
"goals" not in sent,
|
|
not sent,
|
|
)
|
|
|
|
def test_timesheet_needs_tickets_the_same_way(self, admin_client, org_a):
|
|
needs = {"off": ["Timesheet needs Tickets. Switch Tickets on first."]}
|
|
set_modules_off(org_a, "timesheet")
|
|
response = put(admin_client, ["tickets"])
|
|
assert response.status_code == 400 and response.json() == needs
|
|
set_modules_off(org_a, "timesheet", "tickets")
|
|
assert put(admin_client, ["tickets"]).json() == needs
|
|
assert put(admin_client, ["timesheet"]).status_code == 200
|
|
assert put(admin_client, []).status_code == 200
|
|
assert put(admin_client, ["tickets"]).status_code == 200
|
|
assert rows(admin_client.get(URL))["timesheet"]["available"] is False
|
|
|
|
def test_the_order_is_saved_and_returned_within_groups(self, admin_client, org_a):
|
|
order = [
|
|
"goals",
|
|
"leads",
|
|
"deals",
|
|
"documents",
|
|
"tasks",
|
|
"timesheet",
|
|
"invoices",
|
|
]
|
|
response = put(admin_client, [], order)
|
|
assert response.status_code == 200
|
|
assert [r["key"] for r in response.json()["modules"]] == [
|
|
"goals", "leads", "deals",
|
|
"documents", "tasks", "tickets", "knowledge_base",
|
|
"timesheet", "invoices",
|
|
] # fmt: skip
|
|
org_a.refresh_from_db()
|
|
assert org_a.module_order == order
|
|
assert [r["key"] for r in admin_client.get(URL).json()["modules"]][:3] == [
|
|
"goals", "leads", "deals",
|
|
] # fmt: skip
|
|
|
|
def test_a_module_already_off_does_not_block_a_reorder(self, admin_client, org_a):
|
|
set_modules_off(org_a, "tickets")
|
|
form(org_a, "Support", target="ticket")
|
|
response = put(admin_client, ["tickets"], ["leads", "deals"])
|
|
assert response.status_code == 200
|
|
assert [r["key"] for r in response.json()["modules"]][:2] == ["leads", "deals"]
|
|
|
|
|
|
class TestMalformed:
|
|
@pytest.mark.parametrize(
|
|
"body, errors",
|
|
[
|
|
({"order": []}, {"off": ["This field is required."]}),
|
|
({"off": []}, {"order": ["This field is required."]}),
|
|
({"off": "tickets", "order": []}, {"off": ["Send a list of module keys."]}),
|
|
({"off": [3], "order": []}, {"off": ["Send a list of module keys."]}),
|
|
({"off": [], "order": "leads"}, {"order": ["Send a list of module keys."]}),
|
|
({"off": ["crm"], "order": []}, {"off": ['Unknown module: "crm".']}),
|
|
(
|
|
{"off": [], "order": ["leads", "leads"]},
|
|
{"order": ["Name each module once."]},
|
|
),
|
|
(
|
|
{"off": ["x", "tickets"], "order": ["leads", "leads", "y"]},
|
|
{"off": ['Unknown module: "x".'], "order": ['Unknown module: "y".']},
|
|
),
|
|
(
|
|
[],
|
|
{
|
|
"off": ["This field is required."],
|
|
"order": ["This field is required."],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
def test_malformed_bodies_are_400(self, admin_client, org_a, body, errors):
|
|
response = admin_client.put(URL, body, format="json")
|
|
assert response.status_code == 400
|
|
assert response.json() == errors
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == [] and org_a.module_order == []
|
|
assert org_a.modules_on_at == {}
|
|
assert not SecurityAuditLog.objects.filter(event_type="MODULES_CHANGED")
|
|
|
|
|
|
class TestRefusals:
|
|
def test_published_forms_refuse_switching_their_module_off(
|
|
self, admin_client, org_a
|
|
):
|
|
contact = form(org_a, "Contact us")
|
|
form(org_a, "Draft", published=False)
|
|
response = put(admin_client, ["leads"])
|
|
assert response.status_code == 400
|
|
assert response.json() == {
|
|
"off": [
|
|
"Leads still receives submissions from the published web form "
|
|
'"Contact us". Unpublish it first, so no submission is lost.'
|
|
]
|
|
}
|
|
form(org_a, "Zed help", target="ticket")
|
|
form(org_a, "Billing help", target="ticket")
|
|
response = put(admin_client, ["tickets"])
|
|
assert response.json() == {
|
|
"off": [
|
|
"Tickets still receives submissions from 2 published web forms: "
|
|
'"Billing help", "Zed help". Unpublish them first, so no '
|
|
"submission is lost."
|
|
]
|
|
}
|
|
WebForm.objects.filter(org=org_a).update(is_published=False)
|
|
assert put(admin_client, ["leads", "tickets"]).status_code == 200
|
|
assert contact.target_module == "leads"
|
|
|
|
def test_many_forms_are_capped_at_five_names(self, admin_client, org_a):
|
|
for n in range(7):
|
|
form(org_a, f"Form {n}")
|
|
message = put(admin_client, ["leads"]).json()["off"][0]
|
|
assert '"Form 4" and 2 more.' in message
|
|
assert "7 published web forms" in message
|
|
|
|
def test_an_active_mailbox_refuses_switching_tickets_off(self, admin_client, org_a):
|
|
InboundMailbox.objects.create(
|
|
org=org_a, address="off@acme.test", is_active=False
|
|
)
|
|
assert put(admin_client, ["tickets"]).status_code == 200
|
|
put(admin_client, [])
|
|
InboundMailbox.objects.create(org=org_a, address="help@acme.test")
|
|
response = put(admin_client, ["tickets"])
|
|
assert response.status_code == 400
|
|
assert response.json() == {
|
|
"off": [
|
|
"Tickets still receives email at help@acme.test. Switch that "
|
|
"mailbox off first, so no email is lost."
|
|
]
|
|
}
|
|
InboundMailbox.objects.create(org=org_a, address="billing@acme.test")
|
|
assert put(admin_client, ["tickets"]).json() == {
|
|
"off": [
|
|
"Tickets still receives email at 2 mailboxes: billing@acme.test, "
|
|
"help@acme.test. Switch them off first, so no email is lost."
|
|
]
|
|
}
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == []
|
|
|
|
def test_a_dependent_module_is_refused_too(self, admin_client, org_a):
|
|
form(org_a, "Support", target="ticket")
|
|
assert put(admin_client, ["tickets", "timesheet"]).status_code == 400
|
|
|
|
|
|
class TestAudit:
|
|
def test_a_change_is_audited_with_who_and_what(
|
|
self, admin_client, admin_user, org_a
|
|
):
|
|
put(admin_client, ["tickets"])
|
|
entries = SecurityAuditLog.objects.filter(event_type="MODULES_CHANGED")
|
|
assert entries.count() == 1
|
|
entry = entries.get()
|
|
assert entry.user == admin_user and entry.org == org_a
|
|
assert entry.metadata == {
|
|
"switched_off": ["tickets"],
|
|
"switched_on": [],
|
|
"reordered": False,
|
|
}
|
|
put(admin_client, ["tickets"])
|
|
put(admin_client, ["tickets"], DEFAULT_ORDER)
|
|
assert entries.count() == 1
|
|
put(admin_client, [], ["leads", "deals"])
|
|
assert entries.count() == 2
|
|
assert entries.order_by("-created_at").first().metadata == {
|
|
"switched_off": [],
|
|
"switched_on": ["tickets"],
|
|
"reordered": True,
|
|
}
|
|
|
|
def test_the_entry_carries_the_request_and_the_audit_api_returns_it(
|
|
self, admin_client, org_a
|
|
):
|
|
admin_client.put(
|
|
URL,
|
|
{"off": ["tickets"], "order": []},
|
|
format="json",
|
|
HTTP_USER_AGENT="ModulesTest/1.0",
|
|
REMOTE_ADDR="203.0.113.9",
|
|
)
|
|
entry = SecurityAuditLog.objects.get(event_type="MODULES_CHANGED")
|
|
assert entry.ip_address == "203.0.113.9"
|
|
assert entry.user_agent == "ModulesTest/1.0"
|
|
assert entry.request_path == URL
|
|
listed = admin_client.get(
|
|
"/api/org/audit-log/", {"event_type": "MODULES_CHANGED"}
|
|
).json()["results"]
|
|
assert len(listed) == 1
|
|
assert listed[0]["details"] == {
|
|
"switched_off": ["tickets"],
|
|
"switched_on": [],
|
|
"reordered": False,
|
|
}
|
|
assert listed[0]["ip_address"] == "203.0.113.9"
|
|
assert listed[0]["request_path"] == URL
|
|
|
|
|
|
class TestLeavesAlone:
|
|
def test_switching_off_changes_no_record(self, admin_client, org_a, admin_profile):
|
|
account = Account.objects.create(name="Acme", org=org_a)
|
|
lead = Lead.objects.create(
|
|
title="Lead", first_name="A", last_name="B", org=org_a
|
|
)
|
|
case = Case.objects.create(
|
|
name="Ticket", status="New", priority="Normal", org=org_a, account=account
|
|
)
|
|
invoice = Invoice.objects.create(
|
|
org=org_a, account=account, invoice_title="Work"
|
|
)
|
|
task = Task.objects.create(
|
|
title="Task", status="New", priority="High", org=org_a
|
|
)
|
|
records = [lead, case, invoice, task]
|
|
before = {r.pk: r.updated_at for r in records}
|
|
off = ["leads", "tickets", "invoices", "tasks"]
|
|
assert put(admin_client, off).status_code == 200
|
|
org_a.refresh_from_db()
|
|
assert set(org_a.modules_off) == set(off)
|
|
assert put(admin_client, []).status_code == 200
|
|
for r in records:
|
|
r.refresh_from_db()
|
|
assert r.updated_at == before[r.pk]
|
|
assert Lead.objects.count() == Case.objects.count() == 1
|
|
assert Invoice.objects.count() == Task.objects.count() == 1
|
|
|
|
def test_org_settings_cannot_write_module_fields(self, admin_client, org_a):
|
|
response = admin_client.patch(
|
|
"/api/org/settings/",
|
|
{"modules_off": ["leads"], "module_order": ["leads"]},
|
|
format="json",
|
|
)
|
|
assert response.status_code == 200
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == [] and org_a.module_order == []
|
|
|
|
def test_another_org_is_untouched(self, admin_client, org_a, org_b):
|
|
set_modules_off(org_b, "goals")
|
|
assert put(admin_client, ["tickets"], ["leads", "deals"]).status_code == 200
|
|
org_a.refresh_from_db()
|
|
assert org_a.modules_off == ["tickets"]
|
|
assert org_a.module_order == ["leads", "deals"]
|
|
org_b.refresh_from_db()
|
|
assert org_b.modules_off == ["goals"]
|
|
assert org_b.module_order == [] and org_b.modules_on_at == {}
|
|
assert not SecurityAuditLog.objects.filter(org=org_b)
|
|
|
|
|
|
@pytest.mark.postgres_only
|
|
def test_refusals_see_rows_under_rls(admin_client, org_a):
|
|
"""Skipped on SQLite. Under a non-superuser role an empty context matches no
|
|
rows, so a refusal proves the request set the org context itself."""
|
|
from django.db import connection
|
|
|
|
if connection.vendor != "postgresql":
|
|
pytest.skip("RLS requires PostgreSQL")
|
|
form(org_a, "Contact us")
|
|
InboundMailbox.objects.create(org=org_a, address="help@acme.test")
|
|
clear_rls_context()
|
|
response = put(admin_client, ["leads"])
|
|
assert response.status_code == 400
|
|
assert "Contact us" in response.json()["off"][0]
|
|
response = put(admin_client, ["tickets"])
|
|
assert response.status_code == 400
|
|
assert "help@acme.test" in response.json()["off"][0]
|