"""
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
from spa_source import transport_source

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 = transport_source()
    # 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")]
    # `_openTargetInTurn` since target openings were serialised — same call,
    # queued. What is pinned is that it comes before the slot is asked for.
    assert fn.index("_openTargetInTurn") < 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


# ── Pause and resume ────────────────────────────────────────────────────────
#
# The rule the whole design turns on: **a paused transfer holds nothing.** Its
# slot goes back to the node the moment it stops, and resuming rejoins the queue
# at the tail. Anything else lets one member close a node by pausing four
# downloads and going to lunch.


def _pausable_run():
    """A `run` that stops where it is told and reports where it resumed."""
    return """
const mkStore = () => {
  const t = new TransferStore();
  const leases = [];
  const state = { starts: [], paused: null, aborted: false };
  t.start({
    kind: 'download', name: 'f', total: 1000,
    prepare: async () => ({ name: 'f', pausable: true }),
    makeLease: () => { const l = new L(); leases.push(l); return l; },
    run: async ({ signal, from }) => {
      state.starts.push(from);
      state.running = true;
      try {
      // Runs until told to stop, one "chunk" at a time.
      for (let i = from; i < 10; i++) {
        await new Promise(r => setTimeout(r, 5));
        if (signal.aborted) { const e = new Error('c'); e.name = 'AbortError'; throw e; }
        if (signal.paused) {
          signal.resumeFrom = i;
          const e = new Error('p'); e.name = 'PausedError'; throw e;
        }
      }
      } finally { state.running = false; }
    },
  });
  return { t, leases, state };
};
"""


def test_pausing_gives_the_slot_back(tmp_path):
    """The node has to get it back at once, not when the person resumes: the
    whole point of a queue is that a slot nobody is using is a slot somebody
    else can have."""
    out = _run(_lease_stub() + _pausable_run() + """
      const { t, leases } = mkStore();
      await new Promise(r => setTimeout(r, 5));
      leases[0].grant();
      await new Promise(r => setTimeout(r, 20));
      say('before:' + t.list()[0].status);
      t.pause(t.list()[0].id);
      await new Promise(r => setTimeout(r, 30));
      say('after:' + t.list()[0].status);
      say('released:' + leases[0].released.join(','));
      say('leases:' + leases.length);
    """, tmp_path)
    assert out[0] == "before:running"
    assert out[1] == "after:paused"
    assert out[2] == "released:paused", "a paused transfer kept its slot"
    assert out[3] == "leases:1"


def test_resuming_asks_for_a_new_slot_and_continues_where_it_stopped(tmp_path):
    """Rejoining at the tail is the design, not an accident: a paused transfer
    that could reclaim its old place would be a way to hold one."""
    out = _run(_lease_stub() + _pausable_run() + """
      const { t, leases, state } = mkStore();
      await new Promise(r => setTimeout(r, 5));
      leases[0].grant();
      await new Promise(r => setTimeout(r, 20));
      const id = t.list()[0].id;
      t.pause(id);
      await new Promise(r => setTimeout(r, 30));
      t.resume(id);
      await new Promise(r => setTimeout(r, 10));
      say('queued:' + t.list()[0].status, 'leases:' + leases.length);
      leases[1].grant();
      await new Promise(r => setTimeout(r, 120));
      say('end:' + t.list()[0].status);
      say('starts:' + state.starts.join(','));
    """, tmp_path)
    assert out[0] == "queued:queued", "a resumed transfer skipped the queue"
    assert out[1] == "leases:2", "resuming did not ask for a slot again"
    assert out[2] == "end:done"
    starts = out[3].split(":")[1].split(",")
    assert starts[0] == "0" and int(starts[1]) > 0, (
        f"resumed from {starts} — it started again from the beginning")


def test_a_transfer_whose_target_cannot_pause_is_not_paused(tmp_path):
    """A service-worker stream is a download the browser already owns: not
    writing to it stalls it outside our control and an idle worker is killed
    within seconds. A button that silently restarts from zero is worse than no
    button, so `pause` refuses rather than pretending."""
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      t.start({ kind: 'download', name: 'f', total: 10, lease,
                prepare: async () => ({ name: 'f' }),
                run: async ({ signal }) => {
                  while (!signal.aborted) await new Promise(r => setTimeout(r, 5));
                } });
      lease.grant();
      await new Promise(r => setTimeout(r, 20));
      const id = t.list()[0].id;
      say('pausable:' + t.list()[0].pausable);
      t.pause(id);
      await new Promise(r => setTimeout(r, 20));
      say('status:' + t.list()[0].status);
      t.cancel(id);
    """, tmp_path)
    assert out == ["pausable:false", "status:running"]


