224 lines
8.1 KiB
Python
224 lines
8.1 KiB
Python
"""Web forms follow the module they write into.
|
|
|
|
While a form's target module is off, its public endpoints answer the
|
|
not-found an unknown form gets, CORS headers included, publishing it is
|
|
refused, and choosing that target is refused. Forms for a module that is on
|
|
keep working.
|
|
"""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from django.core.cache import cache
|
|
from django.db import connection
|
|
from django.test import RequestFactory
|
|
from django.test.utils import CaptureQueriesContext
|
|
from rest_framework.test import APIClient
|
|
|
|
from cases.models import Case
|
|
from common.modules import module_off_message
|
|
from common.testing import set_modules_off
|
|
from leads.models import Lead
|
|
from webforms.cors import allow_webform_origin
|
|
from webforms.models import WebForm, WebFormDailyStat, WebFormField, WebFormSubmission
|
|
|
|
LIST_URL = "/api/webforms/"
|
|
TICKETS_OFF = "Tickets is switched off for this organization."
|
|
ORIGIN = "https://customer.example"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_throttle_cache():
|
|
cache.clear()
|
|
yield
|
|
cache.clear()
|
|
|
|
|
|
def _form(org, target, *, published=True):
|
|
form = WebForm.objects.create(
|
|
name=f"{target} form", org=org, target=target, is_published=published
|
|
)
|
|
source = (
|
|
WebFormField.SOURCE_LEAD if target == "lead" else WebFormField.SOURCE_TICKET
|
|
)
|
|
WebFormField.objects.create(
|
|
form=form,
|
|
org=org,
|
|
order=0,
|
|
source=source,
|
|
label="Email",
|
|
is_required=True,
|
|
**{f"{target}_field": "email"},
|
|
)
|
|
return form
|
|
|
|
|
|
def _public(org_id, form_id):
|
|
return f"/api/public/forms/{org_id}/{form_id}/"
|
|
|
|
|
|
def _answers(org_id, form_id):
|
|
"""What the three public endpoints say for this form, as (status, body)."""
|
|
client = APIClient()
|
|
base = _public(org_id, form_id)
|
|
responses = (
|
|
client.post(f"{base}submit/", {"email": "pat@example.com"}, format="json"),
|
|
client.get(f"{base}embed/"),
|
|
client.get(f"{base}embed.js"),
|
|
)
|
|
return [(r.status_code, r.content) for r in responses]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"closed, still_open, module",
|
|
[("ticket", "lead", "tickets"), ("lead", "ticket", "leads")],
|
|
)
|
|
def test_a_form_closes_with_its_target_module(org_a, closed, still_open, module):
|
|
off_form = _form(org_a, closed)
|
|
on_form = _form(org_a, still_open)
|
|
unknown = _answers(org_a.id, uuid.uuid4())
|
|
assert unknown[0] == (404, b'{"detail":"Form not found."}')
|
|
assert [status for status, _ in unknown] == [404, 404, 404]
|
|
|
|
set_modules_off(org_a, module)
|
|
assert _answers(org_a.id, off_form.id) == unknown
|
|
assert not WebFormSubmission.objects.filter(form=off_form).exists()
|
|
assert not WebFormDailyStat.objects.filter(form=off_form).exists()
|
|
created = Case if closed == "ticket" else Lead
|
|
assert not created.objects.filter(org=org_a).exists()
|
|
|
|
assert [status for status, _ in _answers(org_a.id, on_form.id)] == [200, 200, 200]
|
|
kept = Lead if still_open == "lead" else Case
|
|
assert kept.objects.filter(org=org_a).count() == 1
|
|
|
|
set_modules_off(org_a)
|
|
assert [status for status, _ in _answers(org_a.id, off_form.id)] == [200, 200, 200]
|
|
assert created.objects.filter(org=org_a).count() == 1
|
|
|
|
|
|
def _cors(org_id, form_id):
|
|
"""(status, CORS headers) for a preflight and a POST from a listed origin."""
|
|
client = APIClient()
|
|
url = f"{_public(org_id, form_id)}submit/"
|
|
responses = (
|
|
client.options(
|
|
url, HTTP_ORIGIN=ORIGIN, HTTP_ACCESS_CONTROL_REQUEST_METHOD="POST"
|
|
),
|
|
client.post(
|
|
url, {"email": "pat@example.com"}, format="json", HTTP_ORIGIN=ORIGIN
|
|
),
|
|
)
|
|
return [
|
|
(
|
|
r.status_code,
|
|
{
|
|
name.lower(): value
|
|
for name, value in r.headers.items()
|
|
if name.lower().startswith("access-control-") or name.lower() == "vary"
|
|
},
|
|
)
|
|
for r in responses
|
|
]
|
|
|
|
|
|
def test_cors_closes_with_the_target_module(org_a):
|
|
form = _form(org_a, "ticket")
|
|
form.allowed_origins = [ORIGIN]
|
|
form.save(update_fields=["allowed_origins"])
|
|
unknown = _cors(org_a.id, uuid.uuid4())
|
|
assert [s for s, _ in unknown] == [200, 404]
|
|
assert not any("access-control-allow-origin" in h for _, h in unknown)
|
|
|
|
set_modules_off(org_a, "tickets")
|
|
assert _cors(org_a.id, form.id) == unknown
|
|
assert not Case.objects.filter(org=org_a).exists()
|
|
|
|
set_modules_off(org_a)
|
|
opened = _cors(org_a.id, form.id)
|
|
assert [s for s, _ in opened] == [200, 200]
|
|
assert [h.get("access-control-allow-origin") for _, h in opened] == [ORIGIN] * 2
|
|
|
|
|
|
def test_the_cors_check_reads_the_form_in_one_query(org_a):
|
|
"""Pin: it runs on every submit request, and read one row before too."""
|
|
form = _form(org_a, "ticket")
|
|
form.allowed_origins = [ORIGIN]
|
|
form.save(update_fields=["allowed_origins"])
|
|
request = RequestFactory().post(
|
|
f"{_public(org_a.id, form.id)}submit/", HTTP_ORIGIN=ORIGIN
|
|
)
|
|
with CaptureQueriesContext(connection) as captured:
|
|
assert allow_webform_origin(None, request) is True
|
|
reads = [q for q in captured.captured_queries if "set_config" not in q["sql"]]
|
|
assert len(reads) == 1
|
|
|
|
|
|
def test_publishing_a_form_for_an_off_module_is_refused(
|
|
admin_client, user_client, org_a
|
|
):
|
|
form = _form(org_a, "ticket", published=False)
|
|
url = f"{LIST_URL}{form.id}/publish/"
|
|
set_modules_off(org_a, "tickets")
|
|
assert user_client.post(url).status_code == 403
|
|
|
|
response = admin_client.post(url)
|
|
assert (response.status_code, response.json()) == (
|
|
400,
|
|
{
|
|
"error": True,
|
|
"errors": f"{TICKETS_OFF} Switch it on before publishing this form.",
|
|
},
|
|
)
|
|
form.refresh_from_db()
|
|
assert form.is_published is False
|
|
published = _form(org_a, "ticket")
|
|
response = admin_client.post(f"{LIST_URL}{published.id}/publish/")
|
|
assert response.json()["errors"] == "This form is already published."
|
|
|
|
set_modules_off(org_a)
|
|
assert admin_client.post(url).status_code == 200
|
|
form.refresh_from_db()
|
|
assert form.is_published is True
|
|
|
|
|
|
def test_choosing_an_off_target_is_refused(admin_client, user_client, org_a):
|
|
ticket_rows = [{"source": "ticket", "ticket_field": "email", "label": "Email"}]
|
|
new_ticket_form = {"name": "Support", "target": "ticket", "fields": ticket_rows}
|
|
lead_form = _form(org_a, "lead", published=False)
|
|
change = {"target": "ticket", "fields": ticket_rows}
|
|
set_modules_off(org_a, "tickets")
|
|
|
|
refused = user_client.post(LIST_URL, new_ticket_form, format="json")
|
|
assert refused.status_code == 403
|
|
response = admin_client.post(LIST_URL, new_ticket_form, format="json")
|
|
assert (response.status_code, response.json()) == (400, {"target": [TICKETS_OFF]})
|
|
response = admin_client.put(f"{LIST_URL}{lead_form.id}/", change, format="json")
|
|
assert (response.status_code, response.json()) == (400, {"target": [TICKETS_OFF]})
|
|
assert list(WebForm.objects.filter(org=org_a)) == [lead_form]
|
|
lead_form.refresh_from_db()
|
|
assert lead_form.target == "lead"
|
|
|
|
set_modules_off(org_a, "leads")
|
|
response = admin_client.post(LIST_URL, {"name": "Contact us"}, format="json")
|
|
assert response.status_code == 400
|
|
assert response.json() == {"target": [module_off_message(org_a, "leads")]}
|
|
|
|
set_modules_off(org_a)
|
|
created = admin_client.post(LIST_URL, new_ticket_form, format="json")
|
|
assert created.status_code == 201
|
|
response = admin_client.put(f"{LIST_URL}{lead_form.id}/", change, format="json")
|
|
assert response.status_code == 200, response.content
|
|
lead_form.refresh_from_db()
|
|
assert lead_form.target == "ticket"
|
|
|
|
|
|
def test_editing_an_existing_forms_name_still_works_while_off(admin_client, org_a):
|
|
"""Pin: a write that keeps the target was never refused."""
|
|
form = _form(org_a, "ticket")
|
|
set_modules_off(org_a, "tickets")
|
|
url = f"{LIST_URL}{form.id}/"
|
|
assert admin_client.put(url, {"name": "Renamed"}, format="json").status_code == 200
|
|
resent = {"name": "Renamed again", "target": "ticket"}
|
|
assert admin_client.put(url, resent, format="json").status_code == 200
|
|
form.refresh_from_db()
|
|
assert (form.name, form.target) == ("Renamed again", "ticket")
|