wagtailcms/apps/forms/tests.py

"""apps.forms 单元测试：线索提交 API（B2B 官网核心转化路径，测试覆盖率要求较高）。"""
from datetime import timedelta

import pytest
from django.core.cache import cache
from django.utils import timezone
from rest_framework.test import APIClient

from apps.forms.models import (
    FormDefinition,
    FormDefinitionField,
    Lead,
    LeadDeletionRequest,
)

pytestmark = pytest.mark.django_db


@pytest.fixture(autouse=True)
def _clear_throttle_cache():
    """ScopedRateThrottle 依赖默认缓存记录请求次数，避免测试间相互影响（forms scope 限流 5/min）。"""
    cache.clear()
    yield


@pytest.fixture
def contact_form():
    form_def = FormDefinition.objects.create(
        name="联系我们",
        success_message="提交成功，我们会尽快联系您。",
    )
    FormDefinitionField.objects.create(
        form=form_def, label="姓名", field_key="name", field_type="text", required=True
    )
    return form_def


def test_lead_submit_creates_lead_and_returns_success_message(contact_form):
    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/",
        {
            "form_id": contact_form.pk,
            "data": {"name": "张三"},
            "source_url": "https://example.com/contact",
        },
        format="json",
    )

    assert response.status_code == 201
    assert response.data["success"] is True
    assert response.data["message"] == contact_form.success_message

    lead = Lead.objects.get()
    assert lead.form_id == contact_form.pk
    assert lead.data == {"name": "张三"}
    assert lead.source_url == "https://example.com/contact"


def test_lead_submit_returns_404_for_unknown_form():
    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/",
        {"form_id": 9999, "data": {}},
        format="json",
    )

    assert response.status_code == 404
    assert response.data["error"]["code"] == "NOT_FOUND"


def test_lead_submit_requires_form_id():
    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/",
        {"data": {"name": "张三"}},
        format="json",
    )

    assert response.status_code == 400


# --- PIPL 数据删除权（详见设计文档 §2.15） ---


def test_deletion_request_returns_generic_message_and_sends_email(
    contact_form, mailoutbox
):
    Lead.objects.create(
        form=contact_form,
        data={"name": "张三", "email": "zhangsan@example.com"},
    )

    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/deletion-requests/",
        {"contact": "zhangsan@example.com"},
        format="json",
    )

    assert response.status_code == 202
    assert LeadDeletionRequest.objects.filter(contact="zhangsan@example.com").exists()
    assert len(mailoutbox) == 1
    assert mailoutbox[0].to == ["zhangsan@example.com"]


def test_deletion_request_returns_same_generic_message_for_unknown_contact(mailoutbox):
    """未提交过表单的邮箱也返回相同提示，避免探测。"""
    client = APIClient()
    response_known = client.post(
        "/api/v1/custom/leads/deletion-requests/",
        {"contact": "known@example.com"},
        format="json",
    )
    response_unknown = client.post(
        "/api/v1/custom/leads/deletion-requests/",
        {"contact": "unknown@example.com"},
        format="json",
    )

    assert response_known.status_code == response_unknown.status_code == 202
    assert response_known.data["message"] == response_unknown.data["message"]


def test_deletion_confirm_deletes_matching_leads_only(contact_form):
    matching_lead = Lead.objects.create(
        form=contact_form,
        data={"name": "张三", "email": "zhangsan@example.com"},
    )
    other_lead = Lead.objects.create(
        form=contact_form,
        data={"name": "李四", "email": "lisi@example.com"},
    )
    deletion_request = LeadDeletionRequest.objects.create(contact="zhangsan@example.com")

    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/deletion-requests/confirm/",
        {"token": str(deletion_request.token)},
        format="json",
    )

    assert response.status_code == 200
    assert response.data["success"] is True
    assert response.data["deleted_count"] == 1
    assert not Lead.objects.filter(pk=matching_lead.pk).exists()
    assert Lead.objects.filter(pk=other_lead.pk).exists()

    deletion_request.refresh_from_db()
    assert deletion_request.status == LeadDeletionRequest.STATUS_COMPLETED
    assert deletion_request.deleted_count == 1
    assert deletion_request.completed_at is not None


def test_deletion_confirm_invalid_token_returns_404():
    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/deletion-requests/confirm/",
        {"token": "00000000-0000-0000-0000-000000000000"},
        format="json",
    )

    assert response.status_code == 404
    assert response.data["error"]["code"] == "INVALID_TOKEN"


def test_deletion_confirm_expired_token_returns_400():
    deletion_request = LeadDeletionRequest.objects.create(contact="zhangsan@example.com")
    LeadDeletionRequest.objects.filter(pk=deletion_request.pk).update(
        requested_at=timezone.now() - timedelta(hours=25)
    )

    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/deletion-requests/confirm/",
        {"token": str(deletion_request.token)},
        format="json",
    )

    assert response.status_code == 400
    assert response.data["error"]["code"] == "TOKEN_EXPIRED"

    deletion_request.refresh_from_db()
    assert deletion_request.status == LeadDeletionRequest.STATUS_EXPIRED


def test_deletion_confirm_already_processed_returns_400():
    deletion_request = LeadDeletionRequest.objects.create(
        contact="zhangsan@example.com",
        status=LeadDeletionRequest.STATUS_COMPLETED,
    )

    client = APIClient()
    response = client.post(
        "/api/v1/custom/leads/deletion-requests/confirm/",
        {"token": str(deletion_request.token)},
        format="json",
    )

    assert response.status_code == 400
    assert response.data["error"]["code"] == "ALREADY_PROCESSED"