def test_cancelling_a_paused_transfer_actually_ends_it(tmp_path):
    """A paused run is parked on a promise. Without waking it, cancel marks the
    row and leaves the work parked for the life of the page, holding its target
    open — a button that lies, in the same way the first test in this file
    describes."""
    out = _run(_lease_stub() + _pausable_run() + """
      const { t, leases, state } = mkStore();
      await new Promise(r => setTimeout(r, 5));
      leases[0].grant();
      await new Promise(r => setTimeout(r, 20));
      const id = t.list()[0].id;
      t.pause(id);
      await new Promise(r => setTimeout(r, 30));
      t.cancel(id);
      // What matters is whether the store's own loop ends, not whether the row
      // says so: the row is marked at once either way.
      const settled = await Promise.race([
        t._items[0].promise.then(() => 'settled', () => 'settled'),
        new Promise(r => setTimeout(() => r('parked'), 60)),
      ]);
      say('status:' + t.list()[0].status);
      say('loop:' + settled);
      say('resumed:' + state.starts.length);
    """, tmp_path)
    assert out[0] == "status:cancelled"
    assert out[1] == "loop:settled", (
        "the run was still parked on the resume promise after a cancel — the "
        "row said cancelled over work that had not stopped")
    assert out[2] == "resumed:1", "cancelling started the work again"


def test_a_paused_transfer_still_counts_as_live(tmp_path):
    """It is not finished, and its transport must not be closed under it — the
    person is coming back to it."""
    out = _run(_lease_stub() + _pausable_run() + """
      const { t, leases } = mkStore();
      await new Promise(r => setTimeout(r, 5));
      leases[0].grant();
      await new Promise(r => setTimeout(r, 20));
      t.pause(t.list()[0].id);
      await new Promise(r => setTimeout(r, 30));
      say('pending:' + t.pending);
    """, tmp_path)
    assert out == ["pending:1"]


def test_an_upload_can_be_paused_without_a_prepare_step(tmp_path):
    """A download learns whether it can pause from its target, because only the
    target knows. An upload has no target to ask: a `File` is seekable and the
    node keeps the position, so it says so outright.

    This was missed when pause shipped — the button appeared on downloads and
    nowhere else, including in the desktop app where everything else works.
    """
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const leases = [];
      t.start({
        kind: 'upload', name: 'f', total: 100, pausable: true,
        makeLease: () => { const l = new L(); leases.push(l); return l; },
        run: async ({ signal, from }) => {
          for (let i = from || 0; i < 10; i++) {
            await new Promise(r => setTimeout(r, 5));
            if (signal.paused) {
              signal.resumeFrom = i;
              const e = new Error('p'); e.name = 'PausedError'; throw e;
            }
          }
        },
      });
      await new Promise(r => setTimeout(r, 5));
      leases[0].grant();
      await new Promise(r => setTimeout(r, 20));
      say('pausable:' + t.list()[0].pausable);
      t.pause(t.list()[0].id);
      await new Promise(r => setTimeout(r, 30));
      say('status:' + t.list()[0].status);
      say('released:' + leases[0].released.join(','));
    """, tmp_path)
    assert out == ["pausable:true", "status:paused", "released:paused"]


def test_an_upload_handed_a_lease_it_cannot_recreate_is_not_offered_pause(tmp_path):
    """Pausing gives the slot back. A transfer that cannot ask for another one
    would pause once and wait for ever, so the button is refused instead."""
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const lease = new L();
      t.start({ kind: 'upload', name: 'f', total: 100, pausable: true, lease,
                run: async ({ signal }) => {
                  while (!signal.aborted) await new Promise(r => setTimeout(r, 5));
                } });
      lease.grant();
      await new Promise(r => setTimeout(r, 20));
      const id = t.list()[0].id;
      t.pause(id);
      await new Promise(r => setTimeout(r, 20));
      say('status:' + t.list()[0].status);
      t.cancel(id);
    """, tmp_path)
    assert out == ["status:running"]


def test_pausing_one_transfer_leaves_the_others_alone(tmp_path):
    """Reported: three downloads running, one upload paused, and the three
    downloads lost their pause buttons.

    The button is drawn from `pausable` and the status, so this asks the store
    what it says about the other three at the moment one of them pauses.
    """
    out = _run(_lease_stub() + """
      const t = new TransferStore();
      const leases = [];
      const mk = (kind, name) => t.start({
        kind, name, total: 100, pausable: true,
        makeLease: () => { const l = new L(); leases.push(l); return l; },
        run: async ({ signal, from }) => {
          for (let i = from || 0; i < 40; i++) {
            await new Promise(r => setTimeout(r, 5));
            if (signal.aborted) { const e = new Error('c'); e.name = 'AbortError'; throw e; }
            if (signal.paused) {
              signal.resumeFrom = i;
              const e = new Error('p'); e.name = 'PausedError'; throw e;
            }
          }
        },
      });
      mk('download', 'd1'); mk('download', 'd2'); mk('download', 'd3');
      mk('upload', 'u1');
      await new Promise(r => setTimeout(r, 5));
      for (const l of leases) l.grant();
      await new Promise(r => setTimeout(r, 20));
      const up = t.list().find(i => i.kind === 'upload');
      say('before:' + t.list().filter(
        i => i.kind === 'download' && i.pausable && i.status === 'running').length);
      t.pause(up.id);
      await new Promise(r => setTimeout(r, 40));
      const rows = t.list();
      say('after:' + rows.filter(
        i => i.kind === 'download' && i.pausable && i.status === 'running').length);
      say('statuses:' + rows.map(i => i.kind[0] + ':' + i.status).join(','));
      for (const r of rows) t.cancel(r.id);
    """, tmp_path)
    assert out[0] == "before:3"
    assert out[1] == "after:3", (
        f"pausing the upload changed the downloads — {out[2]}")


