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

"""
The indexing dock: what it shows, and who feeds it.

Adding a 900 GB directory left the operator nothing to look at once they left
the Settings panel — and nothing at all when the directory was added from
another machine, where that panel's bar never started. The dock sits above the
music bar on every page. What it shows is worked out in `index-dock-model.js`,
which has no imports so that it runs here under node as shipped; the wiring
around it is checked in the source, as the other SPA tests do.
"""

import json
import re
import shutil
import subprocess
from pathlib import Path

import pytest
from spa_source import transport_source

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
MODEL = STATIC / "index-dock-model.js"
DOCK = STATIC / "index-dock.js"
APP = STATIC / "app.js"
GROUP_PAGE = STATIC / "group-page.js"
TRANSPORT = STATIC / "transport.js"

pytestmark = pytest.mark.skipif(not MODEL.exists(), reason="SPA sources unavailable")

GB = 1024 ** 3

needs_node = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")


def _run(tmp_path, body: str):
    (tmp_path / "package.json").write_text('{"type":"module"}', encoding="utf-8")
    (tmp_path / "model.js").write_text(MODEL.read_text(encoding="utf-8"), encoding="utf-8")
    script = tmp_path / "case.js"
    script.write_text("import * as m from './model.js';\n" + body, encoding="utf-8")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True, encoding="utf-8",
                          cwd=str(tmp_path))
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def _job(**over) -> dict:
    job = {"groupId": "g1", "groupName": "outputs", "scanning": True, "kind": "scan",
           "root": "results", "scannedBytes": 0, "totalBytes": 0, "filesDone": 0,
           "filesTotal": 0, "queued": []}
    job.update(over)
    return job


def _frames(tmp_path, frames: list[tuple[int, list[dict]]], hide_at: int | None = None,
            pause_at: int | None = None):
    """Feed dockRows a sequence of (time ms, jobs) and return each frame's rows."""
    return _run(tmp_path, f"""
        const frames = {json.dumps(frames)};
        const hideAt = {json.dumps(hide_at)};
        const pauseAt = {json.dumps(pause_at)};
        let memory = {{}};
        const out = [];
        for (const [now, jobs] of frames) {{
          if (hideAt === now) memory = m.hideRow(memory, 'g1');
          if (pauseAt === now) memory = m.pauseRow(memory, 'g1');
          const r = m.dockRows(jobs, memory, now);
          memory = r.memory;
          out.push(r.rows.map((row) => [row.groupId, row.state]));
        }}
        console.log(JSON.stringify(out));
    """)


# ── What is shown ───────────────────────────────────────────────────────────

@needs_node
def test_a_root_scan_shows_at_once(tmp_path):
    assert _frames(tmp_path, [(0, [_job()])]) == [[["g1", "running"]]]


@needs_node
def test_a_small_burst_waits_five_seconds_and_a_large_one_does_not(tmp_path):
    small = _job(kind="watch", root="", totalBytes=10_000)
    assert _frames(tmp_path, [(0, [small]), (4000, [small]), (5000, [small])]) == [
        [], [], [["g1", "running"]]]
    large = _job(kind="reconcile", totalBytes=2 * GB)
    assert _frames(tmp_path, [(0, [large])]) == [[["g1", "running"]]]


@needs_node
def test_a_short_burst_never_flashes_a_finished_row(tmp_path):
    small = _job(kind="watch", root="", totalBytes=10_000)
    idle = _job(scanning=False, kind="")
    assert _frames(tmp_path, [(0, [small]), (1000, [idle])]) == [[], []]


@needs_node
def test_a_group_between_two_roots_stays_on_screen(tmp_path):
    """The gap between one root's walk ending and the next taking the lock."""
    waiting = _job(scanning=False, kind="", root="", queued=["archive"])
    assert _frames(tmp_path, [(0, [_job()]), (2000, [waiting])]) == [
        [["g1", "running"]], [["g1", "running"]]]


@needs_node
def test_finished_says_so_for_four_seconds_then_goes(tmp_path):
    idle = _job(scanning=False, kind="", root="")
    assert _frames(tmp_path, [(0, [_job()]), (1000, [idle]), (4999, [idle]),
                              (5000, [idle])]) == [
        [["g1", "running"]], [["g1", "done"]], [["g1", "done"]], []]


