main-site/apps/comments/tests/test_v2.py

"""Tests for Comments v2: HTMX, Turnstile, reactions, polling, CSP."""
from __future__ import annotations

from datetime import timedelta
from unittest.mock import patch

import pytest
from django.core.cache import cache
from django.core.management import call_command
from django.test import override_settings
from django.utils import timezone

from apps.blog.models import ArticleIndexPage, ArticlePage
from apps.blog.tests.factories import AuthorFactory
from apps.comments.models import Comment, CommentReaction

# ── Fixtures ──────────────────────────────────────────────────────────────────


@pytest.fixture
def _article(home_page):
    """Create a published article with comments enabled."""
    index = ArticleIndexPage(title="Articles", slug="articles")
    home_page.add_child(instance=index)
    author = AuthorFactory()
    article = ArticlePage(
        title="Test Article",
        slug="test-article",
        author=author,
        summary="summary",
        body=[("rich_text", "<p>body</p>")],
    )
    index.add_child(instance=article)
    article.save_revision().publish()
    return article


@pytest.fixture
def approved_comment(_article):
    return Comment.objects.create(
        article=_article,
        author_name="Alice",
        author_email="alice@example.com",
        body="Great article!",
        is_approved=True,
    )


def _post_comment(client, article, extra=None, htmx=False):
    cache.clear()
    payload = {
        "article_id": article.id,
        "author_name": "Test",
        "author_email": "test@example.com",
        "body": "Hello world",
        "honeypot": "",
    }
    if extra:
        payload.update(extra)
    headers = {}
    if htmx:
        headers["HTTP_HX_REQUEST"] = "true"
    return client.post("/comments/post/", payload, **headers)


# ── HTMX Response Contracts ──────────────────────────────────────────────────


@pytest.mark.django_db
def test_htmx_post_returns_form_with_moderation_on_success(client, _article):
    """HTMX POST with Turnstile disabled returns fresh form + moderation message."""
    resp = _post_comment(client, _article, htmx=True)
    assert resp.status_code == 200
    assert b"awaiting moderation" in resp.content
    # Response swaps the form container (contains form + success message)
    assert b"comment-form-container" in resp.content
    assert "HX-Request" in resp["Vary"]


@pytest.mark.django_db
@override_settings(TURNSTILE_SECRET_KEY="test-secret")
def test_htmx_post_returns_form_plus_oob_comment_when_approved(client, _article):
    """HTMX POST with successful Turnstile returns fresh form + OOB comment."""
    with patch("apps.comments.views._verify_turnstile", return_value=True):
        resp = _post_comment(client, _article, extra={"cf-turnstile-response": "tok"}, htmx=True)
    assert resp.status_code == 200
    content = resp.content.decode()
    # Fresh form container is the primary response
    assert "comment-form-container" in content
    assert "Comment posted!" in content
    # OOB swap appends the comment to #comments-list
    assert "hx-swap-oob" in content
    assert "Hello world" in content
    assert 'id="comments-empty-state" hx-swap-oob="delete"' in content
    comment = Comment.objects.get()
    assert comment.is_approved is True


@pytest.mark.django_db
def test_htmx_post_returns_form_with_errors_on_invalid(client, _article):
    """HTMX POST with invalid data returns form with errors (HTTP 200)."""
    cache.clear()
    resp = client.post(
        "/comments/post/",
        {"article_id": _article.id, "author_name": "T", "author_email": "t@t.com", "body": "   ", "honeypot": ""},
        HTTP_HX_REQUEST="true",
    )
    assert resp.status_code == 200
    assert b"comment-form-container" in resp.content
    assert b"Comment form errors" in resp.content
    assert "HX-Request" in resp["Vary"]
    assert Comment.objects.count() == 0


