devplacepy/tests/e2e/messages.py

# retoor <retoor@molodetz.nl>

import re
import time
from playwright.sync_api import expect
from tests.conftest import BASE_URL, paste_image
from devplacepy.database import get_table
import requests
def _seed_news_seo():
    from datetime import datetime, timezone
    from devplacepy.database import get_table
    from devplacepy.utils import generate_uid, make_combined_slug

    uid = generate_uid()
    title = f"SEO Test News Article {uid.split('-')[-1]}"
    slug = make_combined_slug(title, uid)
    get_table("news").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "slug": slug,
            "title": title,
            "description": "A seeded news article for SEO tests.",
            "content": "Body content for the seeded article.",
            "url": "https://example.com/article",
            "source_name": "ExampleSource",
            "status": "published",
            "synced_at": datetime.now(timezone.utc).isoformat(),
            "show_on_landing": 0,
            "grade": 8,
        }
    )
    return slug, uid
def _seed_owner():
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table

    uid = str(uuid4())
    get_table("users").insert(
        {
            "uid": uid,
            "username": f"seo_{uid[:8]}",
            "terms_version": "1",
            "email": f"{uid[:8]}@seo.test",
            "password_hash": "x",
            "role": "Member",
            "is_active": True,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return uid
def _seed_post_seo(image=None):
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    uid = str(uuid4())
    slug = make_combined_slug("SEO Detail Post", uid)
    get_table("posts").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": owner,
            "slug": slug,
            "title": "SEO Detail Post",
            "content": "Body text for the SEO detail post.",
            "topic": "general",
            "project_uid": None,
            "image": image,
            "stars": 0,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return slug, uid
def _seed_gist():
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    uid = str(uuid4())
    slug = make_combined_slug("SEO Detail Gist", uid)
    get_table("gists").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": owner,
            "slug": slug,
            "title": "SEO Detail Gist",
            "description": "Gist description for SEO tests.",
            "source_code": "print('seo')",
            "language": "python",
            "stars": 0,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return slug, uid
def _seed_project():
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    uid = str(uuid4())
    slug = make_combined_slug("SEO Detail Project", uid)
    get_table("projects").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": owner,
            "slug": slug,
            "title": "SEO Detail Project",
            "description": "Project description for SEO tests.",
            "project_type": "software",
            "platforms": "Linux",
            "status": "Released",
            "stars": 0,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return slug, uid
def _seed_news_image(news_uid):
    from devplacepy.database import get_table

    get_table("news_images").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "news_uid": news_uid,
            "url": "https://example.com/seo-news-image.jpg",
        }
    )
def _seed_feed_posts(count):
    from datetime import datetime, timezone, timedelta
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    topic = f"seopag{owner[:8]}"
    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
    posts = get_table("posts")
    for i in range(count):
        uid = str(uuid4())
        posts.insert(
            {
                "deleted_at": None,
                "deleted_by": None,
                "uid": uid,
                "user_uid": owner,
                "slug": make_combined_slug(f"seo pag {i}", uid),
                "title": None,
                "content": f"seo pag post {i}",
                "topic": topic,
                "project_uid": None,
                "image": None,
                "stars": 0,
                "created_at": (base - timedelta(seconds=i)).isoformat(),
            }
        )
    return topic


def test_send_message_appears_in_thread(alice):
    page, _ = alice
    bob = get_table("users").find_one(username="bob_test")
    page.goto(
        f"{BASE_URL}/messages?with_uid={bob['uid']}", wait_until="domcontentloaded"
    )
    msg = f"Hello bob {int(time.time() * 1000)}"
    page.fill("textarea[name='content']", msg)
    page.locator(".messages-send-btn").click()
    page.wait_for_url("**/messages**", wait_until="domcontentloaded")
    page.locator(f".message-bubble:has-text('{msg}')").first.wait_for(state="visible")


def test_paste_image_attaches_in_chat(alice):
    page, _ = alice
    bob = get_table("users").find_one(username="bob_test")
    page.goto(
        f"{BASE_URL}/messages?with_uid={bob['uid']}", wait_until="domcontentloaded"
    )
    page.locator(".messages-input-area dp-upload .dp-upload-btn").first.wait_for(
        state="visible", timeout=10000
    )
    paste_image(page, ".messages-input-area textarea[name='content']")
    page.locator(".messages-input-area dp-upload .dp-upload-count").first.wait_for(
        state="visible", timeout=15000
    )
    expect(
        page.locator(".messages-input-area dp-upload input[name='attachment_uids']")
    ).to_have_value(re.compile(r".+"))


