diff --git a/tests/api/comments/notifications.py b/tests/api/comments/notifications.py new file mode 100644 index 00000000..1bfa977f --- /dev/null +++ b/tests/api/comments/notifications.py @@ -0,0 +1,263 @@ +# retoor + +import time +import pytest +import requests +from tests.conftest import BASE_URL +from devplacepy.database import get_table, refresh_snapshot, set_setting + +_COUNTER = [0] + + +@pytest.fixture(scope="module", autouse=True) +def _participation_settings(app_server): + for key, value in { + "rate_limit_per_minute": "1000000", + "rate_limit_window_seconds": "60", + "registration_open": "1", + "maintenance_mode": "0", + "max_upload_size_mb": "10", + "allowed_file_types": "", + "max_attachments_per_resource": "10", + "session_max_age_days": "7", + "session_remember_days": "30", + "news_service_interval": "3600", + "news_grade_threshold": "7", + }.items(): + set_setting(key, value) + yield + + +def _db_user(username): + refresh_snapshot() + return get_table("users").find_one(username=username) + + +def _unique(prefix="pn"): + _COUNTER[0] += 1 + return f"{prefix}{int(time.time() * 1000)}{_COUNTER[0]}" + + +def _signup(): + name = _unique("pnuser") + s = requests.Session() + s.post( + f"{BASE_URL}/auth/signup", + data={ + "username": name, + "email": f"{name}@t.dev", + "password": "secret123", + "confirm_password": "secret123", + }, + allow_redirects=True, + ) + return s, name + + +def _create_post(session): + r = session.post( + f"{BASE_URL}/posts/create", + data={ + "title": _unique("pnpost"), + "content": "Post for participation notification test.", + "topic": "devlog", + }, + allow_redirects=False, + ) + slug = r.headers["location"].split("/posts/")[-1] + refresh_snapshot() + return get_table("posts").find_one(slug=slug)["uid"] + + +def _comment_on_post(session, post_uid, content): + r = session.post( + f"{BASE_URL}/comments/create", + data={ + "content": content, + "target_type": "post", + "post_uid": post_uid, + "target_uid": post_uid, + }, + allow_redirects=False, + ) + assert r.status_code in (302, 303), ( + f"Comment creation failed: {r.status_code} {r.text[:300]}" + ) + + +def _reply_to_comment(session, post_uid, parent_uid, content): + r = session.post( + f"{BASE_URL}/comments/create", + data={ + "content": content, + "target_type": "post", + "post_uid": post_uid, + "target_uid": post_uid, + "parent_uid": parent_uid, + }, + allow_redirects=False, + ) + assert r.status_code in (302, 303), ( + f"Reply creation failed: {r.status_code} {r.text[:300]}" + ) + + +def _find_comment(post_uid, content): + refresh_snapshot() + return get_table("comments").find_one(target_uid=post_uid, content=content) + + +def _notifications_for(user_uid): + refresh_snapshot() + return list( + get_table("notifications").find( + user_uid=user_uid, order_by=["-created_at"] + ) + ) + + +def _participation_notifications_for(user_uid): + refresh_snapshot() + return list( + get_table("notifications").find( + user_uid=user_uid, type="participation", order_by=["-created_at"] + ) + ) + + +def test_participation_notification_sent(app_server): + """User A receives a participation notification when User B comments + on a post that User A previously commented on.""" + owner_session, owner_name = _signup() + post_uid = _create_post(owner_session) + + a_session, a_name = _signup() + b_session, b_name = _signup() + + _comment_on_post(a_session, post_uid, "User A first comment") + _comment_on_post(b_session, post_uid, "User B comment") + + a_user = _db_user(a_name) + assert a_user is not None + + participation_notifs = _participation_notifications_for(a_user["uid"]) + assert len(participation_notifs) >= 1, ( + f"User {a_name} should have at least one participation notification, " + f"got {len(participation_notifs)}" + ) + latest = participation_notifs[0] + assert latest["type"] == "participation" + assert b_name in latest["message"], ( + f"Expected notification message to contain {b_name!r}, " + f"got {latest['message']!r}" + ) + assert "also commented" in latest["message"], ( + f"Expected 'also commented' in message, got {latest['message']!r}" + ) + + +def test_post_owner_no_participation_duplicate(app_server): + """Post owner does NOT receive a participation notification. + They already receive the 'comment' notification.""" + owner_session, owner_name = _signup() + post_uid = _create_post(owner_session) + + a_session, a_name = _signup() + b_session, b_name = _signup() + + _comment_on_post(a_session, post_uid, "User A comment for owner test") + _comment_on_post(b_session, post_uid, "User B comment for owner test") + + owner = _db_user(owner_name) + assert owner is not None + + participation_notifs = _participation_notifications_for(owner["uid"]) + assert len(participation_notifs) == 0, ( + f"Post owner {owner_name} should have zero participation notifications, " + f"got {len(participation_notifs)}: " + f"{[n['message'] for n in participation_notifs]}" + ) + + all_notifs = _notifications_for(owner["uid"]) + comment_notifs = [n for n in all_notifs if n["type"] == "comment"] + assert len(comment_notifs) >= 1, ( + f"Post owner should have at least one 'comment' notification, " + f"got {len(comment_notifs)}" + ) + + +def test_commenter_no_self_notification(app_server): + """Commenter does NOT receive a participation notification + for their own comment.""" + owner_session, _ = _signup() + post_uid = _create_post(owner_session) + + a_session, a_name = _signup() + _comment_on_post(a_session, post_uid, "User A self-test comment") + + b_session, b_name = _signup() + _comment_on_post(b_session, post_uid, "User B self-test comment") + + b_user = _db_user(b_name) + assert b_user is not None + + participation_notifs = _participation_notifications_for(b_user["uid"]) + self_notifs = [ + n for n in participation_notifs if b_name in n["message"] + ] + assert len(self_notifs) == 0, ( + f"User {b_name} should not have a participation notification " + f"about themselves, got {len(self_notifs)}" + ) + + +def test_reply_triggers_participation(app_server): + """A reply to a comment triggers participation notifications + for other previous commenters (excluding the reply author and post owner). + The implementation sends participation for every comment on a post, + both top-level and replies.""" + owner_session, owner_name = _signup() + post_uid = _create_post(owner_session) + + a_session, a_name = _signup() + _comment_on_post(a_session, post_uid, "User A top-level comment") + a_comment = _find_comment(post_uid, "User A top-level comment") + + c_session, c_name = _signup() + _comment_on_post(c_session, post_uid, "User C third participant comment") + + b_session, b_name = _signup() + _reply_to_comment(b_session, post_uid, a_comment["uid"], "User B reply") + + a_user = _db_user(a_name) + c_user = _db_user(c_name) + + a_participation = _participation_notifications_for(a_user["uid"]) + c_participation = _participation_notifications_for(c_user["uid"]) + + a_has_participation = any( + b_name in n["message"] for n in a_participation + ) + c_has_participation = any( + b_name in n["message"] for n in c_participation + ) + + assert a_has_participation, ( + f"User {a_name} (parent commenter) should receive a participation " + f"notification when a reply is posted on the same post. " + f"Notifications: {[n['message'] for n in a_participation]}" + ) + assert c_has_participation, ( + f"User {c_name} (previous commenter) should receive a participation " + f"notification when a reply is posted on the same post. " + f"Notifications: {[n['message'] for n in c_participation]}" + ) + + b_user = _db_user(b_name) + b_participation = _participation_notifications_for(b_user["uid"]) + b_self = [n for n in b_participation if b_name in n["message"]] + assert len(b_self) == 0, ( + f"Reply author {b_name} should not receive a participation " + f"notification about themselves." + ) +