@pytest.mark.django_db
@override_settings(TURNSTILE_SECRET_KEY="test-secret")
def test_htmx_reply_returns_oob_reply_when_approved(client, _article, approved_comment):
    """Approved reply via HTMX returns compact reply partial via OOB swap."""
    cache.clear()
    with patch("apps.comments.views._verify_turnstile", return_value=True):
        resp = client.post(
            "/comments/post/",
            {
                "article_id": _article.id,
                "parent_id": approved_comment.id,
                "author_name": "Replier",
                "author_email": "r@r.com",
                "body": "Nice reply",
                "honeypot": "",
                "cf-turnstile-response": "tok",
            },
            HTTP_HX_REQUEST="true",
        )
    content = resp.content.decode()
    assert resp.status_code == 200
    # OOB targets a stable, explicit replies container for the parent comment.
    assert f'hx-swap-oob="beforeend:#replies-for-{approved_comment.id}"' in content
    # Verify content is rendered (not empty due to context mismatch)
    assert "Replier" in content
    assert "Nice reply" in content
    reply = Comment.objects.exclude(pk=approved_comment.pk).get()
    assert f"comment-{reply.id}" in content
    assert reply.parent_id == approved_comment.id
    assert reply.is_approved is True


@pytest.mark.django_db
def test_non_htmx_post_still_redirects(client, _article):
    """Non-HTMX POST continues to redirect (progressive enhancement)."""
    resp = _post_comment(client, _article)
    assert resp.status_code == 302
    assert resp["Location"].endswith("?commented=1")


@pytest.mark.django_db
def test_htmx_error_with_tampered_parent_id_falls_back_to_main_form(client, _article):
    """Tampered/non-numeric parent_id falls back to main form error response."""
    cache.clear()
    resp = client.post(
        "/comments/post/",
        {"article_id": _article.id, "parent_id": "not-a-number", "author_name": "T",
         "author_email": "t@t.com", "body": "   ", "honeypot": ""},
        HTTP_HX_REQUEST="true",
    )
    assert resp.status_code == 200
    assert b"comment-form-container" in resp.content


@pytest.mark.django_db
def test_htmx_invalid_reply_rerenders_reply_form_with_values(client, _article, approved_comment):
    """Invalid reply keeps user input and returns the reply form container."""
    cache.clear()
    resp = client.post(
        "/comments/post/",
        {
            "article_id": _article.id,
            "parent_id": approved_comment.id,
            "author_name": "Reply User",
            "author_email": "reply@example.com",
            "body": "   ",
            "honeypot": "",
        },
        HTTP_HX_REQUEST="true",
    )
    assert resp.status_code == 200
    content = resp.content.decode()
    assert f'id="reply-form-container-{approved_comment.id}"' in content
    assert "Comment form errors" in content
    assert 'value="Reply User"' in content
    assert "reply@example.com" in content


# ── Turnstile Integration ────────────────────────────────────────────────────


@pytest.mark.django_db
@override_settings(TURNSTILE_SECRET_KEY="test-secret")
def test_turnstile_failure_keeps_comment_unapproved(client, _article):
    """When Turnstile verification fails, comment stays unapproved."""
    with patch("apps.comments.views._verify_turnstile", return_value=False):
        _post_comment(client, _article, extra={"cf-turnstile-response": "bad-tok"})
    comment = Comment.objects.get()
    assert comment.is_approved is False


@pytest.mark.django_db
def test_turnstile_disabled_keeps_comment_unapproved(client, _article):
    """When TURNSTILE_SECRET_KEY is empty, comment stays unapproved."""
    _post_comment(client, _article)
    comment = Comment.objects.get()
    assert comment.is_approved is False


@pytest.mark.django_db
@override_settings(TURNSTILE_SECRET_KEY="test-secret", TURNSTILE_EXPECTED_HOSTNAME="nohypeai.com")
def test_turnstile_hostname_mismatch_rejects(client, _article):
    """Turnstile hostname mismatch keeps comment unapproved."""
    mock_resp = type("R", (), {"json": lambda self: {"success": True, "hostname": "evil.com"}})()
    with patch("apps.comments.views.http_requests.post", return_value=mock_resp):
        _post_comment(client, _article, extra={"cf-turnstile-response": "tok"})
    comment = Comment.objects.get()
    assert comment.is_approved is False


@pytest.mark.django_db
@override_settings(TURNSTILE_SECRET_KEY="test-secret")
def test_turnstile_timeout_fails_closed(client, _article):
    """Network error during Turnstile verification fails closed."""
    with patch("apps.comments.views.http_requests.post", side_effect=Exception("timeout")):
        _post_comment(client, _article, extra={"cf-turnstile-response": "tok"})
    comment = Comment.objects.get()
    assert comment.is_approved is False


