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

"""
The transfer store, which is what keeps a download alive after you leave a group.

Run under Node, because the behaviour worth pinning is timing and lifetime:
that a cancel actually stops the work rather than only greying out a row, that a
stalled transfer reads as slow instead of reporting its historical average, and
that a transport handed over by a departing page is closed by the last transfer
using it — not before, and not never.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
TRANSFERS = STATIC / "transfers.js"

pytestmark = pytest.mark.skipif(
    shutil.which("node") is None or not TRANSFERS.exists(),
    reason="node or the SPA sources are not available")


def _run(body, tmp_path):
    module = tmp_path / "transfers.mjs"
    module.write_text(TRANSFERS.read_text())
    script = tmp_path / "case.mjs"
    script.write_text(
        f"import {{ TransferStore, formatSpeed }} from '{module.as_posix()}';\n"
        "const out = [];\n"
        "const say = (...a) => out.push(...a);\n"
        f"{body}\n"
        "console.log(JSON.stringify(out));\n")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_a_cancel_actually_stops_the_work(tmp_path):
    """
    The flag has to be read by the thing doing the work. A store that only
    marks a row cancelled gives you a button that lies.
    """
    result = _run("""
const store = new TransferStore();
let chunksWritten = 0;
const id = store.start({
  kind: 'download', name: 'big.mp4', total: 100,
  run: async ({ signal, onProgress }) => {
    for (let i = 0; i < 100; i++) {
      if (signal.aborted) throw Object.assign(new Error('stop'), { name: 'AbortError' });
      chunksWritten++;
      onProgress(i + 1, 100);
      await new Promise(r => setTimeout(r, 1));
    }
  },
});
await new Promise(r => setTimeout(r, 20));
store.cancel(id);
await new Promise(r => setTimeout(r, 40));
say(chunksWritten < 100, store.list()[0].status, store.active);
""", tmp_path)
    assert result[0] is True, "the work ran to completion despite being cancelled"
    assert result[1] == "cancelled"
    assert result[2] == 0


def test_speed_is_measured_over_a_window_not_since_the_start(tmp_path):
    """A transfer that stalls should read as slow now, not as its own average."""
    result = _run("""
let clock = 0;
const store = new TransferStore(() => clock);
let report;
store.start({
  kind: 'download', name: 'x', total: 100_000_000,
  run: async ({ onProgress }) => { report = onProgress; await new Promise(() => {}); },
});
await new Promise(r => setTimeout(r, 1));   // start() defers run() by a tick
// A fast megabyte a second, for six seconds.
for (let i = 1; i <= 6; i++) { clock = i * 1000; report(i * 1_000_000, 100_000_000); }
say(Math.round(store.list()[0].speed));
// Then it stalls: the clock moves, the bytes do not.
for (let i = 7; i <= 12; i++) { clock = i * 1000; report(6_000_000, 100_000_000); }
say(Math.round(store.list()[0].speed));
""", tmp_path)
    assert 900_000 <= result[0] <= 1_100_000, f"expected ~1 MB/s, got {result[0]}"
    assert result[1] == 0, f"a stalled transfer still reports {result[1]} B/s"


def test_a_transport_is_closed_by_the_last_transfer_that_needed_it(tmp_path):
    """
    Leaving a group page must not kill a running download, and must not leak
    the connection either.
    """
    result = _run("""
const store = new TransferStore();
let closed = 0;
const transport = { close: () => { closed++; }, onIndexSync: () => {} };
let done1, done2;
store.start({ kind: 'download', name: 'a', transport,
  run: () => new Promise(r => { done1 = r; }) });
store.start({ kind: 'download', name: 'b', transport,
  run: () => new Promise(r => { done2 = r; }) });

store.releaseWhenIdle(transport);          // the page goes away
await new Promise(r => setTimeout(r, 5));
say(closed);                                // still working: must stay open

done1(); await new Promise(r => setTimeout(r, 5));
say(closed);                                // one left: still open

done2(); await new Promise(r => setTimeout(r, 5));
say(closed);                                // now it can go
""", tmp_path)
    assert result == [0, 0, 1], f"close() calls after each step: {result}"


def test_an_idle_transport_is_closed_straight_away(tmp_path):
    result = _run("""
