bottleneckcrm/backend/webforms/tests/test_ticket_forms.py
Ashwin e26ab380fc Add unit tests for ticket board provider, status settable, web forms, CSV import, deal aging badge, and tickets selection
- Implement comprehensive tests for the ticket board provider to validate kanban and pipeline modes, including ticket movement and error handling.
- Create tests for ticket status settable to ensure proper status selection logic.
- Enhance web form model tests to cover ticket-specific fields and submission behavior.
- Introduce CSV import sheet tests to verify file handling and error reporting.
- Add deal aging badge tests to confirm correct display of aging statuses.
- Implement tickets selection tests to ensure UI responsiveness and functionality at various screen sizes.
2026-09-26 16:53:22 +05:30

745 lines
26 KiB
Python

"""Web forms that open a ticket instead of creating a lead.
A ticket form is the same anonymous endpoint as a lead form, with the same
controls (captcha, honeypot, throttles, allowed origins). What changes is the
record: a Case created the way a new inbound email creates one, a Contact
found or created by email within the form's org, and none of the lead-only
behaviour (source, merge, lead wording).
"""
import pytest
from django.core import mail
from django.core.cache import cache
from cases.models import Case, RoutingRule
from common.models import CustomFieldDefinition, Profile, Tags, User
from contacts.models import Contact
from leads.models import Lead
from webforms.dynamic_serializer import HONEYPOT_FIELD
from webforms.models import WebForm, WebFormField, WebFormSubmission
from webforms.service import submit_form
LIST_URL = "/api/webforms/"
def submit_url(org, form):
return f"/api/public/forms/{org.id}/{form.id}/submit/"
def detail_url(form):
return f"/api/webforms/{form.id}/"
@pytest.fixture(autouse=True)
def _clear_throttle_cache():
cache.clear()
yield
cache.clear()
@pytest.fixture
def superuser_client(user_client, regular_user):
"""The `user_client` caller, still a plain USER profile, made superuser."""
regular_user.is_superuser = True
regular_user.save(update_fields=["is_superuser"])
return user_client
def _ticket_form(org, **extra):
form = WebForm.objects.create(
name="Support request",
org=org,
target=WebForm.TARGET_TICKET,
is_published=True,
ticket_priority="High",
ticket_type="Incident",
**extra,
)
for order, key in enumerate(
["email", "first_name", "last_name", "phone", "name", "description"]
):
WebFormField.objects.create(
form=form,
org=org,
order=order,
source=WebFormField.SOURCE_TICKET,
ticket_field=key,
label=key,
is_required=key == "email",
)
return form
@pytest.fixture
def form(org_a):
return _ticket_form(org_a)
def _post(client, target_org, web_form, **data):
body = {"email": "pat@example.com", "name": "Printer on fire"}
body.update(data)
return client.post(submit_url(target_org, web_form), body, format="json")
# --------------------------------------------------------------------------
# Submission creates a ticket
# --------------------------------------------------------------------------
@pytest.mark.django_db
class TestTicketSubmission:
def test_an_anonymous_post_opens_a_ticket_in_the_forms_org(
self, unauthenticated_client, org_a, form
):
response = _post(
unauthenticated_client, org_a, form, description="It is smoking."
)
assert response.status_code == 200, response.data
case = Case.objects.get(org=org_a)
assert case.name == "Printer on fire"
assert case.description == "It is smoking."
assert case.status == "New"
assert case.priority == "High"
assert case.case_type == "Incident"
submission = WebFormSubmission.objects.get(form=form)
assert submission.status == WebFormSubmission.ACCEPTED
assert submission.case_id == case.id
assert submission.lead_id is None
def test_no_lead_is_written_by_a_ticket_form(
self, unauthenticated_client, org_a, form
):
_post(unauthenticated_client, org_a, form)
assert not Lead.objects.filter(org=org_a).exists()
def test_sla_targets_are_stamped_like_any_new_case(
self, unauthenticated_client, org_a, form
):
_post(unauthenticated_client, org_a, form)
case = Case.objects.get(org=org_a)
assert case.sla_first_response_hours is not None
assert case.sla_resolution_hours is not None
def test_a_subject_longer_than_case_name_is_truncated_not_refused(
self, unauthenticated_client, org_a, form
):
response = _post(unauthenticated_client, org_a, form, name="x" * 200)
assert response.status_code == 200, response.data
assert Case.objects.get(org=org_a).name == "x" * 64
def test_a_form_without_a_subject_names_the_ticket_after_the_form(
self, unauthenticated_client, org_a, form
):
_post(unauthenticated_client, org_a, form, name="")
assert Case.objects.get(org=org_a).name == "Support request"
def test_a_blank_ticket_type_creates_a_case_with_no_type(
self, unauthenticated_client, org_a, form
):
form.ticket_type = ""
form.save(update_fields=["ticket_type"])
_post(unauthenticated_client, org_a, form)
assert Case.objects.get(org=org_a).case_type is None
def test_the_forms_tags_are_applied(self, unauthenticated_client, org_a, form):
tag = Tags.objects.create(org=org_a, name="web")
form.tags.add(tag)
_post(unauthenticated_client, org_a, form)
assert list(Case.objects.get(org=org_a).tags.all()) == [tag]
def test_a_visitor_cannot_set_priority_status_or_org(
self, unauthenticated_client, org_a, org_b, form
):
"""Fields not on the form are dropped, so a stranger cannot raise
their own ticket to Urgent or aim it at another tenant."""
_post(
unauthenticated_client,
org_a,
form,
priority="Urgent",
status="Closed",
org=str(org_b.id),
)
case = Case.objects.get(org=org_a)
assert case.priority == "High"
assert case.status == "New"
assert not Case.objects.filter(org=org_b).exists()
# --------------------------------------------------------------------------
# Contact matching
# --------------------------------------------------------------------------
@pytest.mark.django_db
class TestTicketContact:
def test_an_existing_contact_is_matched_case_insensitively(
self, unauthenticated_client, org_a, form
):
existing = Contact.objects.create(
org=org_a, first_name="Pat", last_name="Old", email="Pat@Example.com"
)
_post(
unauthenticated_client,
org_a,
form,
email="pat@example.COM",
first_name="Mallory",
phone="+1 555 0100",
)
case = Case.objects.get(org=org_a)
assert list(case.contacts.all()) == [existing]
assert Contact.objects.filter(org=org_a).count() == 1
# A stranger who knows the address cannot rewrite the contact.
existing.refresh_from_db()
assert existing.first_name == "Pat"
assert existing.phone in (None, "")
def test_an_unknown_address_creates_a_contact_from_the_submission(
self, unauthenticated_client, org_a, form
):
_post(
unauthenticated_client,
org_a,
form,
first_name="Pat",
last_name="Lee",
phone="+1 555 0100",
)
contact = Contact.objects.get(org=org_a)
assert contact.email == "pat@example.com"
assert (contact.first_name, contact.last_name) == ("Pat", "Lee")
assert contact.phone == "+1 555 0100"
assert contact.auto_created is True
assert list(Case.objects.get(org=org_a).contacts.all()) == [contact]
def test_a_contact_with_the_address_in_another_org_is_not_used(
self, unauthenticated_client, org_a, org_b, form
):
from common.testing import rls_org
with rls_org(org_b):
foreign = Contact.objects.create(
org=org_b, first_name="Pat", last_name="B", email="pat@example.com"
)
_post(unauthenticated_client, org_a, form)
contact = Case.objects.get(org=org_a).contacts.get()
assert contact.org_id == org_a.id
assert contact.id != foreign.id
def test_a_missing_first_name_falls_back_to_the_address(
self, unauthenticated_client, org_a, form
):
_post(unauthenticated_client, org_a, form, email="sam@example.com")
assert Contact.objects.get(org=org_a).first_name == "sam"
# --------------------------------------------------------------------------
# Assignment and routing
# --------------------------------------------------------------------------
def _member(org, email, *, active=True):
user = User.objects.create_user(email=email, password="x")
return Profile.objects.create(user=user, org=org, role="USER", is_active=active)
@pytest.mark.django_db
class TestTicketAssignment:
def test_an_active_assignee_in_the_org_is_applied(
self, unauthenticated_client, org_a, form
):
agent = _member(org_a, "agent@test.com")
form.assign_to = agent
form.save(update_fields=["assign_to"])
_post(unauthenticated_client, org_a, form)
assert list(Case.objects.get(org=org_a).assigned_to.all()) == [agent]
def test_an_inactive_assignee_is_never_applied(
self, unauthenticated_client, org_a, form
):
gone = _member(org_a, "gone@test.com", active=False)
form.assign_to = gone
form.save(update_fields=["assign_to"])
_post(unauthenticated_client, org_a, form)
assert not Case.objects.get(org=org_a).assigned_to.exists()
def test_an_assignee_from_another_org_is_never_applied(
self, unauthenticated_client, org_a, org_b, form
):
stranger = _member(org_b, "stranger@test.com")
form.assign_to = stranger
form.save(update_fields=["assign_to"])
_post(unauthenticated_client, org_a, form)
assert not Case.objects.get(org=org_a).assigned_to.exists()
def test_routing_matches_the_senders_domain_as_for_inbound_email(
self, unauthenticated_client, org_a, form
):
agent = _member(org_a, "router@test.com")
rule = RoutingRule.objects.create(
org=org_a,
name="Example customers",
conditions=[
{"field": "from_email_domain", "op": "eq", "value": "example.com"}
],
strategy="direct",
)
rule.target_assignees.set([agent])
_post(unauthenticated_client, org_a, form)
assert list(Case.objects.get(org=org_a).assigned_to.all()) == [agent]
def test_a_rule_on_the_forms_tag_routes_the_ticket_once(
self, unauthenticated_client, org_a, form
):
"""Routing used to run on the first save, before the tags were on."""
from common.models import Activity
form.tags.add(Tags.objects.create(org=org_a, name="billing"))
agent = _member(org_a, "billing@test.com")
rule = RoutingRule.objects.create(
org=org_a,
name="Billing",
conditions=[{"field": "tags", "op": "eq", "value": "billing"}],
strategy="direct",
)
rule.target_assignees.set([agent])
_post(unauthenticated_client, org_a, form)
case = Case.objects.get(org=org_a)
assert list(case.assigned_to.all()) == [agent]
assert (
Activity.objects.filter(
entity_type="Case", entity_id=case.pk, action="ROUTED"
).count()
== 1
)
# --------------------------------------------------------------------------
# Controls unchanged on ticket forms
# --------------------------------------------------------------------------
@pytest.mark.django_db
class TestTicketFormControls:
def test_the_honeypot_writes_no_ticket(self, unauthenticated_client, org_a, form):
response = _post(
unauthenticated_client, org_a, form, **{HONEYPOT_FIELD: "http://spam"}
)
assert response.status_code == 200
assert not Case.objects.filter(org=org_a).exists()
assert (
WebFormSubmission.objects.get(form=form).status
== WebFormSubmission.REJECTED_SPAM
)
def test_a_failed_captcha_is_400_and_writes_no_ticket(
self, unauthenticated_client, org_a, form, monkeypatch
):
monkeypatch.setattr(
"webforms.public_views.captcha.verify", lambda *a, **k: False
)
response = _post(unauthenticated_client, org_a, form)
assert response.status_code == 400
assert not Case.objects.filter(org=org_a).exists()
assert not Contact.objects.filter(org=org_a).exists()
def test_a_passing_captcha_opens_the_ticket(
self, unauthenticated_client, org_a, form, monkeypatch
):
monkeypatch.setattr(
"webforms.public_views.captcha.verify", lambda *a, **k: True
)
assert _post(unauthenticated_client, org_a, form).status_code == 200
assert Case.objects.filter(org=org_a).count() == 1
def test_an_origin_outside_the_allowed_list_is_403(
self, unauthenticated_client, org_a, form
):
form.allowed_origins = ["https://good.example"]
form.save(update_fields=["allowed_origins"])
response = unauthenticated_client.post(
submit_url(org_a, form),
{"email": "pat@example.com"},
format="json",
HTTP_ORIGIN="https://evil.example",
)
assert response.status_code == 403
assert not Case.objects.filter(org=org_a).exists()
def test_a_lead_column_posted_to_a_ticket_form_is_dropped(
self, unauthenticated_client, org_a, form
):
_post(unauthenticated_client, org_a, form, company_name="Acme", website="x")
submission = WebFormSubmission.objects.get(form=form)
assert "website" not in submission.payload
# --------------------------------------------------------------------------
# Custom fields
# --------------------------------------------------------------------------
@pytest.mark.django_db
class TestTicketCustomFields:
def test_a_case_custom_field_lands_on_the_ticket(
self, unauthenticated_client, org_a, form
):
definition = CustomFieldDefinition.objects.create(
org=org_a,
target_model="Case",
key="order_number",
label="Order",
field_type="text",
)
WebFormField.objects.create(
form=form,
org=org_a,
order=9,
source=WebFormField.SOURCE_CUSTOM,
custom_field=definition,
label="Order",
)
_post(unauthenticated_client, org_a, form, order_number="A-1")
assert Case.objects.get(org=org_a).custom_fields == {"order_number": "A-1"}
# --------------------------------------------------------------------------
# Notification wording
# --------------------------------------------------------------------------
@pytest.mark.django_db
class TestTicketNotification:
def test_the_email_says_ticket_and_links_the_ticket(
self, unauthenticated_client, org_a, form, admin_profile, settings
):
settings.CELERY_TASK_ALWAYS_EAGER = True
form.notify_profiles.add(admin_profile)
mail.outbox.clear()
from webforms.tasks import send_webform_submission_email
submission = submit_form(form, {"email": "pat@example.com", "name": "Help"})
send_webform_submission_email(str(submission.id), str(org_a.id))
assert len(mail.outbox) == 1
message = mail.outbox[0]
assert message.subject == "New ticket: Support request"
body = message.alternatives[0][0] if message.alternatives else message.body
assert "Open the ticket" in body
assert f"/tickets/{submission.case_id}" in body
assert "lead" not in body.lower()
# --------------------------------------------------------------------------
# The management API
# --------------------------------------------------------------------------
def _ticket_row(key, **extra):
return {"source": "ticket", "ticket_field": key, "label": key, **extra}
@pytest.mark.django_db
class TestTicketFormWrites:
def test_an_admin_creates_a_ticket_form(self, admin_client, org_a):
response = admin_client.post(
LIST_URL,
{
"name": "Support",
"target": "ticket",
"ticket_priority": "Urgent",
"ticket_type": "Problem",
"fields": [_ticket_row("email"), _ticket_row("name")],
},
format="json",
)
assert response.status_code == 201, response.data
form = WebForm.objects.get(id=response.data["id"])
assert form.target == "ticket"
assert (form.ticket_priority, form.ticket_type) == ("Urgent", "Problem")
assert [f.ticket_field for f in form.fields.order_by("order")] == [
"email",
"name",
]
def test_a_superuser_without_an_admin_role_can_create(
self, superuser_client, org_a
):
response = superuser_client.post(
LIST_URL, {"name": "Support", "target": "ticket"}, format="json"
)
assert response.status_code == 201, response.data
def test_a_member_cannot_create_a_ticket_form(self, user_client, org_a):
response = user_client.post(
LIST_URL, {"name": "Support", "target": "ticket"}, format="json"
)
assert response.status_code == 403
assert not WebForm.objects.filter(org=org_a).exists()
def test_a_member_cannot_update_a_ticket_form(self, user_client, form):
response = user_client.put(
detail_url(form), {"ticket_priority": "Low"}, format="json"
)
assert response.status_code == 403
form.refresh_from_db()
assert form.ticket_priority == "High"
def test_an_admin_updates_the_ticket_defaults(self, admin_client, form):
response = admin_client.put(
detail_url(form),
{"ticket_priority": "Low", "ticket_type": ""},
format="json",
)
assert response.status_code == 200, response.data
form.refresh_from_db()
assert (form.ticket_priority, form.ticket_type) == ("Low", "")
@pytest.mark.parametrize(
"body",
[
{"ticket_priority": "Critical"},
{"ticket_type": "Complaint"},
{"target": "opportunity"},
],
)
def test_values_outside_the_case_choices_are_400(self, admin_client, form, body):
assert (
admin_client.put(detail_url(form), body, format="json").status_code == 400
)
def test_a_lead_field_on_a_ticket_form_is_400(self, admin_client, form):
response = admin_client.put(
detail_url(form),
{"fields": [{"source": "lead", "lead_field": "email", "label": "E"}]},
format="json",
)
assert response.status_code == 400
assert form.fields.filter(source="ticket").count() == 6
def test_a_ticket_field_on_a_lead_form_is_400(self, admin_client, org_a):
lead_form = WebForm.objects.create(name="Contact", org=org_a)
response = admin_client.put(
detail_url(lead_form),
{"fields": [_ticket_row("email")]},
format="json",
)
assert response.status_code == 400
assert not lead_form.fields.exists()
def test_a_ticket_field_outside_the_whitelist_is_400(self, admin_client, form):
response = admin_client.put(
detail_url(form),
{"fields": [_ticket_row("priority")]},
format="json",
)
assert response.status_code == 400
def test_a_row_whose_source_does_not_match_its_field_is_400(
self, admin_client, form
):
response = admin_client.put(
detail_url(form),
{"fields": [{"source": "custom", "ticket_field": "email", "label": "E"}]},
format="json",
)
assert response.status_code == 400
def test_a_row_with_no_label_is_400_not_500(self, admin_client, form):
response = admin_client.put(
detail_url(form),
{"fields": [{"source": "ticket", "ticket_field": "email"}]},
format="json",
)
assert response.status_code == 400
def test_a_case_custom_field_is_accepted_on_a_ticket_form(
self, admin_client, org_a, form
):
definition = CustomFieldDefinition.objects.create(
org=org_a, target_model="Case", key="sev", label="S", field_type="text"
)
response = admin_client.put(
detail_url(form),
{
"fields": [
_ticket_row("email"),
{
"source": "custom",
"custom_field": str(definition.id),
"label": "S",
},
]
},
format="json",
)
assert response.status_code == 200, response.data
def test_a_lead_custom_field_is_refused_on_a_ticket_form(
self, admin_client, org_a, form
):
definition = CustomFieldDefinition.objects.create(
org=org_a, target_model="Lead", key="budget", label="B", field_type="text"
)
response = admin_client.put(
detail_url(form),
{
"fields": [
{
"source": "custom",
"custom_field": str(definition.id),
"label": "B",
}
]
},
format="json",
)
assert response.status_code == 400
def test_a_case_custom_field_from_another_org_is_refused(
self, admin_client, org_b, form
):
from common.testing import rls_org
with rls_org(org_b):
foreign = CustomFieldDefinition.objects.create(
org=org_b, target_model="Case", key="sev", label="S", field_type="text"
)
response = admin_client.put(
detail_url(form),
{
"fields": [
{"source": "custom", "custom_field": str(foreign.id), "label": "S"}
]
},
format="json",
)
assert response.status_code == 400
def test_the_detail_and_list_carry_the_target(self, user_client, form):
assert user_client.get(detail_url(form)).data["target"] == "ticket"
assert user_client.get(LIST_URL).data["results"][0]["target"] == "ticket"
@pytest.mark.django_db
class TestTargetChange:
def test_the_target_changes_with_its_field_list_before_any_submission(
self, admin_client, form
):
response = admin_client.put(
detail_url(form),
{
"target": "lead",
"fields": [{"source": "lead", "lead_field": "email", "label": "E"}],
},
format="json",
)
assert response.status_code == 200, response.data
form.refresh_from_db()
assert form.target == "lead"
def test_the_target_cannot_change_without_a_new_field_list(
self, admin_client, form
):
response = admin_client.put(detail_url(form), {"target": "lead"}, format="json")
assert response.status_code == 400
assert "fields" in response.data
form.refresh_from_db()
assert form.target == "ticket"
def test_the_target_cannot_change_once_the_form_has_submissions(
self, admin_client, form
):
submit_form(form, {"email": "pat@example.com"})
response = admin_client.put(
detail_url(form),
{
"target": "lead",
"fields": [{"source": "lead", "lead_field": "email", "label": "E"}],
},
format="json",
)
assert response.status_code == 400
assert "submissions" in str(response.data["target"])
form.refresh_from_db()
assert form.target == "ticket"
def test_a_form_backing_a_legacy_api_key_stays_a_lead_form(
self, admin_client, org_a
):
from common.models import APISettings
setting = APISettings.objects.create(
org=org_a, title="Site", website="https://example.com"
)
legacy = WebForm.objects.create(
name="Legacy", org=org_a, legacy_api_setting=setting
)
response = admin_client.put(
detail_url(legacy),
{"target": "ticket", "fields": [_ticket_row("email")]},
format="json",
)
assert response.status_code == 400
legacy.refresh_from_db()
assert legacy.target == "lead"
def test_resending_the_same_target_after_submissions_is_fine(
self, admin_client, form
):
submit_form(form, {"email": "pat@example.com"})
response = admin_client.put(
detail_url(form), {"target": "ticket", "name": "Renamed"}, format="json"
)
assert response.status_code == 200, response.data
@pytest.mark.django_db
class TestTicketPublishAndSubmissions:
def test_a_ticket_form_without_an_email_field_cannot_publish(
self, admin_client, org_a
):
draft = WebForm.objects.create(name="S", org=org_a, target="ticket")
WebFormField.objects.create(
form=draft,
org=org_a,
source=WebFormField.SOURCE_TICKET,
ticket_field="name",
label="Subject",
)
response = admin_client.post(f"{detail_url(draft)}publish/")
assert response.status_code == 400
assert "contact" in response.data["errors"]
def test_a_ticket_form_with_an_email_field_publishes(self, admin_client, form):
form.is_published = False
form.save(update_fields=["is_published"])
response = admin_client.post(f"{detail_url(form)}publish/")
assert response.status_code == 200, response.data
def test_the_submission_list_links_the_ticket(self, user_client, form):
submission = submit_form(form, {"email": "pat@example.com", "name": "Help"})
response = user_client.get(f"{detail_url(form)}submissions/")
row = response.data["results"][0]
assert row["case"] == submission.case_id
assert row["case_name"] == "Help"
assert row["lead"] is None
@pytest.mark.django_db
class TestSubmissionConstraint:
def test_a_submission_cannot_link_both_a_lead_and_a_case(self, org_a, form):
from django.db import IntegrityError, transaction
lead = Lead.objects.create(org=org_a, first_name="P", email="p@example.com")
case = Case.objects.create(org=org_a, name="C", status="New", priority="Low")
with pytest.raises(IntegrityError), transaction.atomic():
WebFormSubmission.objects.create(
org=org_a,
form=form,
lead=lead,
case=case,
status=WebFormSubmission.ACCEPTED,
)