# ── Polling ───────────────────────────────────────────────────────────────────


@pytest.mark.django_db
def test_comment_poll_returns_new_comments(_article, client, approved_comment):
    """Poll endpoint returns only comments after the given ID."""
    resp = client.get(f"/comments/poll/{_article.id}/?after_id=0")
    assert resp.status_code == 200
    assert b"Alice" in resp.content

    resp2 = client.get(f"/comments/poll/{_article.id}/?after_id={approved_comment.id}")
    assert resp2.status_code == 200
    assert b"Alice" not in resp2.content


@pytest.mark.django_db
def test_comment_poll_no_duplicates(_article, client, approved_comment):
    """Polling with current latest ID returns empty."""
    resp = client.get(f"/comments/poll/{_article.id}/?after_id={approved_comment.id}")
    assert b"comment-" not in resp.content


# ── Reactions ─────────────────────────────────────────────────────────────────


@pytest.mark.django_db
def test_react_creates_reaction(client, approved_comment):
    cache.clear()
    resp = client.post(
        f"/comments/{approved_comment.id}/react/",
        {"reaction_type": "heart"},
        HTTP_HX_REQUEST="true",
    )
    assert resp.status_code == 200
    assert CommentReaction.objects.count() == 1


@pytest.mark.django_db
def test_react_toggle_removes_reaction(client, approved_comment):
    """Second reaction of same type removes it (toggle)."""
    cache.clear()
    client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "heart"})
    client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "heart"})
    assert CommentReaction.objects.count() == 0


@pytest.mark.django_db
def test_react_different_types_coexist(client, approved_comment):
    cache.clear()
    client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "heart"})
    client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "plus_one"})
    assert CommentReaction.objects.count() == 2


@pytest.mark.django_db
def test_react_invalid_type_returns_400(client, approved_comment):
    cache.clear()
    resp = client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "invalid"})
    assert resp.status_code == 400


@pytest.mark.django_db
def test_react_on_unapproved_comment_returns_404(client, _article):
    cache.clear()
    comment = Comment.objects.create(
        article=_article, author_name="B", author_email="b@b.com", body="x", is_approved=False,
    )
    resp = client.post(f"/comments/{comment.id}/react/", {"reaction_type": "heart"})
    assert resp.status_code == 404


@pytest.mark.django_db
@override_settings(REACTION_RATE_LIMIT_PER_MINUTE=2)
def test_react_rate_limit(client, approved_comment):
    cache.clear()
    for _ in range(2):
        client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "heart"})
    resp = client.post(f"/comments/{approved_comment.id}/react/", {"reaction_type": "plus_one"})
    assert resp.status_code == 429


# ── CSP ───────────────────────────────────────────────────────────────────────


@pytest.mark.django_db
def test_csp_allows_turnstile(client, _article):
    """CSP header includes Cloudflare Turnstile domains."""
    resp = client.get(_article.url)
    csp = resp.get("Content-Security-Policy", "")
    assert "challenges.cloudflare.com" in csp
    assert "frame-src" in csp


# ── Purge Command Extension ──────────────────────────────────────────────────


@pytest.mark.django_db
def test_purge_clears_reaction_session_keys(home_page):
    index = ArticleIndexPage(title="Articles", slug="articles")
    home_page.add_child(instance=index)
    author = AuthorFactory()
    article = ArticlePage(title="A", slug="a", author=author, summary="s", body=[("rich_text", "<p>b</p>")])
    index.add_child(instance=article)
    article.save_revision().publish()

    comment = Comment.objects.create(
        article=article, author_name="X", author_email="x@x.com", body="y", is_approved=True,
    )
    reaction = CommentReaction.objects.create(
        comment=comment, reaction_type="heart", session_key="abc123",
    )
    CommentReaction.objects.filter(pk=reaction.pk).update(created_at=timezone.now() - timedelta(days=800))

    call_command("purge_old_comment_data")
    reaction.refresh_from_db()
    assert reaction.session_key == ""