const store = new TransferStore();
let closed = 0;
store.releaseWhenIdle({ close: () => { closed++; } });
say(closed);
""", tmp_path)
    assert result == [1], "leaving a group with nothing running should close it"


def test_signing_out_cancels_everything_and_lets_go(tmp_path):
    """
    Navigating is not signing out. The second one has to stop transfers running
    on tokens that are about to stop being ours.
    """
    result = _run("""
const store = new TransferStore();
let closed = 0;
const transport = { close: () => { closed++; } };
store.start({ kind: 'upload', name: 'a', transport,
  run: () => new Promise(() => {}) });
store.reset();
await new Promise(r => setTimeout(r, 5));
say(store.list().length, closed, store.active);
""", tmp_path)
    assert result == [0, 1, 0]


def test_a_failure_is_kept_and_named(tmp_path):
    """A transfer that dies silently is one the user retries at random."""
    result = _run("""
const store = new TransferStore();
store.start({ kind: 'download', name: 'gone.mp4',
  run: async () => { throw new Error('File not found'); } });
await new Promise(r => setTimeout(r, 5));
const row = store.list()[0];
say(row.status, row.error);
store.clearFinished();
say(store.list().length);
""", tmp_path)
    assert result[0] == "failed"
    assert result[1] == "File not found"
    assert result[2] == 0


def test_the_rate_is_readable(tmp_path):
    result = _run("""
say(formatSpeed(0), formatSpeed(2048), formatSpeed(5 * 1024 * 1024));
""", tmp_path)
    assert result == ["", "2 KB/s", "5.0 MB/s"]


# ── What the widget shows without being opened ──────────────────────────────

APP = STATIC / "app.js"


def _widget() -> str:
    source = APP.read_text(encoding="utf-8")
    start = source.index("function TransferWidget(")
    return source[start:source.index("\nfunction ", start + 1)]


def test_the_widget_marks_itself_while_transfers_run():
    """
    The badge counts them, but a count has to be read. Colour is what says
    "something is moving" from across the room, which is the point of a widget
    that lives in the nav bar rather than on the page.
    """
    widget = _widget()
    assert "running.length ? 'active' : ''" in widget, (
        "the button no longer marks itself while transfers are running")


def test_the_mark_comes_off_when_the_last_one_finishes():
    """`running` is derived from the live list on every render, not stored — so
    there is no state to forget to clear."""
    widget = _widget()
    assert "const running = items.filter(i => i.status === 'running');" in widget


def test_the_colour_is_defined_for_that_mark():
    """The class is set in one file and coloured in another; either alone does
    nothing, and neither fails loudly."""
    css = (STATIC / "style.css").read_text(encoding="utf-8")
    assert ".transfer-btn.active" in css


# ── The file list ───────────────────────────────────────────────────────────

def test_a_folder_name_carries_no_trailing_slash():
    """
    The folder icon in the cell beside it already says what it is.

    Anchored on `key=${full}`, which is the row that renders a directory. The
    first `dir-row` in the file is the ".." row added later, whose only cell is
    an ellipsis — slicing from there found no `${d}` and failed on a name it
    had never looked at.
    """
    source = STATIC.joinpath("files-app.js").read_text(encoding="utf-8")
    start = source.rindex("<tr", 0, source.index("key=${full}"))
    row = source[start:source.index("</tr>", start)]
    assert "dir-row" in row, "the anchor no longer lands on the directory row"
    assert "${d}/" not in row, "the folder name is rendered with a trailing slash"
    assert "${d}" in row


# ── Transfer slots, client side ─────────────────────────────────────────────
#
# A queue can lie in two directions, and both are worse than no queue: a
# transfer that shows "waiting" on a node that already granted it, and a slot
# the page holds after it has stopped using it. Everything below is one of
# those two.

def _lease_stub():
    """A Lease as the store sees it, driveable from the test."""
    return """
