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

"""
The music player's queue: replace, append, play-next, remove, reshuffle.

The queue was replaceable and nothing else — every `onPlayQueue` reset
`tracks`, `order` and `pos` together, which is all an album needs. Playlists
add "play next" and "add to queue", which do not replace, and the three
`useState`s they would have been built on cannot express an append correctly:
`setOrder` needs the length `setTracks` is about to produce and cannot see it,
so two enqueues batched into one tick both read the stale length and write
indices past the end of `tracks`. `queue-ops.js` is one reducer over one state
object, which is the only shape in which that is not a defect.

The whole module is executed here rather than a regex-extracted function of it:
it has no imports precisely so that it can be, and a copy of the reducer in a
test would keep agreeing with the original right up until one of them changed.

See docs/playlists.md §9.
"""

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

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
SRC = STATIC / "queue-ops.js"

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

IMPORT = re.compile(r"^\s*import\b", re.M)
EXPORT = re.compile(r"^export \{[^}]*\};?\s*$", re.M)


@pytest.fixture(scope="module")
def module_source():
    text = SRC.read_text()
    assert not IMPORT.search(text), (
        "queue-ops.js has gained an import. It is executed standalone here, "
        "and the queue is untested from the moment it cannot be — keep the "
        "module free of imports, or this test needs a bundler")
    stripped, n = EXPORT.subn("", text)
    assert n == 1, (
        "queue-ops.js no longer ends in a single export statement — the test "
        "can no longer strip it to run the module")
    return stripped


def _run(tmp_path, module_source, body):
    script = tmp_path / "case.js"
    script.write_text(f"{module_source}\n{body}\n")
    out = subprocess.run(
        ["node", str(script)], capture_output=True, text=True, timeout=30)
    assert out.returncode == 0, out.stderr
    return json.loads(out.stdout)


def _tracks(*ids):
    return [{"id": i, "name": f"{i}.flac", "size": 1, "groupId": "g"} for i in ids]


def _reduce(tmp_path, module_source, actions, start=None):
    """Fold `actions` over the reducer, one after another, and report the end
    state as ids so a test reads as the play order it means."""
    body = f"""
      let s = {json.dumps(start) if start else "emptyQueue()"};
      for (const a of {json.dumps(actions)}) {{
        // A deterministic shuffle: reverses the order, then keepFirst is
        // pulled to the front by the reducer itself. Real randomness would
        // make every shuffle assertion a coin toss.
        s = queueReducer(s, {{ ...a, rand: () => 0 }});
      }}
      console.log(JSON.stringify({{
        play: s.order.map((i) => s.tracks[i].id),
        pos: s.pos,
        playing: s.order.length ? s.tracks[s.order[s.pos]].id : null,
        nTracks: s.tracks.length,
      }}));
    """
    return _run(tmp_path, module_source, body)


# ── replace: the path that already worked, pinned ────────────────────────────

def test_replace_unshuffled_is_the_plain_order_from_the_requested_index(
        tmp_path, module_source):
    """What playing track 3 of an album has always done. If this changes,
    every album in the application changed with it."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c", "d"), "startIndex": 2},
    ])
    assert out["play"] == ["a", "b", "c", "d"]
    assert out["pos"] == 2
    assert out["playing"] == "c"


def test_replace_shuffled_keeps_the_requested_track_first(tmp_path, module_source):
    """Shuffling on a chosen track must not start a different one."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c", "d"),
         "startIndex": 2, "shuffle": True},
    ])
    assert out["pos"] == 0
    assert out["playing"] == "c"
    assert sorted(out["play"]) == ["a", "b", "c", "d"]


def test_replace_discards_the_previous_queue(tmp_path, module_source):
    """Loading a playlist replaces; it does not accumulate (docs §9.1)."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b")},
        {"type": "replace", "tracks": _tracks("x", "y", "z")},
    ])
    assert out["play"] == ["x", "y", "z"]
    assert out["nTracks"] == 3


# ── append and play-next ─────────────────────────────────────────────────────

def test_append_to_an_empty_queue_plays_it(tmp_path, module_source):
    """Enqueueing with nothing playing has to start something, or the button
    does nothing at all the first time it is pressed."""
    out = _reduce(tmp_path, module_source, [
        {"type": "append", "tracks": _tracks("a", "b")},
    ])
    assert out["play"] == ["a", "b"]
    assert out["playing"] == "a"


def test_append_goes_to_the_end_and_does_not_move_the_playhead(
        tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c"), "startIndex": 1},
        {"type": "append", "tracks": _tracks("x", "y")},
    ])
    assert out["play"] == ["a", "b", "c", "x", "y"]
    assert out["playing"] == "b"


def test_play_next_lands_immediately_after_what_is_playing(
        tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c"), "startIndex": 0},
        {"type": "insertNext", "tracks": _tracks("x")},
    ])
    assert out["play"] == ["a", "x", "b", "c"]
    assert out["playing"] == "a"


def test_play_next_on_the_last_track_still_lands_after_it(tmp_path, module_source):
    """The splice index is past the end of `order`; a slice must cope rather
    than dropping the entry silently."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b"), "startIndex": 1},
        {"type": "insertNext", "tracks": _tracks("x")},
    ])
    assert out["play"] == ["a", "b", "x"]
    assert out["playing"] == "b"


