from datetime import datetime, timedelta, timezone from tests.conftest import BASE_URL from devplacepy.database import get_table from devplacepy.utils import generate_uid def test_notifications_pagination(alice): page, _ = alice alice_row = get_table("users").find_one(username="alice_test") notifications_table = get_table("notifications") notifications_table.delete(user_uid=alice_row["uid"]) now = datetime.now(timezone.utc) for i in range(30): notifications_table.insert({ "uid": generate_uid(), "user_uid": alice_row["uid"], "type": "test", "message": f"pagination-test-msg-{i:02d}", "related_uid": alice_row["uid"], "target_url": "/feed", "read": False, "created_at": (now - timedelta(seconds=i)).isoformat(), }) page.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") cards = page.locator(".notification-card") assert cards.count() == 25, f"expected 25 cards on page 1, got {cards.count()}" load_more = page.locator(".load-more-wrap a") load_more.wait_for(state="visible", timeout=5000) href = load_more.get_attribute("href") assert href and "before=" in href, f"Load More should carry cursor: {href}" page.goto(f"{BASE_URL}{href}", wait_until="domcontentloaded") cards2 = page.locator(".notification-card") assert cards2.count() == 5, f"expected 5 cards on page 2, got {cards2.count()}" assert page.locator(".load-more-wrap").count() == 0, "Load More should be absent on final page" notifications_table.delete(user_uid=alice_row["uid"]) def test_notifications_no_load_more_when_under_page_size(alice): page, _ = alice alice_row = get_table("users").find_one(username="alice_test") notifications_table = get_table("notifications") notifications_table.delete(user_uid=alice_row["uid"]) now = datetime.now(timezone.utc) for i in range(3): notifications_table.insert({ "uid": generate_uid(), "user_uid": alice_row["uid"], "type": "test", "message": f"small-set-msg-{i}", "related_uid": alice_row["uid"], "target_url": "/feed", "read": False, "created_at": (now - timedelta(seconds=i)).isoformat(), }) page.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") assert page.locator(".notification-card").count() == 3 assert page.locator(".load-more-wrap").count() == 0 notifications_table.delete(user_uid=alice_row["uid"]) def test_notifications_page_loads(alice): page, _ = alice page.goto(f"{BASE_URL}/notifications") assert page.is_visible("h2:has-text('Notifications')") def test_mark_all_read(alice): page, _ = alice page.goto(f"{BASE_URL}/notifications") clear = page.locator("button:has-text('Clear')") if clear.is_visible(): clear.click() page.wait_for_timeout(500) def test_notifications_navigation(alice): page, _ = alice page.goto(f"{BASE_URL}/feed") page.goto(f"{BASE_URL}/notifications") assert page.is_visible("h2:has-text('Notifications')") def test_notifications_bell_visible(alice): page, _ = alice page.goto(f"{BASE_URL}/feed") bell = page.locator(".topnav-icon[href='/notifications']") assert bell.is_visible() def test_notifications_empty_state(alice): page, _ = alice page.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") assert page.is_visible("h2:has-text('Notifications')") or page.is_visible("text=No notifications yet") def test_notifications_header_has_clear(alice): page, _ = alice page.goto(f"{BASE_URL}/notifications") clear = page.locator("button:has-text('Clear')") assert clear.is_visible() def test_notifications_topnav_user(alice): page, user = alice page.goto(f"{BASE_URL}/notifications") assert page.is_visible(f"text={user['username']}") def test_notifications_avatar_visible(alice): page, _ = alice page.goto(f"{BASE_URL}/notifications") avatar = page.locator("img.avatar-img").first assert avatar.is_visible() def test_vote_notification_on_post(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pb.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pb.locator(".feed-fab").first.click() pb.fill("#post-content", "Post for vote notification test") pb.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pb.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") post_url = pb.url pa.goto(post_url, wait_until="domcontentloaded") pa.wait_for_timeout(1000) vote_btn = pa.locator("button.post-action-btn").filter(has_text="+").first vote_btn.wait_for(state="visible", timeout=10000) vote_btn.click() pa.wait_for_timeout(1500) pa.wait_for_timeout(500) pa_body = pa.locator("body").text_content() assert "Internal Server Error" not in pa_body, f"Alice got 500 after vote: {pa_body[:300]}" pb.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pb.wait_for_timeout(1500) body = pb.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications page: {body[:500]}" assert "alice_test" in body, f"Expected 'alice_test' in notifications, got: {body[:500]}" assert "++'d" in body, f"Expected '++\\'d' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_comment_notification_on_post(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pa.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pa.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pa.locator(".feed-fab").first.click() pa.fill("#post-content", "Post for comment notification test") pa.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pa.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") post_url = pa.url pb.goto(post_url, wait_until="domcontentloaded") pb.wait_for_timeout(1000) comment_textarea = pb.locator("form.comment-form textarea[name='content']").first comment_textarea.wait_for(state="visible", timeout=10000) comment_textarea.fill("Nice post!") pb.locator("button.comment-form-submit").first.click() pb.wait_for_timeout(1500) pb_body = pb.locator("body").text_content() assert "Internal Server Error" not in pb_body, f"Bob got 500: {pb_body[:300]}" pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(1500) body = pa.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications: {body[:500]}" assert "bob_test" in body, f"Expected 'bob_test' in notifications, got: {body[:500]}" assert "commented on your" in body, f"Expected 'commented on your' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_follow_notification(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/profile/alice_test", wait_until="domcontentloaded") pb.wait_for_timeout(1000) follow_btn = pb.locator("button:has-text('Follow')") if follow_btn.is_visible(): follow_btn.click() pb.wait_for_timeout(1000) pb_body = pb.locator("body").text_content() assert "Internal Server Error" not in pb_body, f"Bob got 500: {pb_body[:300]}" pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(1500) body = pa.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications: {body[:500]}" assert "bob_test" in body, f"Expected 'bob_test' in notifications, got: {body[:500]}" assert "started following you" in body, f"Expected 'started following you' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_message_notification(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/messages?search=alice_test", wait_until="domcontentloaded") pb.wait_for_timeout(2000) msg_input = pb.locator("input[name='content']").first msg_input.wait_for(state="visible", timeout=10000) msg_input.fill("Hello from bob_test!") pb.locator("button[type='submit']").last.click() pb.wait_for_timeout(1500) pb_body = pb.locator("body").text_content() assert "Internal Server Error" not in pb_body, f"Bob got 500: {pb_body[:300]}" pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(1500) body = pa.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications: {body[:500]}" assert "bob_test" in body, f"Expected 'bob_test' in notifications, got: {body[:500]}" assert "sent you a message" in body, f"Expected 'sent you a message' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_mention_notification_in_post(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pb.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pb.locator(".feed-fab").first.click() pb.fill("#post-content", "Check this out @alice_test how are you?") pb.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pb.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(2000) body = pa.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications: {body[:500]}" assert "bob_test" in body, f"Expected 'bob_test' in notifications for mention, got: {body[:500]}" assert "mentioned you" in body, f"Expected 'mentioned you' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_mention_notification_in_comment(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pa.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pa.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pa.locator(".feed-fab").first.click() pa.fill("#post-content", "Post for mention in comment test") pa.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pa.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") post_url = pa.url pb.goto(post_url, wait_until="domcontentloaded") pb.wait_for_timeout(1000) comment_textarea = pb.locator("form.comment-form textarea[name='content']").first comment_textarea.wait_for(state="visible", timeout=10000) comment_textarea.fill("Hey @alice_test look at this!") pb.locator("button.comment-form-submit").first.click() pb.wait_for_timeout(1500) pb_body = pb.locator("body").text_content() assert "Internal Server Error" not in pb_body, f"Bob got 500: {pb_body[:300]}" pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(2000) body = pa.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications: {body[:500]}" assert "bob_test" in body, f"Expected 'bob_test' in notifications for mention, got: {body[:500]}" assert "mentioned you" in body, f"Expected 'mentioned you' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_reply_notification(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pa.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pa.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pa.locator(".feed-fab").first.click() pa.fill("#post-content", "Post for reply notification test") pa.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pa.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") post_url = pa.url pa.goto(post_url, wait_until="domcontentloaded") pa.wait_for_timeout(1000) comment_textarea = pa.locator("form.comment-form textarea[name='content']").first comment_textarea.wait_for(state="visible", timeout=10000) comment_textarea.fill("Alice's top-level comment") pa.locator("button.comment-form-submit").first.click() pa.wait_for_timeout(1500) pb.goto(post_url, wait_until="domcontentloaded") pb.wait_for_timeout(1000) reply_btn = pb.locator("button:has-text('Reply')").first if reply_btn.is_visible(): reply_btn.click() pb.wait_for_timeout(500) reply_textarea = pb.locator("form.comment-form textarea[name='content']").first if reply_textarea.is_visible(): reply_textarea.fill("Bob's reply to Alice's comment") pb.locator("button.comment-form-submit").first.click() pb.wait_for_timeout(1500) pb_body = pb.locator("body").text_content() assert "Internal Server Error" not in pb_body, f"Bob got 500: {pb_body[:300]}" pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(2000) body = pa.locator("body").text_content() assert "Internal Server Error" not in body, f"Got 500 error on notifications: {body[:500]}" assert "bob_test" in body, f"Expected 'bob_test' in notifications, got: {body[:500]}" assert "replied to your comment" in body, f"Expected 'replied to your comment' in notifications, got: {body[:500]}" ctx_a.close() ctx_b.close() def test_mention_notification_names_actor(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pb.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pb.locator(".feed-fab").first.click() pb.fill("#post-content", "Naming check @alice_test please read") pb.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pb.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(1500) texts = pa.locator(".notification-text").all_text_contents() assert any("@bob_test mentioned you" in t for t in texts), f"mention message must name the actor (bob): {texts}" assert not any("@alice_test mentioned you" in t for t in texts), f"mention message must not name the mentioned user (alice): {texts}" ctx_a.close() ctx_b.close() def test_comment_notification_click_opens_comment(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pa.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pa.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pa.locator(".feed-fab").first.click() pa.fill("#post-content", "Post for click navigation test") pa.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pa.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") post_url = pa.url pb.goto(post_url, wait_until="domcontentloaded") textarea = pb.locator("form.comment-form textarea[name='content']").first textarea.wait_for(state="visible", timeout=10000) textarea.fill("Bob comment that alice should jump to") pb.locator("button.comment-form-submit").first.click() pb.wait_for_timeout(1500) pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") card = pa.locator(".notification-card").filter(has_text="commented on your post").first card.wait_for(state="visible", timeout=10000) href = card.locator("a.card-link").get_attribute("href") assert href.startswith("/notifications/open/"), f"unexpected href: {href}" card.locator("a.card-link").click() pa.wait_for_url("**/posts/**", timeout=10000, wait_until="domcontentloaded") assert "#comment-" in pa.url, f"click did not deep-link to a comment: {pa.url}" comment_uid = pa.url.split("#comment-")[1] pa.locator(f"#comment-{comment_uid}").wait_for(state="visible", timeout=10000) pa.wait_for_selector(f"#comment-{comment_uid}.comment-highlight", timeout=5000) pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") pa.wait_for_timeout(1000) target = pa.locator(f'.notification-card:has(a.card-link[href="{href}"])') target.wait_for(state="visible", timeout=10000) assert "unread" not in (target.get_attribute("class") or ""), "opening a notification should mark it read" ctx_a.close() ctx_b.close() def test_follow_notification_click_opens_profile(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/profile/alice_test", wait_until="domcontentloaded") pb.wait_for_timeout(1000) follow_btn = pb.locator("button:has-text('Follow')") if follow_btn.is_visible(): follow_btn.click() pb.wait_for_timeout(1000) pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") card = pa.locator(".notification-card").filter(has_text="started following you").first card.wait_for(state="visible", timeout=10000) card.locator("a.card-link").click() pa.wait_for_url("**/profile/bob_test", timeout=10000, wait_until="domcontentloaded") assert pa.url.endswith("/profile/bob_test"), f"follow notification should open the follower profile: {pa.url}" ctx_a.close() ctx_b.close() def test_message_notification_click_opens_conversation(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/messages?search=alice_test", wait_until="domcontentloaded") pb.wait_for_timeout(2000) msg_input = pb.locator("input[name='content']").first msg_input.wait_for(state="visible", timeout=10000) msg_input.fill("Click-through message from bob") pb.locator("button[type='submit']").last.click() pb.wait_for_timeout(1500) pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") card = pa.locator(".notification-card").filter(has_text="sent you a message").first card.wait_for(state="visible", timeout=10000) card.locator("a.card-link").click() pa.wait_for_url("**/messages**", timeout=10000, wait_until="domcontentloaded") assert "with_uid=" in pa.url, f"message notification should open the conversation: {pa.url}" ctx_a.close() ctx_b.close() def test_vote_notification_click_opens_target(app_server, browser, seeded_db): from tests.conftest import login_user ctx_a = browser.new_context(viewport={"width": 1400, "height": 900}) ctx_b = browser.new_context(viewport={"width": 1400, "height": 900}) pa = ctx_a.new_page() pb = ctx_b.new_page() pa.set_default_timeout(15000) pb.set_default_timeout(15000) login_user(pa, seeded_db["alice"]) login_user(pb, seeded_db["bob"]) pb.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded") pb.locator(".feed-fab").first.wait_for(state="visible", timeout=10000) pb.locator(".feed-fab").first.click() pb.fill("#post-content", "Post for vote click-through test") pb.locator("#create-post-modal button.btn-primary:has-text('Post')").click() pb.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded") post_path = "/posts/" + pb.url.split("/posts/")[1] pa.goto(f"{BASE_URL}{post_path}", wait_until="domcontentloaded") pa.wait_for_timeout(1000) vote_btn = pa.locator("button.post-action-btn").filter(has_text="+").first vote_btn.wait_for(state="visible", timeout=10000) vote_btn.click() pa.wait_for_timeout(1500) pb.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded") card = pb.locator(".notification-card").filter(has_text="++'d").first card.wait_for(state="visible", timeout=10000) card.locator("a.card-link").click() pb.wait_for_url(f"**{post_path}", timeout=10000, wait_until="domcontentloaded") assert post_path in pb.url, f"vote notification should open the voted post: {pb.url}" ctx_a.close() ctx_b.close()