class L {
  constructor() {
    this.state = 'queued'; this.ahead = 2; this.closed = false;
    this.released = []; this.tr = 'tr1';
    this._wait = new Promise(r => { this._go = r; });
  }
  acquire() { return this._wait; }
  release(reason) { if (!this.closed) { this.closed = true; this.released.push(reason); } }
  grant() { this.state = 'granted'; if (this._onState) this._onState(this); this._go(); }
  push(state, ahead) { this.state = state; this.ahead = ahead; if (this._onState) this._onState(this); }
}
"""


def test_a_transfer_waiting_for_a_slot_is_queued_not_running(tmp_path):
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      t.start({ kind: 'download', name: 'f', total: 10, lease,
                run: async () => { say('ran'); } });
      say(t.list()[0].status, t.list()[0].ahead);
      await new Promise(r => setTimeout(r, 0));
      say('still:' + t.list()[0].status);
    """, tmp_path)
    assert out[:2] == ["queued", 2]
    assert "ran" not in out, "the work started before the slot was granted"
    assert out[-1] == "still:queued"


def test_the_grant_starts_the_work(tmp_path):
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      t.start({ kind: 'download', name: 'f', total: 10, lease,
                run: async () => { say('ran:' + t.list()[0].status); } });
      lease.grant();
      await new Promise(r => setTimeout(r, 10));
      say('after:' + t.list()[0].status);
    """, tmp_path)
    assert out[0] == "ran:running"
    assert out[1] == "after:done"


def test_the_slot_comes_back_however_the_transfer_ends(tmp_path):
    """A slot not returned is a member who cannot transfer again until the node
    times it out — so this must hold for a throw as much as for a success."""
    out = _run(_lease_stub() + """
      for (const mode of ['ok', 'throw']) {
        const t = new TransferStore();
        const lease = new L();
        t.start({ kind: 'download', name: 'f', total: 10, lease,
                  run: async () => { if (mode === 'throw') throw new Error('x'); } });
        lease.grant();
        await new Promise(r => setTimeout(r, 10));
        say(mode + ':' + lease.released.join(',') + ':' + t.list()[0].status);
      }
    """, tmp_path)
    assert out == ["ok:done:done", "throw:done:failed"]


def test_cancelling_while_queued_gives_the_slot_back(tmp_path):
    """The transfer somebody is most likely to give up on is the one that has
    not started. Its queue entry has to go, or the node grants a slot to a
    transfer that will never use it."""
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      const id = t.start({ kind: 'download', name: 'f', total: 10, lease,
                           run: async () => { say('ran'); } });
      t.cancel(id);
      say(t.list()[0].status, lease.released.join(','));
      lease.grant();
      await new Promise(r => setTimeout(r, 10));
      say('ran?', out.includes('ran'));
    """, tmp_path)
    assert out[0] == "cancelled"
    assert out[1] == "cancelled"
    assert out[-1] is False, "a cancelled transfer ran anyway once granted"


def test_a_queue_position_update_reaches_the_view(tmp_path):
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      const seen = [];
      t.subscribe(items => seen.push(items[0].ahead));
      t.start({ kind: 'download', name: 'f', total: 10, lease, run: async () => {} });
      lease.push('queued', 1);
      lease.push('queued', 0);
      say(seen.join('>'));
    """, tmp_path)
    assert out[0].endswith("1>0"), "the widget never learns it is moving up"


def test_a_transport_with_a_queued_transfer_is_not_closed(tmp_path):
    """Closing it would leave the transfer waiting for a grant that can never
    arrive — waiting for ever, with nothing left to answer."""
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      let closed = false;
      const transport = { close() { closed = true; } };
      t.start({ kind: 'download', name: 'f', total: 10, transport, lease,
                run: async () => {} });
      t.releaseWhenIdle(transport);
      say('closed while queued:', closed);
      lease.grant();
      await new Promise(r => setTimeout(r, 10));
      say('closed after:', closed);
    """, tmp_path)
    assert out[1] is False
    assert out[3] is True