@needs_node
def test_a_hidden_row_stays_hidden_until_the_group_is_idle(tmp_path):
    second = _job(root="archive")
    idle = _job(scanning=False, kind="", root="")
    frames = [(0, [_job()]), (1000, [_job()]), (2000, [second]), (3000, [idle]),
              (9000, [idle]), (10000, [_job(root="photos")])]
    assert _frames(tmp_path, frames, hide_at=1000) == [
        [["g1", "running"]], [], [], [], [], [["g1", "running"]]]


# ── Where it comes from ─────────────────────────────────────────────────────

@needs_node
def test_a_paused_walk_leaves_at_once_and_is_not_called_finished(tmp_path):
    """The node takes a moment to stop: its next reports still say scanning."""
    idle = _job(scanning=False, kind="")
    assert _frames(tmp_path, [
        (0, [_job()]), (1000, [_job()]), (2000, [_job()]), (3000, [idle]),
    ], pause_at=1000) == [[["g1", "running"]], [], [], []]


@needs_node
def test_pausing_one_root_does_not_hide_the_next_one_queued(tmp_path):
    first = _job(root="results", queued=["archive"])
    second = _job(root="archive")
    assert _frames(tmp_path, [
        (0, [first]), (1000, [first]), (2000, [second]),
    ], pause_at=1000) == [[["g1", "running"]], [], [["g1", "running"]]]


@needs_node
def test_the_loopback_answer_wins_and_the_hub_names_the_group(tmp_path):
    out = _run(tmp_path, """
        const local = { g1: m.fromLoopback({ group_id: 'g1', group_name: 'node name',
          scanning: true, kind: 'scan', root: 'results', queued: ['archive'] }) };
        const pushed = {
          g1: m.fromPush('g1', { scanning: true, kind: 'rescan', root_pos: 0 }, [{ name: 'x' }]),
          g2: m.fromPush('g2', { scanning: true }, []),
        };
        console.log(JSON.stringify(
          m.mergeActivity(local, pushed, [{ id: 'g1', name: 'outputs' }])));
    """)
    by_id = {j["groupId"]: j for j in out}
    assert (by_id["g1"]["kind"], by_id["g1"]["root"], by_id["g1"]["groupName"]) == (
        "scan", "results", "outputs")
    assert by_id["g1"]["queued"] == ["archive"]
    assert by_id["g2"]["groupName"] == "g2"


@needs_node
def test_a_push_is_named_from_the_roots_table_it_does_not_carry(tmp_path):
    out = _run(tmp_path, """
        const roots = [{ name: 'outputs' }, { name: 'results' }];
        console.log(JSON.stringify([
          m.fromPush('g', { scanning: true, kind: 'scan', root_pos: 1, queued: 2 }, roots),
          m.fromPush('g', { scanning: true, root_pos: 7 }, roots),
          m.fromPush('g', { scanning: true, scanned_bytes: 5, total_bytes: 10 }, null),
        ]));
    """)
    assert (out[0]["root"], out[0]["queued"]) == ("results", 2)
    assert out[1]["root"] == "", "a position past the table must name nothing"
    assert (out[2]["kind"], out[2]["root"], out[2]["queued"]) == ("", "", 0), (
        "a node older than the dock must read as a plain scan")


# ── Wiring ──────────────────────────────────────────────────────────────────

def test_the_dock_is_on_every_page_not_in_a_route():
    app = APP.read_text(encoding="utf-8")
    mount = app.index("<${IndexingDock}")
    assert app.index("</main>") < mount < app.index("<${MusicPlayerBar}"), (
        "the dock must sit outside the routed page, right before the music bar")


def test_only_an_operator_connection_feeds_the_dock():
    page = GROUP_PAGE.read_text(encoding="utf-8")
    calls = [m.start() for m in re.finditer(r"reportIndexPush\(groupId, (?!null)", page)]
    assert len(calls) == 2, "the push handler and the handshake ack"
    for at in calls:
        guard = page[max(0, at - 200):at]
        assert "transport.memberRole === 'operator'" in guard, (
            "a member's connection must keep the sidebar dot only")


def test_the_pause_button_reaches_the_node_by_either_door():
    """The local node over loopback; any other through the page's connection,
    signed by its operator — the dock holds no transport of its own."""
    dock = DOCK.read_text(encoding="utf-8")
    assert "node.op('pauseRoot'" in dock
    page = GROUP_PAGE.read_text(encoding="utf-8")
    assert "transport.pauseRoot(groupId, rootName, transport.signFn)" in page
    for at in [m.start() for m in re.finditer(r"reportIndexPush\(groupId, (?!null)", page)]:
        assert "pauseFromDock" in page[at:at + 140]


