path: root/packages/meshbay-hub/tests/test_transport_contracts.py

"""
Two wire contracts in transport.js that fail quietly when broken.

Neither can be reached from Python, and `QE/deploy/e2e.py` is a second
implementation of the client rather than a test of this one, so these read the
source. That is weak evidence in general, and the right kind here: both defects
below produce a plausible screen rather than an error.

  - History paged forward from the oldest message, so a group with more than
    200 of them opened on its first screen and the recent conversation could
    not be reached. Nothing threw; the wrong messages were simply shown.
  - A reply that is not routed falls through to "resolve the oldest pending
    request". Adding a ping made that dangerous: a pong handed to a waiting
    history request satisfies it with a message that has no `messages` field,
    and the conversation renders empty.
"""

import re
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
TRANSPORT = STATIC / "transport.js"
APP = STATIC / "app.js"

pytestmark = pytest.mark.skipif(
    not TRANSPORT.exists(), reason="the SPA sources are not available")


@pytest.fixture(scope="module")
def transport():
    return TRANSPORT.read_text()


@pytest.fixture(scope="module")
def app():
    return APP.read_text()


def test_chat_history_pages_backwards(transport):
    body = transport[transport.index("async fetchChatHistory"):]
    body = body[:body.index("\n  }")]
    assert "before" in body, "the request must carry a backward cursor"
    assert "since" not in body, (
        "`since` pages forward from the oldest message — that was the bug")


def test_chat_history_reports_whether_more_exists(transport):
    body = transport[transport.index("async fetchChatHistory"):]
    body = body[:body.index("\n  }")]
    assert "has_more" in body, (
        "without it the 'load older' control cannot know when to stop offering")


def test_the_browser_asks_for_the_newest_page_first(app):
    """A group opens on the newest messages, not the oldest."""
    assert "fetchChatHistory({ limit: CHAT_PAGE })" in app
    assert re.search(r"const CHAT_PAGE\s*=\s*100", app)
    assert re.search(r"const CHAT_OLDER_PAGE\s*=\s*50", app)


def test_older_pages_are_requested_with_a_cursor_not_an_offset(app):
    assert "before: messages[0].id" in app, (
        "paging by offset repeats or skips messages when one arrives mid-scroll")


def test_pong_is_routed_by_its_echoed_token(transport):
    """Not left to the oldest-pending fallback, which would empty a chat."""
    assert "if (msg.type === 'pong')" in transport
    routing = transport[transport.index("if (msg.type === 'pong')"):]
    routing = routing[:routing.index("const oldest")]
    assert "handler._key === key" in routing
    assert "return;" in routing, (
        "a pong for a timed-out probe must stop here, not fall through")


def test_ping_requests_are_keyed(transport):
    assert "`ping:${obj.token}`" in transport, (
        "an unkeyed ping cannot be matched to its pong")


def test_ping_can_time_out_sooner_than_a_transfer(transport):
    """30 s is right for a chunk and useless for a liveness probe."""
    assert "_sendAndWait(obj, timeoutMs = 30000)" in transport
    body = transport[transport.index("  async ping("):]
    body = body[:body.index("\n  }")]
    assert "timeoutMs" in body


def test_scroll_position_is_anchored_when_older_messages_are_prepended(app):
    """Everything above the viewport grows, so scrollTop alone is not enough."""
    assert "scrollHeight - list.scrollTop" in app, "the anchor is measured from the bottom"
    assert "list.scrollTop = list.scrollHeight - anchorRef.current" in app
    assert "useLayoutEffect" in app, (
        "correcting after paint shows the jump it is meant to prevent")


def test_the_view_only_follows_new_messages_when_already_at_the_bottom(app):
    assert "if (atBottomRef.current) list.scrollTop = list.scrollHeight" in app, (
        "scrolling unconditionally fights someone reading back through history")


def test_the_bottom_is_reached_by_scrollTop_not_a_sentinel(app):
    """scrollIntoView on a zero-height marker stops short of the true bottom.

    The list has padding and a flex gap below the last bubble, so aligning an
    empty div to the viewport bottom left the bar a few pixels from the end —
    visible on opening a group and again after sending a message.
    """
    # The call, not the word: a comment explaining why it is gone should not
    # be able to fail this.
    assert ".scrollIntoView(" not in app
    assert "list.scrollTo({ top: list.scrollHeight" in app, (
        "jumping to the latest should also land on the real bottom")


def test_messages_are_keyed_by_id_not_index(app):
    """Index keys plus prepending makes Preact reuse the wrong bubbles."""
    assert "key=${m.id}" in app
    assert "key=${i}" not in app.split("function ChatPanel")[1].split("\n}")[0]


def test_presence_has_three_states_and_a_label_for_each(app):
    for state in ("online", "offline", "unknown"):
        assert f"presence-{state}" in (STATIC / "style.css").read_text()
    assert "t('presence.' + state)" in app, (
        "red and green are the pair colour-blind readers cannot separate, so "
        "the dot needs a title and an aria-label, not just a colour")


def _string(source: str, key: str) -> str:
    """One locale entry's text, whether it is written on one line or spliced
    across several with `+`."""
    start = source.index(f"'{key}':") + len(f"'{key}':")
    end = source.index("\n  '", start)
    return source[start:end]