def test_play_next_while_shuffled_inserts_into_the_play_order(
        tmp_path, module_source):
    """Not into `tracks` — the played sequence is `order`, and "next" means
    next in what is actually being played."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c", "d"),
         "startIndex": 0, "shuffle": True},
        {"type": "insertNext", "tracks": _tracks("x")},
    ])
    assert out["play"][0] == "a"
    assert out["play"][1] == "x"
    assert out["playing"] == "a"


def test_appending_nothing_is_not_a_change(tmp_path, module_source):
    """An album card with no tracks must not clear the playhead."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b"), "startIndex": 1},
        {"type": "append", "tracks": []},
    ])
    assert out["play"] == ["a", "b"]
    assert out["playing"] == "b"


# ── the defect this module exists to prevent ─────────────────────────────────

def test_two_appends_in_one_tick_do_not_write_indices_past_the_end(
        tmp_path, module_source):
    """*The* reason the queue is a reducer (docs §9.3).

    Two `useState`s updated from one event both read the length captured when
    the handler was built, so the second append's indices collide with the
    first's — a double click on "add to queue" produced a queue playing the
    wrong tracks, or holes. Folding through the reducer is what a batched
    render does, and every index has to be distinct and in range.
    """
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b")},
        {"type": "append", "tracks": _tracks("x")},
        {"type": "append", "tracks": _tracks("y")},
    ])
    assert out["play"] == ["a", "b", "x", "y"], (
        "an append read a stale track count — this is the three-useState bug")
    assert out["nTracks"] == 4


def test_many_appends_stay_in_range(tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a")},
    ] + [{"type": "append", "tracks": _tracks(f"t{n}")} for n in range(20)])
    assert out["play"] == ["a"] + [f"t{n}" for n in range(20)]
    assert out["nTracks"] == 21


# ── removal ──────────────────────────────────────────────────────────────────

def test_removing_the_playing_track_slides_the_next_one_in(tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c"), "startIndex": 1},
        {"type": "removeAt", "at": 1},
    ])
    assert out["play"] == ["a", "c"]
    assert out["playing"] == "c"


def test_removing_before_the_playhead_keeps_the_same_track_playing(
        tmp_path, module_source):
    """The index moved; what is playing must not."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c"), "startIndex": 2},
        {"type": "removeAt", "at": 0},
    ])
    assert out["play"] == ["b", "c"]
    assert out["playing"] == "c"


def test_removing_the_last_remaining_track_does_not_leave_pos_dangling(
        tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a")},
        {"type": "removeAt", "at": 0},
    ])
    assert out["play"] == []
    assert out["pos"] == 0
    assert out["playing"] is None


def test_removing_out_of_range_is_not_a_change(tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b"), "startIndex": 1},
        {"type": "removeAt", "at": 7},
    ])
    assert out["play"] == ["a", "b"]
    assert out["playing"] == "b"


# ── shuffle ──────────────────────────────────────────────────────────────────

def test_shuffling_on_mid_album_does_not_interrupt_what_is_playing(
        tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c", "d"), "startIndex": 2},
        {"type": "reshuffle", "shuffle": True},
    ])
    assert out["pos"] == 0
    assert out["playing"] == "c"
    assert sorted(out["play"]) == ["a", "b", "c", "d"]


def test_shuffling_off_returns_to_the_album_order_at_the_same_track(
        tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b", "c", "d"),
         "startIndex": 2, "shuffle": True},
        {"type": "reshuffle", "shuffle": False},
    ])
    assert out["play"] == ["a", "b", "c", "d"]
    assert out["playing"] == "c"


def test_shuffle_after_an_append_covers_the_appended_tracks(
        tmp_path, module_source):
    """`order` is rebuilt from `tracks.length`, so it has to have grown."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b")},
        {"type": "append", "tracks": _tracks("x", "y")},
        {"type": "reshuffle", "shuffle": True},
    ])
    assert sorted(out["play"]) == ["a", "b", "x", "y"]
    assert out["playing"] == "a"


def test_shuffling_a_queue_holding_the_same_track_twice_keeps_the_right_copy(
        tmp_path, module_source):
    """A queue could not hold duplicates until "add to queue" existed, and the
    first reshuffle recovered the current index by searching `tracks` for the
    playing entry's id — which finds the *first* copy. Enqueue a track that is
    already in the queue, play the second copy, toggle shuffle, and playback
    jumped backwards. `order[pos]` is the index and needs no search."""
    out = _reduce(tmp_path, module_source, [
        {"type": "replace", "tracks": _tracks("a", "b")},
        {"type": "append", "tracks": _tracks("a")},
        {"type": "skipTo", "pos": 2},
        {"type": "reshuffle", "shuffle": False},
    ])
    assert out["pos"] == 2, "the reshuffle jumped to the first copy of 'a'"
    assert out["playing"] == "a"


def test_reshuffling_an_empty_queue_does_not_throw(tmp_path, module_source):
    out = _reduce(tmp_path, module_source, [
        {"type": "reshuffle", "shuffle": True},
    ])
    assert out["play"] == []
    assert out["pos"] == 0