def test_clearing_finished_keeps_what_is_waiting(tmp_path):
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      t.start({ kind: 'download', name: 'waiting', total: 1, lease, run: async () => {} });
      t.start({ kind: 'download', name: 'done', total: 1, run: async () => {} });
      await new Promise(r => setTimeout(r, 10));
      t.clearFinished();
      say(t.list().map(i => i.name + ':' + i.status).join(','));
    """, tmp_path)
    assert out[0] == "waiting:queued"


def test_asking_for_a_slot_on_a_dead_channel_does_not_throw(tmp_path):
    """
    The transport reconnects on its own and re-asks for every live lease when it
    does, so a closed channel at the moment a transfer starts is a wait, not a
    failure. `_fetchChunkResilient` has always treated it that way — and before
    leases existed a chunk request was the first thing to touch the channel, so
    a download begun on a briefly dead connection simply retried.

    Asking for a slot first made `_send` the first contact. It threw
    "DataChannel not open (state: closed)" straight out of `downloadEntry`,
    where nothing catches it: a download that used to recover became an error
    with no row in the widget to show it. Found live, by downloading a file
    just after a connection dropped.
    """
    module = tmp_path / "transport_lease.mjs"
    # The real Lease, lifted out as text — the class is not exported, and a
    # second copy of it here would agree with whatever it was copied from.
    src = (STATIC / "transport.js").read_text()
    # From the constant the class depends on, not from the class: lifting only
    # the class left LEASE_WATCHDOG_MS undefined, which the class reads the
    # first time it arms its watchdog.
    start = src.index("const LEASE_WATCHDOG_MS")
    end = src.index("\nclass MeshBayTransport")
    module.write_text(src[start:end] + "\nexport { Lease };\n")

    script = tmp_path / "case.mjs"
    script.write_text(f"""