def test_a_paused_transfer_is_not_filed_under_finished(tmp_path):
    """"Finished" was defined by exclusion — everything that is not running,
    queued or preparing — so it quietly swallowed `paused` the day pausing
    shipped. A transfer somebody stopped on purpose then sat beside the ones
    that are actually over, offering a resume button in the section of things
    that cannot be resumed.

    The three filters are lifted out of `app.js` and run, rather than described
    here: a copy of them in this file would agree with a broken version by
    construction.
    """
    src = (STATIC / "app.js").read_text()
    start = src.index("  const running = items.filter(")
    block = src[start:src.index("const active =", start)]

    script = tmp_path / "groups.mjs"
    script.write_text("""
const items = [
  { id: 1, status: 'running' },
  { id: 2, status: 'queued' },
  { id: 3, status: 'preparing' },
  { id: 4, status: 'paused' },
  { id: 5, status: 'done' },
  { id: 6, status: 'failed' },
  { id: 7, status: 'cancelled' },
];
""" + block + """
const seen = { running, waiting, paused, finished };
console.log(JSON.stringify(Object.fromEntries(
  Object.entries(seen).map(([k, v]) => [k, v.map(i => i.id)]))));
""")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    groups = json.loads(proc.stdout)

    assert groups["paused"] == [4]
    assert groups["finished"] == [5, 6, 7], (
        f"paused landed in {groups['finished']}")
    assert groups["running"] == [1] and groups["waiting"] == [2, 3]
    # Every row appears exactly once: a state added later that lands in no group
    # is a transfer the panel simply does not show.
    placed = sum((groups[k] for k in groups), [])
    assert sorted(placed) == [1, 2, 3, 4, 5, 6, 7]


def test_a_paused_transfer_still_counts_as_active(tmp_path):
    """The badge says how much is going on. A paused transfer is not over — the
    person means to come back to it — so counting it as nothing would be a
    panel that says "0" over work that is still there."""
    src = (STATIC / "app.js").read_text()
    start = src.index("  const running = items.filter(")
    block = src[start:src.index("\n\n", src.index("const active =", start))]

    script = tmp_path / "active.mjs"
    script.write_text("""
const items = [{ id: 1, status: 'paused' }, { id: 2, status: 'done' }];
""" + block + """
console.log(JSON.stringify({ active }));
""")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    assert json.loads(proc.stdout)["active"] == 1


def test_a_row_that_cannot_pause_says_so_where_the_button_would_be():
    """Reported from Chrome: four downloads with no pause button and an upload
    with one, and no way to tell why.

    The reason is real — without a granted folder the browser writes through the
    service worker, a download it already owns and cannot pause — but it was
    stated only in a Settings line nobody reads on the way to a download. A gap
    where the row above has a button is not an explanation.

    Shown only where a folder can actually be chosen: Firefox and Safari have
    none to choose, and "choose a folder" would be advice that cannot be taken.
    """
    src = (STATIC / "app.js").read_text()
    row = src[src.index("function TransferRow"):]
    row = row[:row.index("\n}\n")]

    hint = row[row.index("!it.pausable"):]
    hint = hint[:hint.index("`}")]
    assert "downloads.SUPPORTED" in hint, (
        "the hint would tell a Firefox user to choose a folder it cannot offer")
    assert "it.kind === 'download'" in hint, (
        "an upload is always pausable; this is about download targets")
    assert "transfers.not_pausable" in hint, "the reason is not stated"
    # Not a button. There is nothing to click, and a disabled one invites the
    # click anyway.
    assert "<button" not in hint


def test_the_reason_is_translated_everywhere():
    """`t()` falls back to the key, so a missing catalogue entry shows
    `transfers.not_pausable` in a tooltip rather than a sentence."""
    for path in sorted((STATIC / "locales").glob("*.js")):
        assert "'transfers.not_pausable'" in path.read_text(), path.name