intent/tests/test_comments.py

import pytest
from django.core.exceptions import PermissionDenied, ValidationError
from django.urls import reverse

from intent.comments import services
from intent.comments.models import Comment, CommentThread, ThreadKind
from intent.features.services import create_feature, update_feature
from intent.issues.models import IssueStatus
from intent.issues.services import create_issue, promote_thread, set_status

DESCRIPTION = "Intro line\n\nSecond para\nwraps here.\n"


def furl(name, feature, **kwargs):
    kwargs = {
        "org_slug": "catbriar",
        "project_slug": "intent",
        "feature_key": feature.key,
        **kwargs,
    }
    return reverse(f"features:{name}", kwargs=kwargs)


def curl(name, **kwargs):
    return reverse(
        f"comments:{name}", kwargs={"org_slug": "catbriar", "project_slug": "intent", **kwargs}
    )


@pytest.fixture
def feature(open_project, owner):
    return create_feature(open_project, owner, title="Sign in", description=DESCRIPTION)


@pytest.fixture
def thread(feature, outsider):
    return services.start_feature_thread(feature, outsider, body="Why two paragraphs?")


def test_line_thread_must_match_current_revision_and_lines(feature, outsider, owner):
    t = services.start_feature_thread(
        feature, outsider, body="Hm", revision=1, line_start=3, line_end=4
    )
    assert t.kind == ThreadKind.LINE
    with pytest.raises(ValidationError, match="changed"):
        services.start_feature_thread(
            feature, outsider, body="x", revision=2, line_start=1, line_end=1
        )
    with pytest.raises(ValidationError):
        services.start_feature_thread(
            feature, outsider, body="x", revision=1, line_start=3, line_end=9
        )
    with pytest.raises(ValidationError):
        services.start_feature_thread(feature, outsider, body="  ")


def test_private_project_outsiders_cannot_comment(project, owner, make_user):
    feature = create_feature(project, owner, title="Secret")
    with pytest.raises(PermissionDenied):
        services.start_feature_thread(feature, make_user(), body="hi")


def test_edit_keeps_history_and_only_author_edits(thread, outsider, owner):
    comment = thread.comments.get()
    with pytest.raises(PermissionDenied):
        services.edit_comment(comment, owner, body="Rewritten")
    services.edit_comment(comment, outsider, body="Why two?")
    comment.refresh_from_db()
    assert comment.edited_at and comment.edits.get().body == "Why two paragraphs?"
    with pytest.raises(ValidationError):
        services.edit_comment(comment, outsider, body="Why two?")


def test_delete_leaves_tombstone_and_drops_history(thread, outsider, make_user, owner, editor):
    comment = thread.comments.get()
    services.edit_comment(comment, outsider, body="Edited")
    with pytest.raises(PermissionDenied):
        services.delete_comment(comment, make_user())
    with pytest.raises(PermissionDenied):
        services.delete_comment(comment, editor)  # Collaborators don't moderate; owners do.
    services.delete_comment(comment, owner)
    comment.refresh_from_db()
    assert comment.is_deleted and comment.body == "" and not comment.edits.exists()
    assert comment.deleted_by == owner


def test_resolve_by_thread_author_or_team(thread, outsider, editor, make_user):
    with pytest.raises(PermissionDenied):
        services.set_resolved(thread, make_user(), True)
    services.set_resolved(thread, outsider, True)
    assert thread.is_resolved
    services.set_resolved(thread, editor, False)
    assert not thread.is_resolved


def test_issue_threads_cannot_be_resolved(open_project, outsider, owner):
    issue = create_issue(open_project, outsider, title="Bug")
    with pytest.raises(PermissionDenied):
        services.set_resolved(issue.threads.get(), owner, True)


def test_promote_by_author_or_team_only(thread, feature, outsider, make_user, editor):
    services.reply(thread, editor, body="Good question")
    with pytest.raises(PermissionDenied):
        promote_thread(thread, make_user(), title="X", body="")
    issue = promote_thread(thread, outsider, title="Two paragraphs", body="Quoted")
    thread.refresh_from_db()
    assert thread.promoted_to == issue and thread.is_resolved
    assert list(issue.features.all()) == [feature]
    assert issue.author == outsider
    with pytest.raises(PermissionDenied):
        promote_thread(thread, editor, title="Again", body="")


def test_promotion_text_quotes_lines_and_comments(feature, outsider, editor):
    from intent.issues.services import promotion_text

    t = services.start_feature_thread(
        feature, outsider, body="Unclear wording", revision=1, line_start=3, line_end=4
    )
    services.reply(t, editor, body="Agreed")
    title, body = promotion_text(t)
    assert title == "Unclear wording"
    assert "lines 3–4 of revision 1" in body
    assert "> Second para\n> wraps here." in body
    assert "**eddie** wrote:\n\n> Agreed" in body


# Views


def test_feature_page_shows_line_threads_under_their_block(client, feature, outsider):
    services.start_feature_thread(
        feature, outsider, body="About para two", revision=1, line_start=3, line_end=4
    )
    client.force_login(outsider)
    page = client.get(furl("detail", feature)).content.decode()
    block = page.index('id="L3-4"')
    assert block < page.index("About para two") < page.index('id="discussion"')
    assert 'id="body-L1-1"' in page  # A comment form for each block.