def test_messages_page_loads(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    assert page.is_visible(".messages-layout")


def test_chat_presence_is_driven_by_the_shared_roster(alice, bob):
    bob_page, _ = bob
    bob_page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
    bob_uid = get_table("users").find_one(username="bob_test")["uid"]
    page, _ = alice
    page.goto(
        f"{BASE_URL}/messages?with_uid={bob_uid}", wait_until="domcontentloaded"
    )
    presence = page.locator("#messages-presence")
    expect(presence).to_be_visible()
    expect(presence).to_have_class(re.compile(r"\bonline\b"))
    # the chat header and the feed roster read the same subscribed uid, never a
    # second client-side clock over a frozen last_seen
    assert presence.get_attribute("data-presence-uid") == bob_uid
    dots = page.locator(f".presence-dot[data-presence-uid='{bob_uid}']")
    assert dots.count() >= 1
    expect(dots.first).to_have_class(re.compile(r"\bonline\b"))


def test_messages_search_input(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    search = page.locator("#message-search")
    assert search.is_visible()
    assert search.get_attribute("placeholder") == "Search conversations..."


def test_messages_empty_state(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    assert page.is_visible(".messages-layout") or page.is_visible(
        "text=No conversations yet"
    )


def test_messages_search_for_bob(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    page.fill("#message-search", "bob_test")
    result = page.locator(".search-dropdown-item:has-text('bob_test')")
    result.wait_for(state="visible", timeout=5000)
    assert result.is_visible()


def test_messages_search_first_result_highlighted(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    page.fill("#message-search", "bob_test")
    result = page.locator(".search-dropdown-item:has-text('bob_test')")
    result.wait_for(state="visible", timeout=5000)
    expect(result.first).to_have_class(re.compile(r"\bactive\b"))


def test_messages_search_enter_opens_conversation(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    page.fill("#message-search", "bob_test")
    page.locator(".search-dropdown-item:has-text('bob_test')").wait_for(
        state="visible", timeout=5000
    )
    page.locator("#message-search").press("Enter")
    page.wait_for_url(re.compile(r"with_uid="), wait_until="domcontentloaded")


def test_messages_search_arrow_then_enter_opens_conversation(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    page.fill("#message-search", "bob_test")
    page.locator(".search-dropdown-item:has-text('bob_test')").wait_for(
        state="visible", timeout=5000
    )
    search = page.locator("#message-search")
    search.press("ArrowDown")
    search.press("Enter")
    page.wait_for_url(re.compile(r"with_uid="), wait_until="domcontentloaded")


def test_messages_search_escape_closes_dropdown(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    page.fill("#message-search", "bob_test")
    result = page.locator(".search-dropdown-item:has-text('bob_test')")
    result.wait_for(state="visible", timeout=5000)
    page.locator("#message-search").press("Escape")
    expect(result).to_be_hidden()


def test_messages_header_visible(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    assert page.is_visible("text=Home")
    assert page.is_visible("text=Projects")


def test_messages_bell_icon(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    bell = page.locator(".topnav-icon[href='/notifications']")
    assert bell.is_visible()


def test_messages_topnav_user(alice):
    page, user = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    assert page.is_visible(f"text={user['username']}")


def test_messages_avatar_on_page(alice):
    page, user = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    avatar = page.locator("img.avatar-img").first
    assert avatar.is_visible()


def test_messages_page_is_noindex(alice):
    page, user = alice
    page.goto(f"{BASE_URL}/messages", wait_until="domcontentloaded")
    page.wait_for_url("**/messages", wait_until="domcontentloaded")
    meta = page.locator('meta[name="robots"]')
    content = meta.get_attribute("content")
    assert content and "noindex" in content


def test_optimistic_send_pending_then_reconciles_no_double_render(alice, bob):
    page_a, user_a = alice
    page_b, user_b = bob
    uid_a = get_table("users").find_one(username=user_a["username"])["uid"]
    uid_b = get_table("users").find_one(username=user_b["username"])["uid"]

    page_a.goto(
        f"{BASE_URL}/messages?with_uid={uid_b}", wait_until="domcontentloaded"
    )
    page_b.goto(
        f"{BASE_URL}/messages?with_uid={uid_a}", wait_until="domcontentloaded"
    )

    msg = f"Optimistic hello {int(time.time() * 1000)}"
    textarea = page_a.locator(".messages-input-area textarea[name='content']")
    textarea.wait_for(state="visible")
    textarea.fill(msg)
    page_a.locator(".messages-send-btn").click()

    bubble = page_a.locator(f".message-bubble.mine:has-text('{msg}')").first
    bubble.wait_for(state="visible", timeout=10000)
    client_id = bubble.get_attribute("data-client-id")
    assert client_id

    reconciled = page_a.locator(
        f".message-bubble.mine[data-client-id='{client_id}'][data-msg-uid]:not(.pending)"
    )
    reconciled.wait_for(state="visible", timeout=10000)
    assert page_a.locator(f".message-bubble.mine:has-text('{msg}')").count() == 1

    received = page_b.locator(f".message-bubble.theirs:has-text('{msg}')")
    received.wait_for(state="visible", timeout=10000)
    assert received.count() == 1


def test_mobile_single_pane_back_button(mobile_page):
    page, user = mobile_page
    bob = get_table("users").find_one(username="bob_test")
    page.goto(
        f"{BASE_URL}/messages?with_uid={bob['uid']}", wait_until="domcontentloaded"
    )

    thread = page.locator(".messages-main")
    conv_list = page.locator(".messages-list")
    thread.wait_for(state="visible")
    expect(conv_list).to_be_hidden()

    page.locator("#messages-back-btn").click()

    expect(conv_list).to_be_visible()
    expect(thread).to_be_hidden()
    assert page.url == f"{BASE_URL}/messages"