bottleneckcrm/backend/common/tests/test_custom_fields.py

705 lines
25 KiB
Python

"""Tests for the generic custom-fields foundation: CustomFieldDefinition model,
the validate_payload helper, and the /api/custom-fields/ admin API.
See docs/cases/tier1/custom-fields.md.
"""
from __future__ import annotations
import pytest
from cases.models import Case
from common import object_registry
from common.custom_fields import validate_definition_options, validate_payload
from common.models import CustomFieldDefinition
from common.tests.object_records import make_record
from conftest import rls_org
from leads.models import Lead
@pytest.mark.django_db
class TestValidatorHelpers:
def test_dropdown_options_must_be_nonempty_list(self):
with pytest.raises(Exception):
validate_definition_options("dropdown", None)
with pytest.raises(Exception):
validate_definition_options("dropdown", [])
def test_dropdown_options_reject_duplicates(self):
with pytest.raises(Exception):
validate_definition_options(
"dropdown",
[{"value": "S1", "label": "S1"}, {"value": "S1", "label": "Duplicate"}],
)
def test_options_forbidden_for_non_dropdown(self):
with pytest.raises(Exception):
validate_definition_options("text", [{"value": "x", "label": "y"}])
def test_options_none_ok_for_non_dropdown(self):
validate_definition_options("text", None)
validate_definition_options("number", None)
@pytest.mark.django_db
class TestValidatePayload:
def _defn(self, org, **kwargs):
defaults = {
"target_model": "Case",
"key": "severity",
"label": "Severity",
"field_type": "dropdown",
"options": [
{"value": "S1", "label": "S1"},
{"value": "S2", "label": "S2"},
],
"is_required": False,
"is_active": True,
}
defaults.update(kwargs)
return CustomFieldDefinition.objects.create(org=org, **defaults)
def test_unknown_keys_dropped(self, org_a):
self._defn(org_a)
cleaned, errors = validate_payload(
"Case", {"severity": "S1", "made_up": "nope"}, org_a
)
assert errors == {}
assert cleaned == {"severity": "S1"}
def test_dropdown_invalid_value_rejected(self, org_a):
self._defn(org_a)
cleaned, errors = validate_payload("Case", {"severity": "S99"}, org_a)
assert "severity" in errors
assert cleaned == {}
def test_required_field_missing_errors(self, org_a):
self._defn(org_a, is_required=True)
cleaned, errors = validate_payload("Case", {}, org_a)
assert errors.get("severity") == "is required"
def test_required_field_preserved_from_existing(self, org_a):
self._defn(org_a, is_required=True)
cleaned, errors = validate_payload(
"Case", {}, org_a, existing={"severity": "S1"}
)
assert errors == {}
assert cleaned == {"severity": "S1"}
def test_number_coercion(self, org_a):
self._defn(
org_a, key="hours", field_type="number", options=None, is_required=False
)
cleaned, errors = validate_payload("Case", {"hours": "3.5"}, org_a)
assert errors == {}
assert cleaned == {"hours": 3.5}
def test_number_invalid_errors(self, org_a):
self._defn(org_a, key="hours", field_type="number", options=None)
cleaned, errors = validate_payload("Case", {"hours": "abc"}, org_a)
assert "hours" in errors
def test_checkbox_string_coercion(self, org_a):
self._defn(org_a, key="vip", field_type="checkbox", options=None)
cleaned, errors = validate_payload("Case", {"vip": "true"}, org_a)
assert errors == {}
assert cleaned == {"vip": True}
def test_date_iso_validated(self, org_a):
self._defn(org_a, key="due", field_type="date", options=None)
cleaned, errors = validate_payload("Case", {"due": "2026-08-01"}, org_a)
assert errors == {}
assert cleaned == {"due": "2026-08-01"}
cleaned, errors = validate_payload("Case", {"due": "not-a-date"}, org_a)
assert "due" in errors
def test_inactive_definition_treated_as_unknown(self, org_a):
self._defn(org_a, is_active=False)
cleaned, errors = validate_payload("Case", {"severity": "S1"}, org_a)
# Inactive definition: drop silently, no error.
assert cleaned == {}
assert errors == {}
def test_inactive_definition_preserves_existing_value(self, org_a):
"""Soft-deleting a definition must NOT wipe historical values on
the next save. Admins can deactivate a field without data loss."""
self._defn(org_a, is_active=False)
cleaned, errors = validate_payload(
"Case", {}, org_a, existing={"severity": "S1"}
)
assert errors == {}
assert cleaned == {"severity": "S1"}
def test_inactive_field_sent_back_keeps_its_stored_value(self, org_a):
"""The phone sends a record's whole stored map back on save, so a
turned-off field arrives with its own value. Sending the key must not
erase what leaving it out would have kept."""
self._defn(org_a, is_active=False)
cleaned, errors = validate_payload(
"Case", {"severity": "S1"}, org_a, existing={"severity": "S1"}
)
assert errors == {}
assert cleaned == {"severity": "S1"}
@pytest.mark.parametrize("sent", ["S2", None, ""])
def test_inactive_field_cannot_be_changed_or_cleared(self, org_a, sent):
self._defn(org_a, is_active=False)
cleaned, errors = validate_payload(
"Case", {"severity": sent}, org_a, existing={"severity": "S1"}
)
assert errors == {}
assert cleaned == {"severity": "S1"}
def test_removed_option_sent_back_unchanged_is_kept(self, org_a):
"""A record still holding an option an admin has since removed must
stay savable. Its value comes back exactly as stored and is kept."""
self._defn(org_a)
cleaned, errors = validate_payload(
"Case", {"severity": "S9"}, org_a, existing={"severity": "S9"}
)
assert errors == {}
assert cleaned == {"severity": "S9"}
def test_removed_option_cannot_replace_another_value(self, org_a):
self._defn(org_a)
cleaned, errors = validate_payload(
"Case", {"severity": "S9"}, org_a, existing={"severity": "S1"}
)
assert "severity" in errors
def test_unchanged_value_keeps_the_stored_copy(self, org_a):
"""A value equal to the stored one skips validation, and the stored
copy is what gets kept, so nothing a request sends is written
unchecked: `1 == True` keeps the stored boolean, not the integer."""
self._defn(org_a, key="vip", field_type="checkbox", options=None)
cleaned, errors = validate_payload(
"Case", {"vip": 1}, org_a, existing={"vip": True}
)
assert errors == {}
assert cleaned["vip"] is True
def test_whole_number_against_stored_float_is_unchanged(self, org_a):
self._defn(org_a, key="hours", field_type="number", options=None)
cleaned, errors = validate_payload(
"Case", {"hours": 3}, org_a, existing={"hours": 3.0}
)
assert errors == {}
assert cleaned == {"hours": 3.0}
# The phone's lead form clears a text field with an empty string; the
# web and the phone's shared custom-fields form send null.
@pytest.mark.parametrize("sent", ["", None])
def test_empty_value_clears_field(self, org_a, sent):
self._defn(org_a)
cleaned, errors = validate_payload(
"Case", {"severity": sent}, org_a, existing={"severity": "S1"}
)
assert cleaned == {}
assert errors == {}
@pytest.mark.django_db
class TestCustomFieldDefinitionAPI:
URL = "/api/custom-fields/"
def _payload(self, **overrides):
body = {
"target_model": "Case",
"key": "severity",
"label": "Severity",
"field_type": "dropdown",
"options": [
{"value": "S1", "label": "S1"},
{"value": "S2", "label": "S2"},
],
"is_required": False,
"is_filterable": True,
"display_order": 0,
"is_active": True,
}
body.update(overrides)
return body
def test_post_admin_creates(self, admin_client, org_a):
response = admin_client.post(self.URL, self._payload(), format="json")
assert response.status_code == 201, response.content
body = response.json()
assert body["key"] == "severity"
assert (
CustomFieldDefinition.objects.filter(org=org_a, key="severity").count() == 1
)
def test_post_non_admin_forbidden(self, user_client):
response = user_client.post(self.URL, self._payload(), format="json")
assert response.status_code == 403
def test_get_lists_definitions(self, admin_client, org_a):
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = admin_client.get(self.URL + "?target_model=Case")
assert response.status_code == 200
body = response.json()
assert len(body["definitions"]) == 1
assert body["definitions"][0]["key"] == "severity"
def test_get_filters_active_only(self, admin_client, org_a):
CustomFieldDefinition.objects.create(
org=org_a, target_model="Case", key="a", label="A", field_type="text"
)
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="b",
label="B",
field_type="text",
is_active=False,
)
response = admin_client.get(self.URL + "?target_model=Case&active_only=true")
keys = [d["key"] for d in response.json()["definitions"]]
assert keys == ["a"]
def test_get_include_counts_false_omits_the_expensive_counts(
self, admin_client, org_a
):
"""`records_missing_value` is one COUNT over every record of the target
model per definition. Record pages fetch definitions only for labels and
types, so they opt out; the payload must then carry neither the per-row
count nor the totals block."""
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Lead",
key="budget",
label="Budget",
field_type="number",
)
response = admin_client.get(
self.URL + "?target_model=Lead&include_counts=false"
)
assert response.status_code == 200
body = response.json()
assert len(body["definitions"]) == 1
assert body["definitions"][0]["key"] == "budget"
assert "records_missing_value" not in body["definitions"][0]
assert body["totals"] is None
def test_get_counts_included_by_default(self, admin_client, org_a):
"""The opt-out must be opt-in: an unqualified GET, and an explicit
anything-other-than-"false", keep the settings page's numbers."""
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Lead",
key="budget",
label="Budget",
field_type="number",
)
for query in ("?target_model=Lead", "?target_model=Lead&include_counts=true"):
body = admin_client.get(self.URL + query).json()
assert "records_missing_value" in body["definitions"][0], query
assert body["totals"] is not None, query
assert body["totals"]["count"] == 1, query
def test_get_allowed_for_non_admin(self, user_client, org_a):
# Non-admins read definitions so the case detail page can render fields.
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = user_client.get(self.URL + "?target_model=Case")
assert response.status_code == 200
def test_put_updates_label_and_options(self, admin_client, org_a):
defn = CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="dropdown",
options=[{"value": "S1", "label": "S1"}],
)
response = admin_client.put(
f"{self.URL}{defn.id}/",
{
"label": "Severity Level",
"options": [
{"value": "S1", "label": "S1"},
{"value": "S2", "label": "S2"},
],
},
format="json",
)
assert response.status_code == 200, response.content
defn.refresh_from_db()
assert defn.label == "Severity Level"
assert len(defn.options) == 2
def test_put_rejects_key_change(self, admin_client, org_a):
defn = CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = admin_client.put(
f"{self.URL}{defn.id}/", {"key": "severity_new"}, format="json"
)
assert response.status_code == 400
assert "key" in str(response.content)
def test_put_rejects_field_type_change(self, admin_client, org_a):
defn = CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = admin_client.put(
f"{self.URL}{defn.id}/", {"field_type": "number"}, format="json"
)
assert response.status_code == 400
def test_delete_soft_deactivates(self, admin_client, org_a):
defn = CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = admin_client.delete(f"{self.URL}{defn.id}/")
assert response.status_code == 200
defn.refresh_from_db()
assert defn.is_active is False
def test_delete_non_admin_forbidden(self, user_client, org_a):
defn = CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = user_client.delete(f"{self.URL}{defn.id}/")
assert response.status_code == 403
def test_unsupported_target_rejected(self, admin_client):
response = admin_client.post(
self.URL,
self._payload(target_model="Comment"),
format="json",
)
assert response.status_code == 400
assert "target_model" in response.json()["errors"]
def test_duplicate_key_per_target_rejected(self, admin_client, org_a):
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
response = admin_client.post(self.URL, self._payload(), format="json")
assert response.status_code == 400
def test_invalid_key_format_rejected(self, admin_client):
response = admin_client.post(
self.URL, self._payload(key="Severity"), format="json"
)
assert response.status_code == 400
def test_dropdown_without_options_rejected(self, admin_client):
response = admin_client.post(
self.URL,
self._payload(field_type="dropdown", options=None),
format="json",
)
assert response.status_code == 400
def test_cross_org_isolation(self, admin_client, org_b_client, org_a, org_b):
admin_client.post(self.URL, self._payload(), format="json")
org_b_response = org_b_client.get(self.URL + "?target_model=Case")
assert org_b_response.status_code == 200
assert org_b_response.json()["definitions"] == []
@pytest.mark.django_db
class TestCustomFieldAnalytics:
"""records_missing_value + totals, the settings/custom-fields stat cards."""
URL = "/api/custom-fields/"
def _case(self, org, custom_fields):
return Case.objects.create(
org=org,
name="C",
status="New",
priority="Normal",
custom_fields=custom_fields,
)
def _row(self, response, key):
return next(d for d in response.json()["definitions"] if d["key"] == key)
def test_records_missing_value_counts_rows_without_the_key(
self, admin_client, org_a
):
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
is_required=True,
)
self._case(org_a, {"severity": "S1"}) # has a value
self._case(org_a, {}) # missing
self._case(org_a, {"other": "x"}) # missing (different key)
row = self._row(admin_client.get(self.URL), "severity")
assert row["records_missing_value"] == 2
def test_totals_count_active_models_and_gaps(self, admin_client, org_a):
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="S",
field_type="text",
is_required=True,
)
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="region",
label="R",
field_type="text",
is_required=False,
)
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Lead",
key="score",
label="Sc",
field_type="text",
is_required=True,
)
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="legacy",
label="L",
field_type="text",
is_active=False,
)
self._case(org_a, {}) # missing severity + region
self._case(org_a, {"severity": "S1", "region": "eu"}) # has both
Lead.objects.create(org=org_a, custom_fields={}) # missing score
totals = admin_client.get(self.URL).json()["totals"]
assert totals == {
"count": 4, # includes the inactive legacy field
"active": 3,
"models_extended": 2, # Case + Lead have active fields
# severity (1 case missing) + score (1 lead missing); region isn't
# required, legacy is inactive & not required.
"required_with_gaps": 2,
}
def test_records_missing_value_is_org_scoped(self, admin_client, org_a, org_b):
"""The record count must not include another org's rows. Proven on
SQLite where RLS is inert, so only the explicit org filter scopes it."""
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="S",
field_type="text",
is_required=True,
)
self._case(org_a, {}) # one org_a case, missing the value
for _ in range(3):
self._case(org_b, {}) # foreign-org cases must not be counted
row = self._row(admin_client.get(self.URL), "severity")
assert row["records_missing_value"] == 1 # not 4
def test_non_admin_sees_analytics(self, user_client, org_a):
# Reads are open to members (shared config), analytics included.
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="S",
field_type="text",
)
resp = user_client.get(self.URL)
assert resp.status_code == 200
assert "totals" in resp.json()
assert resp.json()["definitions"][0]["records_missing_value"] == 0
TARGET_KEYS = {
"Account": "account",
"Case": "ticket",
"Contact": "contact",
"Estimate": "estimate",
"Invoice": "invoice",
"Lead": "lead",
"Opportunity": "deal",
"RecurringInvoice": "recurring_invoice",
"Task": "task",
}
@pytest.mark.django_db
class TestEveryTarget:
URL = "/api/custom-fields/"
@pytest.mark.parametrize("target", TARGET_KEYS)
def test_an_admin_creates_a_definition_on_every_target(self, target, admin_client):
payload = TestCustomFieldDefinitionAPI()._payload(target_model=target)
response = admin_client.post(self.URL, payload, format="json")
assert response.status_code == 201, response.content
@pytest.mark.parametrize("value", ["Comment", "", "x" * 33])
def test_a_bad_target_gets_todays_exact_error(self, value, admin_client):
payload = TestCustomFieldDefinitionAPI()._payload(target_model=value)
response = admin_client.post(self.URL, payload, format="json")
assert response.status_code == 400
assert response.json()["errors"]["target_model"] == [
f'"{value}" is not a valid choice.'
]
def test_missing_counts_and_totals_cover_every_target(
self, admin_client, org_a, org_b
):
for target, key in TARGET_KEYS.items():
CustomFieldDefinition.objects.create(
org=org_a,
target_model=target,
key="probe",
label="Probe",
field_type="text",
is_required=True,
)
make_record(key, org_a)
filled = make_record(key, org_a)
type(filled).objects.filter(pk=filled.pk).update(
custom_fields={"probe": "set"}
)
with rls_org(org_b):
make_record(key, org_b)
body = admin_client.get(self.URL).json()
assert {
d["target_model"]: d["records_missing_value"] for d in body["definitions"]
} == dict.fromkeys(TARGET_KEYS, 1)
assert body["totals"] == {
"count": 9,
"active": 9,
"models_extended": 9,
"required_with_gaps": 9,
}
def test_the_registry_hooks_run_once_per_list(
self, admin_client, org_a, monkeypatch
):
calls = []
monkeypatch.setattr(
object_registry, "_HOOKS", [lambda org: calls.append(org) or ()]
)
for target in TARGET_KEYS:
CustomFieldDefinition.objects.create(
org=org_a,
target_model=target,
key="probe",
label="Probe",
field_type="text",
)
assert admin_client.get(self.URL).status_code == 200
assert len(calls) == 1
def test_a_stored_target_the_registry_does_not_know_is_not_listed(
self, admin_client, org_a
):
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Ghost",
key="probe",
label="Probe",
field_type="text",
)
response = admin_client.get(self.URL)
assert response.status_code == 200
assert response.json()["definitions"] == []
@pytest.mark.django_db
class TestObjectFieldRowsStayInvisible:
URL = "/api/custom-fields/"
@pytest.fixture
def plant(self, admin_client, org_a):
CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="severity",
label="Severity",
field_type="text",
)
before = admin_client.get(self.URL).json()
row = CustomFieldDefinition.objects.create(
org=org_a,
target_model="obj:example",
key="amount",
label="Amount",
field_type="number",
)
return before, row
def test_the_list_and_totals_do_not_change(self, admin_client, plant):
before, _row = plant
assert admin_client.get(self.URL).json() == before
def test_filtering_by_an_object_target_finds_nothing(self, admin_client, plant):
response = admin_client.get(self.URL + "?target_model=obj:example")
assert response.status_code == 200
assert response.json()["definitions"] == []
@pytest.mark.parametrize("method", ["get", "put", "delete"])
def test_the_detail_route_answers_404(self, method, admin_client, plant):
_before, row = plant
kwargs = {"format": "json"} if method == "put" else {}
if method == "put":
kwargs["data"] = {"label": "Changed", "is_active": False}
response = getattr(admin_client, method)(f"{self.URL}{row.id}/", **kwargs)
assert response.status_code == 404
row.refresh_from_db()
assert row.label == "Amount"
assert row.is_active is True
def test_creating_one_is_refused(self, admin_client):
response = admin_client.post(
self.URL,
{
"target_model": "obj:example",
"key": "amount",
"label": "Amount",
"field_type": "number",
},
format="json",
)
assert response.status_code == 400
assert response.json()["errors"]["target_model"] == [
'"obj:example" is not a valid choice.'
]