def test_line_threads_go_outdated_only_when_text_changes(client, feature, outsider, owner):
    services.start_feature_thread(
        feature, outsider, body="About para two", revision=1, line_start=3, line_end=4
    )
    update_feature(feature, owner, note="Withdraw", status="withdrawn")  # Text unchanged.
    client.force_login(outsider)
    page = client.get(furl("detail", feature)).content.decode()
    assert "Comments on earlier revisions" not in page
    update_feature(feature, owner, note="Reword", description="Intro\n\nNew text\n")
    page = client.get(furl("detail", feature)).content.decode()
    outdated = page.index("Comments on earlier revisions")
    assert outdated < page.index("About para two")
    assert "Second para\nwraps here." in page  # The original lines are quoted.
    assert furl("revision", feature, revision=1) + "#L3-4" in page


def test_create_thread_via_form(client, feature, outsider):
    client.force_login(outsider)
    response = client.post(furl("thread_create", feature), {"body": "General point"})
    thread = CommentThread.objects.get()
    assert response["Location"] == furl("detail", feature) + f"#thread-{thread.pk}"
    assert thread.kind == ThreadKind.FEATURE


def test_stale_line_comment_shows_the_draft(client, feature, outsider, owner):
    update_feature(feature, owner, note="Edit", description="Changed\n")
    client.force_login(outsider)
    response = client.post(
        furl("thread_create", feature),
        {"body": "My long thought", "revision": 1, "line_start": 1, "line_end": 1},
    )
    assert response.status_code == 400
    assert b"My long thought" in response.content and b"changed" in response.content
    assert not CommentThread.objects.exists()


def test_reply_without_js_redirects_to_the_comment(client, thread, outsider):
    client.force_login(outsider)
    response = client.post(curl("reply", thread_id=thread.pk), {"body": "More"})
    comment = Comment.objects.get(body="More")
    assert response["Location"].endswith(f"#comment-{comment.pk}")


def test_reply_with_htmx_returns_the_thread_fragment(client, thread, outsider):
    client.force_login(outsider)
    response = client.post(
        curl("reply", thread_id=thread.pk), {"body": "More"}, headers={"HX-Request": "true"}
    )
    html = response.content.decode()
    assert response.status_code == 200
    assert html.lstrip().startswith(f'<div id="thread-{thread.pk}"')
    assert "<html" not in html
    comment = Comment.objects.get(body="More")
    assert f'id="comment-{comment.pk}" class="comment" tabindex="-1" data-autofocus' in html


def test_htmx_empty_reply_keeps_thread_and_shows_error(client, thread, outsider):
    client.force_login(outsider)
    response = client.post(
        curl("reply", thread_id=thread.pk), {"body": " "}, headers={"HX-Request": "true"}
    )
    assert response.status_code == 200 and b'role="alert"' in response.content


def test_resolve_edit_delete_via_views(client, thread, outsider, owner):
    comment = thread.comments.get()
    client.force_login(outsider)
    client.post(curl("resolve", thread_id=thread.pk), {"resolved": "1"})
    thread.refresh_from_db()
    assert thread.is_resolved
    client.post(curl("edit", comment_id=comment.pk), {"body": "Edited"})
    history = client.get(curl("history", comment_id=comment.pk)).content.decode()
    assert "Why two paragraphs?" in history and "Edited" in history
    client.force_login(owner)
    client.post(curl("delete", comment_id=comment.pk))
    comment.refresh_from_db()
    assert comment.is_deleted


def test_threads_from_another_project_are_404(client, thread, org, owner):
    from intent.projects.services import create_project

    create_project(
        org, owner, slug="other", name="Other", feature_prefix="OTH", visibility="public",
        open_contribution=True,
    )  # fmt: skip
    client.force_login(owner)
    other = reverse(
        "comments:reply",
        kwargs={"org_slug": "catbriar", "project_slug": "other", "thread_id": thread.pk},
    )
    assert client.post(other, {"body": "Sneaky"}).status_code == 404


def test_promote_page_prefills_and_creates_issue(client, thread, outsider):
    client.force_login(outsider)
    page = client.get(curl("promote", thread_id=thread.pk)).content.decode()
    assert "Why two paragraphs?" in page
    response = client.post(
        curl("promote", thread_id=thread.pk), {"title": "Two paragraphs", "body": "See thread"}
    )
    thread.refresh_from_db()
    issue = thread.promoted_to
    assert response["Location"].endswith(f"/issues/{issue.number}/")
    issue_page = client.get(response["Location"]).content.decode()
    assert f"#thread-{thread.pk}" in issue_page  # Links back to where it started.


def test_someone_elses_thread_cannot_be_promoted_by_outsider(client, thread, make_user):
    client.force_login(make_user())
    assert client.get(curl("promote", thread_id=thread.pk)).status_code == 403


def test_closed_issue_still_accepts_comments(client, open_project, outsider, editor):
    issue = create_issue(open_project, outsider, title="Bug")
    set_status(issue, editor, IssueStatus.REJECTED)
    client.force_login(outsider)
    client.post(curl("reply", thread_id=issue.threads.get().pk), {"body": "Why rejected?"})
    assert Comment.objects.filter(body="Why rejected?").exists()