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"
# Chat's history paging/scroll-anchoring logic and GroupPage's own connect()
# effect moved out of app.js in the group-page refactor.
CHAT_APP = STATIC / "chat-app.js"
GROUP_PAGE = STATIC / "group-page.js"
CREATE_GROUP = STATIC / "create-group-page.js"
SPLIT_FILES = [APP, GROUP_PAGE, CHAT_APP, STATIC / "files-app.js",
              STATIC / "video-player.js", STATIC / "video-app.js",
              STATIC / "music-app.js", STATIC / "music-player.js",
              STATIC / "photos-app.js",
              STATIC / "group-settings.js",
              STATIC / "auth-page.js", STATIC / "explore-page.js",
              CREATE_GROUP]

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()


@pytest.fixture(scope="module")
def chat():
    return CHAT_APP.read_text()


@pytest.fixture(scope="module")
def group_page():
    return GROUP_PAGE.read_text()


@pytest.fixture(scope="module")
def create_group():
    return CREATE_GROUP.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(chat):
    """A group opens on the newest messages, not the oldest."""
    assert "fetchChatHistory({ limit: CHAT_PAGE })" in chat
    assert re.search(r"const CHAT_PAGE\s*=\s*100", chat)
    assert re.search(r"const CHAT_OLDER_PAGE\s*=\s*50", chat)


def test_older_pages_are_requested_with_a_cursor_not_an_offset(chat):
    assert "before: messages[0].id" in chat, (
        "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(chat):
    """Everything above the viewport grows, so scrollTop alone is not enough."""
    assert "scrollHeight - list.scrollTop" in chat, "the anchor is measured from the bottom"
    assert "list.scrollTop = list.scrollHeight - anchorRef.current" in chat
    assert "useLayoutEffect" in chat, (
        "correcting after paint shows the jump it is meant to prevent")


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


def test_every_authorize_admin_op_call_is_registered_in_admin_op_types(transport):
    """
    Found live (docs/photos.md's photo_roots): `setPhotoRoots` called
    `_authorizeAdminOp(msg, 'photo_roots', ...)` like every other admin op,
    but `photo_roots` was never added to `ADMIN_OP_TYPES` — so its initial
    request was never keyed `admin:photo_roots`, the node's `admin_challenge`
    reply matched no pending request (_dispatch's own keyed block, which
    `return`s unconditionally whether or not it found a match), and the
    request sat until its 30 s timeout with no error and no admin prompt.
    `ADMIN_OP_TYPES`'s own comment already narrates this exact bug once,
    for `audio_root`/`apps_enabled` — this pins it so a third op cannot
    reintroduce it silently.
    """
    set_body = transport[transport.index("const ADMIN_OP_TYPES = new Set(["):]
    set_body = set_body[:set_body.index("]);")]
    registered = set(re.findall(r"'([a-z_]+)'", set_body))

    called = set(re.findall(r"_authorizeAdminOp\(\s*\w+,\s*'([a-z_]+)'", transport))
    assert called, "the extraction pattern itself found nothing — check it against transport.js"
    missing = called - registered
    assert not missing, (
        f"{sorted(missing)} call _authorizeAdminOp but are missing from ADMIN_OP_TYPES — "
        "their admin_challenge will silently time out instead of ever reaching the user")


def test_the_bottom_is_reached_by_scrollTop_not_a_sentinel(chat):
    """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 chat
    assert "list.scrollTo({ top: list.scrollHeight" in chat, (
        "jumping to the latest should also land on the real bottom")


def test_messages_are_keyed_by_id_not_index(chat):
    """Index keys plus prepending makes Preact reuse the wrong bubbles."""
    assert "key=${m.id}" in chat
    assert "key=${i}" not in chat.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(group_page):
    """The node answering "no" proves it is up; only silence proves nothing."""
    assert "err.reason ? 'online' : 'offline'" in group_page


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

def test_the_form_asks_one_question_not_two(create_group):
    """
    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 = create_group[create_group.index("function CreateGroupFormSimple"):]
    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(create_group):
    form = create_group[create_group.index("function CreateGroupFormSimple"):]
    assert "useState('invite')" in form[:form.index("return html")]


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

def test_no_setter_survives_the_state_it_belonged_to():
    """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.

    Checked per file rather than on one concatenated blob: the group-page
    refactor split what used to be one app.js into several, and a setter
    defined in one (e.g. `setAuth`, imported from hub-client.js) must not be
    mistaken for covering an orphan call of the same name in another. Lifting
    state to the shared shell and passing its setter down as a prop is the
    same idea one level lower — `FilesPanel`'s `setEntries` is real, just
    declared in group-page.js's own `useState` rather than here — so a setter
    named in a component's own destructured props is treated as defined too.
    """
    import re
    for path in SPLIT_FILES:
        app = path.read_text()
        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))
        # A setter named in a `function Component({ ..., setX, ... })` prop
        # list is handed down from wherever it is really declared.
        for params in re.findall(r"^function [A-Z]\w*\(\{([^}]*)\}", app, re.M):
            defined.update(re.findall(r"\b(set[A-Z]\w*)\b", params))

        orphans = sorted(called - declared - imported - builtin - defined)
        assert not orphans, (
            f"{path.name}: setter(s) called with no useState behind them: "
            f"{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