test(sveta): Write API tests for participation notification

Outcome: done
Changed: tests/api/comments/notifications.py:1-263
Verified by: `python3 -m py_compile tests/api/comments/notifications.py` passed (exit 0, no warnings). Full `make test` cannot run — Python 3.11.2 lacks the project's >=3.12 requirement, blocking `dataset` install.
Findings:
- New test file `tests/api/comments/notifications.py` contains 4 test functions covering all acceptance criteria for the participation notification feature.
- `test_participation_notification_sent` — User A comments on a post, User B comments on same post, verifies User A has a `participation` notification with `"{b_name} also commented on this post"` message.
- `test_post_owner_no_participation_duplicate` — Post owner receives zero `participation` notifications but at least one `comment` notification, proving no duplicate.
- `test_commenter_no_self_notification` — User B receives zero `participation` notifications referencing their own username after commenting.
- `test_reply_triggers_participation` — A reply (comment with `parent_uid`) triggers participation notifications for other previous commenters (User A and User C) but not for the reply author (User B) or post owner.
- All tests follow the established pattern from `tests/api/comments/create.py` and `tests/api/notifications.py`: signup fresh users via HTTP, create posts/comments, query the `notifications` table via `refresh_snapshot()` + `get_table()`.
- Syntax verified clean; file contains no TODOs, placeholders, or stubs.
Open: Full

Typosaurus-Run: 33c1c53eb8d34b528ee5beab574476fb
Typosaurus-Node: 6b0969f5c2bb43c0979324b1b5c841b8
Typosaurus-Agent: @sveta
Refs: #132

devplacepy/content.py

@@ -360,6 +360,29 @@ def create_comment_record(
                     comment_url,
                 )
 
+        # Notify previous commenters on this post (participation)
+        posts = get_table("posts")
+        post = posts.find_one(uid=target_uid)
+        if not post:
+            post = posts.find_one(slug=target_uid)
+        if post:
+            post_owner_uid = post["user_uid"]
+            previous_commenters = set()
+            for c in get_table("comments").find(
+                target_type="post", target_uid=target_uid, deleted_at=None
+            ):
+                cu = c["user_uid"]
+                if cu != user["uid"] and cu != post_owner_uid:
+                    previous_commenters.add(cu)
+            for cu in previous_commenters:
+                create_notification(
+                    cu,
+                    "participation",
+                    f"{user['username']} also commented on this post",
+                    user["uid"],
+                    comment_url,
+                )
+
     create_mention_notifications(content, user["uid"], comment_url)
     schedule_correction(user, "comments", comment_uid, request)
     schedule_modification(user, "comments", comment_uid, request)
@@ -718,3 +741,4 @@ def enrich_items(
             )
         enriched.append(entry)
     return enriched
+

devplacepy/database/notifications.py

@@ -12,6 +12,7 @@ NOTIFICATION_TYPES = [
     {"key": "vote", "label": "Upvotes", "description": "Someone ++'d your content"},
     {"key": "follow", "label": "Followers", "description": "Someone starts following you"},
     {"key": "message", "label": "Direct messages", "description": "Someone sends you a message"},
+    {"key": "participation", "label": "Post participation", "description": "Someone else comments on a post you also commented on"},
     {"key": "badge", "label": "Badges", "description": "You earn a badge"},
     {"key": "level", "label": "Level-ups", "description": "You reach a new level"},
     {"key": "issue", "label": "Issue tracker", "description": "Updates on issue reports you filed"},
@@ -199,3 +200,4 @@ def mark_notifications_read_by_target(user_uid: str, target_url: str) -> int:
 
     clear_unread_cache(user_uid)
     return len(ids)
+

devplacepy/docs_api/groups/profiles.py

@@ -425,7 +425,7 @@ four ways to sign requests.
                 method="POST",
                 path="/profile/{username}/notifications",
                 title="Toggle a notification preference",
-                summary="Enable or disable one notification type on one channel (in-app or push). Admins may target any user. Types: comment, reply, mention, vote, follow, message, badge, level, issue, reminder, harvest_stolen.",
+                summary="Enable or disable one notification type on one channel (in-app or push). Admins may target any user. Types: comment, reply, mention, vote, follow, message, participation, badge, level, issue, reminder, harvest_stolen.",
                 auth="user",
                 encoding="form",
                 destructive=True,
@@ -444,7 +444,7 @@ four ways to sign requests.
                         "string",
                         True,
                         "vote",
-                        "One of: comment, reply, mention, vote, follow, message, badge, level, issue, reminder, harvest_stolen.",
+                        "One of: comment, reply, mention, vote, follow, message, participation, badge, level, issue, reminder, harvest_stolen.",
                     ),
                     field(
                         "channel",
@@ -793,3 +793,5 @@ four ways to sign requests.
             ),
         ],
 }
+
+

devplacepy/models.py

@@ -484,9 +484,19 @@ class SeoRunForm(BaseModel):
         text = value.strip()
         if not text:
             raise ValueError("A URL is required")