def test_a_refusal_from_the_node_counts_as_present(app):
    """The node answering "no" proves it is up; only silence proves nothing."""
    assert "err.reason ? 'online' : 'offline'" in app


# ── The create-group form ─────────────────────────────────────────────────────

def test_the_form_asks_one_question_not_two(app):
    """
    Visibility and admission were separate selectors that could only ever be set
    together, and the form knew it — picking Public reached over and set the
    policy. Two of the four combinations were impossible: the API refused
    public+invite with a 422, and private+open is a directory listing nobody can
    find, joining being through the node rather than a link.

    So there is one selector. "Open" is what makes a group listed, and the
    request derives the rest.
    """
    form = app[app.index("function CreateGroupPage"):]
    form = form[:form.index("\n}\n")]

    assert "setVisibility(" not in form, "the visibility selector is back"
    assert "t('create_group.join_policy')" in form
    assert "joinPolicy === 'open' ? 'public' : 'private'" in form, (
        "the request must derive visibility rather than leave it unset")


def test_the_form_says_what_each_choice_means_for_finding_the_group(app):
    """Dropping the visibility box removes the words "public" and "private"
    from the page. If the descriptions do not say it, nothing does — and
    somebody publishes a group without meaning to."""
    en = (STATIC / "locales" / "en.js").read_text(encoding="utf-8")
    invite = _string(en, "create_group.invite_desc")
    open_ = _string(en, "create_group.open_desc")
    assert "not listed" in invite.lower()
    assert "listed" in open_.lower() and "anyone" in open_.lower()


def test_the_strings_the_visibility_box_used_are_gone(app):
    """A key nobody reads is a key that rots, and ten locales carry each one."""
    for locale in (STATIC / "locales").glob("*.js"):
        text = locale.read_text(encoding="utf-8")
        for key in ("create_group.visibility", "create_group.private",
                    "create_group.public_is_open", "create_group.public_desc"):
            assert f"'{key}'" not in text, f"{locale.name} still carries {key}"


def test_the_form_starts_on_a_combination_the_api_accepts(app):
    form = app[app.index("function CreateGroupPage"):]
    assert "useState('invite')" in form[:form.index("return html")]


# ── Dead references in the SPA ────────────────────────────────────────────────

def test_no_setter_survives_the_state_it_belonged_to(app):
    """A removed useState leaves its setter behind, and nothing complains.

    `setActionsOpen` outlived `actionsOpen` when the Actions dropdown became a
    row of buttons, and shipped: every action in the Files panel threw
    ReferenceError on click. No Python test could see it and e2e.py does not
    drive the SPA, so the shape is checked here instead.

    A grep for the state name does not find it — `setActionsOpen` does not
    contain `actionsOpen`, the capital breaks the match. That is exactly how it
    got through.
    """
    import re
    declared = set(re.findall(r"const \[\s*\w+\s*,\s*(set\w+)\s*\]\s*=\s*useState", app))
    # Names brought in from another module are defined, just not here.
    imported = set()
    for names in re.findall(r"import\s*\{([^}]*)\}\s*from", app):
        imported.update(n.strip().split(" as ")[-1].strip() for n in names.split(","))
    # A bare call only: `downloads.setMode(...)` and `view.setUint32(...)` belong
    # to their object, not to this component.
    called = set(re.findall(r"(?<![.\w])(set[A-Z]\w*)\s*\(", app))
    builtin = {"setTimeout", "setInterval"}
    # A `setX` that is a plain function of this module is not an orphan setter:
    # `setAuth` writes the session to localStorage and has no `useState` behind
    # it by design. Without this the rule reports every such helper, and a rule
    # that cries wolf is one someone eventually silences.
    defined = set(re.findall(r"^(?:async\s+)?function\s+(set[A-Z]\w*)\s*\(", app, re.M))
    defined |= set(re.findall(r"^\s*const\s+(set[A-Z]\w*)\s*=", app, re.M))

    orphans = sorted(called - declared - imported - builtin - defined)
    assert not orphans, (
        f"setter(s) called with no useState behind them: {orphans} — "
        "each one is a ReferenceError the moment that code path runs")


# ── Parallel uploads ──────────────────────────────────────────────────────────

def test_an_upload_refusal_names_the_file_it_is_about(transport):
    """Reported 2026-08-16: a second upload started in parallel killed both.

    An error used to carry no filename, so the client could not tell whose it
    was and failed every upload in flight — one name the node disliked took the
    other file with it. The node names the file now, and only that upload stops.
    """
    body = transport[transport.index("if (msg.type === 'error' && this._uploaders.size)"):]
    body = body[:body.index("\n    if (msg.type === 'chat_msg'")]
    assert "this._uploaders.has(msg.filename)" in body, (
        "a named refusal must reach one uploader, not all of them")
    assert "if (!msg.filename)" in body, (
        "an unnamed error from an older node must still stop everything — "
        "guessing which upload it belongs to would be worse")


def test_uploads_are_tracked_per_file(transport):
    """Acks interleave when two files are in flight."""
    assert "this._uploaders = new Map()" in transport
    assert "this._uploaders.set(file.name" in transport