def test_leaving_a_group_clears_its_row():
    page = GROUP_PAGE.read_text(encoding="utf-8")
    cleanup = page[page.index("    return () => {\n      cancelled = true;"):]
    cleanup = cleanup[:cleanup.index("};")]
    assert "reportIndexPush(groupId, null)" in cleanup


def test_the_transport_passes_the_new_counters_through():
    source = transport_source()
    block = source[source.index("msg.type === 'index_progress'"):]
    block = block[:block.index("return;")]
    for field in ("files_done", "files_total", "kind", "root_pos", "queued"):
        assert f"{field}:" in block, f"{field} is dropped before it reaches the page"


def test_the_dock_polls_the_node_wide_route_not_one_group():
    source = DOCK.read_text(encoding="utf-8")
    assert "node.op('indexStatus')" in source
    assert "groupIndexStatus" not in source


# ── A disk that stops answering ─────────────────────────────────────────────

@needs_node
def test_a_silent_disk_is_named_by_the_loopback_answer_only(tmp_path):
    """A scan that sat still showed a bar that had stopped and a folder's name.
    What it is waiting on is a path: the node's own machine, never a push."""
    paths = ["C:\\Music\\Some Album\\track 02.flac", "/srv/music/some album/track 02.flac", ""]
    out = _run(tmp_path, f"""
        const paths = {json.dumps(paths)};
        console.log(JSON.stringify({{
          local: m.fromLoopback({{ group_id: 'g', scanning: true, slow_path: paths[0],
                                   slow_secs: 23 }}),
          idle: m.fromLoopback({{ group_id: 'g', scanning: true }}),
          push: m.fromPush('g', {{ scanning: true, slow_path: paths[0], slow_secs: 23 }}, []),
          names: paths.map(m.baseName),
        }}));
    """)
    assert (out["local"]["slowPath"], out["local"]["slowSecs"]) == (paths[0], 23)
    assert (out["idle"]["slowPath"], out["idle"]["slowSecs"]) == ("", 0)
    assert (out["push"]["slowPath"], out["push"]["slowSecs"]) == ("", 0), (
        "a member's connection must not be able to put a path on screen")
    assert out["names"] == ["track 02.flac", "track 02.flac", ""], "either separator"


def test_the_row_says_what_the_disk_is_silent_on():
    dock = DOCK.read_text(encoding="utf-8")
    row = dock[dock.index("function DockRow("):dock.index("export function IndexingDock(")]
    assert ("t('indexdock.slow', { secs: number(row.slowSecs), "
            "file: baseName(row.slowPath) })") in row
    # Before the percentage: a bar that no longer moves is not the news.
    assert row.index("else if (slow)") < row.index("else if (listing)")
    assert "title=${slow ? row.slowPath : null}" in row, "the whole path, on hover"


def test_group_settings_list_what_the_node_could_not_read():
    """With paths, so asked of the loopback API and of nothing else: an operator
    elsewhere sees the folder paused, as any member does."""
    settings = (STATIC / "group-settings.js").read_text(encoding="utf-8")
    effect = settings[settings.index("if (!onNodeMachine) { setTrouble(null)"):]
    effect = effect[:effect.index("}, [onNodeMachine, groupId, troubleTick]);")]
    assert "platform.node.op('groupIndexStatus', { groupId })" in effect
    assert "transport." not in effect

    note = settings[settings.index("const troubleNote ="):]
    note = note[:note.index("if (!displayRoots.length)")]
    assert "t('node.root_stalled_hint', { name, file })" in note
    assert "t('node.unread_summary', { count: trouble.total })" in note
    assert "u.reason === 'no_answer'" in note and "t('node.unread_no_answer')" in note
    # Only for a folder that is still paused: resumed, the note has had its day.
    assert "r.name === s.root && r.paused" in settings
    assert "${troubleNote}" in settings[settings.index('<table class="shared-dirs-tbl">'):]

    css = (STATIC / "style.css").read_text(encoding="utf-8")
    rule = css[css.index(".unread-files li {"):]
    assert "word-break: break-word" in rule[:rule.index("}")], (
        "a path has no spaces to break at, and unpins the page on a phone")