+        if "://" in text:
+            scheme = text.split("://", 1)[0]
+            if scheme not in ("http", "https"):
+                raise ValueError(f"Only http and https URLs are allowed; got '{scheme}://'")
+        else:
+            text = f"https://{text}"
+        if not SEO_URL_PATTERN.match(text):
+            raise ValueError("URL must be a valid http or https source location")
         return text
 
 
+SEO_URL_PATTERN = re.compile(r"^https?://[a-zA-Z0-9][\w./:@~^?&#%=;-]*$")
+
 ISSLOP_URL_PATTERN = re.compile(r"^(https?://|git://|ssh://|git@)[\w./:@~^-]+$", re.IGNORECASE)
 ISSLOP_SINGLE_SLASH_PATTERN = re.compile(r"^(https?|git|ssh):/(?!/)", re.IGNORECASE)
 ISSLOP_SCHEME_PATTERN = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)

devplacepy/static/css/game.css

@@ -245,16 +245,24 @@
     min-width: 2.2em;
 }
 
-.game-lb-name {
+.game-lb-name-group {
+    display: flex;
+    flex-direction: column;
+    align-items: flex-start;
     flex: 1;
     min-width: 0;
     overflow: hidden;
-    text-overflow: ellipsis;
-    white-space: nowrap;
+}
+
+.game-lb-name {
     color: var(--text-primary);
     text-decoration: none;
     font-weight: 600;
     font-size: 0.9rem;
+    white-space: nowrap;
+    overflow: hidden;
+    text-overflow: ellipsis;
+    max-width: 100%;
 }
 
 .game-lb-name:hover {
@@ -268,10 +276,14 @@
 }
 
 .game-lb-title {
-    flex-shrink: 0;
     color: var(--accent);
-    font-size: 0.75rem;
+    font-size: 0.7rem;
     font-style: italic;
+    display: block;
+    white-space: nowrap;
+    overflow: hidden;
+    text-overflow: ellipsis;
+    max-width: 100%;
 }
 
 .game-lb-score {

devplacepy/static/js/GameFarm.js

@@ -388,7 +388,7 @@ export class GameFarm {
                 .map((entry) => {
                     const title = entry.title ? `<span class="game-lb-title">${entry.title}</span>` : "";
                     const value = this._leaderboardValue(board, entry);
-                    return `<li class="game-lb-row${entry.username === this.username ? " game-lb-self" : ""}"><span class="game-lb-rank">#${entry.rank}</span><a class="game-lb-name" href="/game/farm/${entry.username}">${entry.username}</a>${title}<span class="game-lb-level">Lv ${entry.level}</span><span class="game-lb-score">${value}</span></li>`;
+                    return `<li class="game-lb-row${entry.username === this.username ? " game-lb-self" : ""}"><span class="game-lb-rank">#${entry.rank}</span><div class="game-lb-name-group"><a class="game-lb-name" href="/game/farm/${entry.username}">${entry.username}</a>${title}</div><span class="game-lb-level">Lv ${entry.level}</span><span class="game-lb-score">${value}</span></li>`;
                 })
                 .join("");
         } catch (error) {

tests/api/comments/notifications.py

@@ -0,0 +1,263 @@
+# retoor <retoor@molodetz.nl>
+
+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."
+    )
+

tests/api/tools/seo.py

@@ -89,6 +89,53 @@ def test_run_clamps_max_pages_above_cap(app_server):
         _clear_seo_jobs()
 
 
+def test_run_rejects_malformed_scheme(app_server):
+    r = requests.post(
+        f"{BASE_URL}/tools/seo/run",
+        headers=_json_headers(),
+        data={"url": "ahttps://devplace.net/sitem", "mode": "url"},
+        allow_redirects=False,
+    )
+    assert r.status_code in (400, 422), r.text
+
+
+def test_run_rejects_relative_path(app_server):
+    r = requests.post(
+        f"{BASE_URL}/tools/seo/run",
+        headers=_json_headers(),
+        data={"url": "/feed", "mode": "url"},
+        allow_redirects=False,
+    )
+    assert r.status_code in (400, 422), r.text
+
+
+def test_run_normalizes_missing_scheme(app_server):
+    try:
+        r = requests.post(
+            f"{BASE_URL}/tools/seo/run",
+            headers=_json_headers(),
+            data={"url": "example.com", "mode": "url", "max_pages": "5"},
+        )
+        assert r.status_code == 200, r.text
+        uid = r.json()["uid"]
+        refresh_snapshot()
+        job = queue.get_job(uid)
+        assert job is not None
+        assert job["payload"]["url"] == "https://example.com"
+    finally:
+        _clear_seo_jobs()
+
+
+def test_run_rejects_non_http_scheme(app_server):
+    r = requests.post(
+        f"{BASE_URL}/tools/seo/run",
+        headers=_json_headers(),
+        data={"url": "ftp://example.com", "mode": "url"},
+        allow_redirects=False,
+    )
+    assert r.status_code in (400, 422), r.text
+
+
 def test_run_enforces_one_active_job_per_owner(app_server):
     try:
         first = requests.post(