import {{ Lease }} from '{module.as_posix()}';
const out = [];
const transport = {{
  supportsTransferSlots: true,
  _leases: new Map(),
  _send() {{ throw new Error('DataChannel not open (state: closed)'); }},
}};
let threw = null;
const lease = new Lease(transport, 'tr1', 'download', 10, 1, null);
try {{ lease._request(); }} catch (e) {{ threw = e.message; }}
out.push(threw);
// And releasing one must be just as safe: a lease not released is a member who
// cannot start another transfer until the node times it out.
try {{ lease.release('cancelled'); out.push('release ok'); }}
catch (e) {{ out.push('release threw: ' + e.message); }}
clearTimeout(lease._watchdog);
console.log(JSON.stringify(out));
""")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    out = json.loads(proc.stdout)
    assert out[0] is None, f"asking for a slot threw: {out[0]}"
    assert out[1] == "release ok"


def test_the_slot_is_asked_for_after_there_is_somewhere_to_write():
    """
    A granted slot has to be taken up within the node's acceptance deadline, so
    it must not be asked for until the download can actually start.

    Asking first reads better — the widget could draw a row while the target is
    being chosen — and is wrong: opening a target takes thirty seconds of
    streamed-download timeouts, or as long as somebody leaves a Save As dialog
    open. The node revokes the grant, passes it to the next in the queue
    (`transfer: reclaimed … (not_taken_up)` in its log), and the download then
    fetches under a `tr` that is no longer granted. Three downloads started, one
    arrived.

    Source-reading, because the ordering is the whole property and it has no
    behaviour of its own to drive: what matters is which call comes first.
    """
    src = (STATIC / "file-utils.js").read_text()
    fn = src[src.index("async function downloadEntry"):]
    fn = fn[:fn.index("\n}\n")]
    assert fn.index("_openDownloadTarget") < fn.index("openTransfer"), (
        "downloadEntry asks for a transfer slot before it has anywhere to "
        "write — the grant expires before the download can use it")


# ── the row exists from the click ───────────────────────────────────────────

def test_the_row_appears_before_the_target_is_open(tmp_path):
    """
    Opening a target is the slow part — the streamed path waits for the worker
    twice, a Save As dialog waits for a person — and the row used to be created
    only after it returned. Three clicks produced no panel at all, not even the
    icon, and then several rows at once.
    """
    out = _run("""
      const t = new TransferStore();
      let release;
      const opened = new Promise(r => { release = r; });
      t.start({ kind: 'download', name: 'film.mkv', total: 10,
                prepare: async () => { await opened; return { name: 'saved.mkv' }; },
                run: async () => { say('ran'); } });
      const shot = (when) => say(when + '=' + t.list().length + ':'
        + t.list().map(i => i.status + '/' + i.name).join(','));
      shot('click');
      release();
      await new Promise(r => setTimeout(r, 10));
      shot('after');
    """, tmp_path)
    # Tagged, not indexed. An earlier version counted pushes by hand and was one
    # out, which reads exactly like a failing assertion about the code.
    seen = dict(line.split("=", 1) for line in out
                if isinstance(line, str) and "=" in line)
    assert {"click", "after"} <= set(seen), f"probe produced: {out}"
    assert seen["click"] == "1:preparing/film.mkv", (
        f"no row, or the wrong one, at the moment of the click: {seen['click']}")
    assert seen["after"] == "1:done/saved.mkv", (
        f"the row must keep the name it was saved under: {seen['after']}")


def test_a_dismissed_dialog_leaves_nothing_behind(tmp_path):
    """Dismissing a Save As dialog is not a failure and not a cancellation:
    nothing was started, so nothing should be left on screen explaining it."""
    out = _run("""
      const t = new TransferStore();
      t.start({ kind: 'download', name: 'film.mkv', total: 10,
                prepare: async () => false,
                run: async () => { say('ran'); } });
      say('at click:', t.list().length);
      await new Promise(r => setTimeout(r, 10));
      say('after:', t.list().length, out.includes('ran'));
    """, tmp_path)
    assert out[1] == 1
    assert out[3] == 0, "a dismissed dialog left a row behind"
    assert out[4] is False


def test_the_slot_is_only_asked_for_once_there_is_somewhere_to_write(tmp_path):
    """
    A granted slot must be taken up within the node's deadline, and opening a
    target can outlast it. Asking first cost two of three downloads.
    """
    out = _run("""
      const t = new TransferStore();
      let release;
      const opened = new Promise(r => { release = r; });
      let asked = false;
      t.start({ kind: 'download', name: 'f', total: 10,
                prepare: async () => { await opened; return true; },
                makeLease: () => { asked = true; return {
                  state: 'granted', ahead: 0, tr: 'x',
                  acquire: () => Promise.resolve(), release: () => {} }; },
                run: async () => {} });
      say('while preparing, asked?', asked);
      release();
      await new Promise(r => setTimeout(r, 10));
      say('after preparing, asked?', asked, t.list()[0].status);
    """, tmp_path)
    assert out[1] is False, "the slot was taken before there was a target"
    assert out[3] is True
    assert out[4] == "done"


def test_a_target_that_cannot_be_opened_fails_the_row_it_already_has(tmp_path):
    """The refusal above the memory ceiling lands in the panel, on the row that
    is already there, rather than in a console nobody opens."""
    out = _run("""
      const t = new TransferStore();
      t.start({ kind: 'download', name: 'film.mkv', total: 10,
                prepare: async () => { throw new Error('too large for memory'); },
                run: async () => { say('ran'); } });
      await new Promise(r => setTimeout(r, 10));
      const it = t.list()[0];
      say(it.status, it.error, out.includes('ran'));
    """, tmp_path)
    assert out[0] == "failed"
    assert "too large" in out[1]
    assert out[2] is False


def test_a_transport_is_not_closed_under_a_preparing_transfer(tmp_path):
    """It has no lease yet and has moved no bytes, but closing its transport
    would strand it exactly like a queued one."""
    out = _run("""
      const t = new TransferStore();
      let release;
      const opened = new Promise(r => { release = r; });
      let closed = false;
      const transport = { close() { closed = true; } };
      t.start({ kind: 'download', name: 'f', total: 10, transport,
                prepare: async () => { await opened; return true; },
                run: async () => {} });
      t.releaseWhenIdle(transport);
      say('closed while preparing:', closed);
      release();
      await new Promise(r => setTimeout(r, 10));
      say('closed after:', closed);
    """, tmp_path)
    assert out[1] is False
    assert out[3] is True