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

"""
Showing a subtitle track in the player.

The node extracts one track whole, converts it to WebVTT and caches it under
its own hash; this side asks for it by ordinal, pulls the blob through the
ordinary chunk path and hangs a `<track>` on the video element. Four things
have to hold here:

**The selector exists only where the node said there was something to show.**
It is drawn from `subtitle_tracks` in `stream_init` and from nothing else —
there is no version check in the player — so a node too old to enumerate them
draws no selector and is never sent a `subtitle_req` it would answer "unknown
message type" to. The list also carries only the tracks the node can convert,
so what the menu offers is what will actually appear.

**The ordinal travels untouched.** It counts every subtitle stream in the
container, including the bitmap ones that are never listed, so it is not the
track's position in the list this client received. Renumbering it here would
map a different stream on the node and show the wrong language, or nothing.

**A seek does not take the subtitles down.** The extraction is whole-file, so
the cues are absolute: the `<track>` outlives every restart of the
MediaSource, including the one an audio-language change produces.

**The reply is matched to the request by file *and* track.** Two extractions
for one film can be in flight when the viewer changes their mind, and the
reply that arrives first is not necessarily the one asked for first.
"""

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

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
APP = STATIC / "video-player.js"
TRANSPORT = STATIC / "transport.js"

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


@pytest.fixture(scope="module")
def app():
    return APP.read_text()


@pytest.fixture(scope="module")
def transport():
    return TRANSPORT.read_text()


def _lift(src: str, name: str) -> str:
    """One top-level function, as text, for node to execute."""
    start = src.index(f"function {name}(")
    depth, i, seen = 0, start, False
    while i < len(src):
        if src[i] == "{":
            depth += 1
            seen = True
        elif src[i] == "}":
            depth -= 1
            if seen and depth == 0:
                return src[start:i + 1]
        i += 1
    raise AssertionError(f"{name} never closes")


def _player(app: str) -> str:
    """The whole component, by position in the file.

    Not the brace matcher below: `function VideoPlayer({ entry, ... })`
    destructures its props, so counting from the first `{` closes on the
    parameter list and returns the signature alone — a lift that finds
    nothing it was asked about and asserts happily against an empty string.
    """
    i = app.index("function VideoPlayer(")
    nxt = app.find("\nfunction ", i + 1)
    return app[i:nxt if nxt > 0 else len(app)]


def _block(src: str, opener: str) -> str:
    """The body of one `x = (args) => {` assignment, braces matched."""
    start = src.index(opener)
    depth, i, seen = 0, start, False
    while i < len(src):
        if src[i] == "{":
            depth += 1
            seen = True
        elif src[i] == "}":
            depth -= 1
            if seen and depth == 0:
                return src[start:i + 1]
        i += 1
    raise AssertionError(f"{opener!r} never closes")


# ── The label, run rather than read ───────────────────────────────────────────

def _label_cases(tmp_path, app, cases, locale="en"):
    script = tmp_path / "sublabel.mjs"
    src = "\n".join([
        app[app.index("const _ISO639 = {"):app.index("};", app.index("const _ISO639 = {")) + 2],
        _lift(app, "_languageName"),
        _lift(app, "subtitleTrackLabel"),
    ])
    script.write_text(
        f"const getLocale = () => '{locale}';\n"
        # Not every key is interpolated — the forced/SDH qualifiers take no
        # parameters — and a stub that assumes one throws where the real `t`
        # returns a string. A fixture narrower than production tests itself.
        "const t = (k, p) => (p ? `${k}:${p.n}` : k);\n"
        + src
        + "\nconst out = JSON.parse(process.argv[2]).map(subtitleTrackLabel);\n"
          "console.log(JSON.stringify(out));\n")
    proc = subprocess.run(
        ["node", str(script), json.dumps(cases)],
        capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_the_language_is_named_not_shown_as_a_tag(tmp_path, app):
    """Same fold as the audio tracks: ffprobe reports either ISO 639-2 variant
    and a library uses both."""
    out = _label_cases(tmp_path, app, [
        {"i": 0, "lang": "fre"},
        {"i": 1, "lang": "fra"},
        {"i": 2, "lang": "eng"},
    ])
    assert out[0] == out[1], f"fre and fra must agree: {out}"
    assert "French" in out[0] and "English" in out[2], out


def test_the_container_title_tells_a_forced_track_from_a_full_one(tmp_path, app):
    """The difference the title carries is not cosmetic here.

    "Forced" and "SDH" are tagged with the same language as the ordinary track
    they sit beside. Dropping the title makes them one menu entry repeated,
    and picking the wrong one is the difference between a full translation and
    three lines of signage in a whole film.
    """
    out = _label_cases(tmp_path, app, [
        {"i": 0, "lang": "eng", "title": "Forced"},
        {"i": 1, "lang": "eng", "title": "SDH"},
        {"i": 2, "lang": "eng"},
    ])
    assert len({*out}) == 3, f"three tracks must give three entries: {out}"
    assert "Forced" in out[0] and "SDH" in out[1]


def test_a_forced_track_is_named_as_one_in_the_reader_s_language(tmp_path, app):
    """The entry that was reported as a broken feature.

    A forced track shows signage and foreign dialogue only — on a real film,
    77 seconds of text across 2h32 — so picking it and seeing nothing is its
    normal behaviour. It has to be possible to tell it from the full track
    beside it, which carries the same language tag, and the disposition says
    so where the container's English title tag is often simply absent.
    """
    out = _label_cases(tmp_path, app, [
        {"i": 0, "lang": "fre", "forced": True},
        {"i": 1, "lang": "fre"},
        {"i": 2, "lang": "eng", "sdh": True},
    ])
    assert out[0] != out[1], f"a forced track must not read like the full one: {out}"
    # The catalogues are stubbed here, so what is asserted is that the label
    # goes through `t` at all: the qualifier has to be translated, not the
    # English word a muxer typed into the container.
    assert out[0] == 'French — video.subtitles_forced', out
    assert out[1] == 'French', out
    assert out[2] == 'English — video.subtitles_sdh', out


def test_the_disposition_wins_over_the_container_title(tmp_path, app):
    """A muxer's "Forced" is one spelling of many, and in one language."""
    out = _label_cases(tmp_path, app, [
        {"i": 0, "lang": "fre", "title": "FORCE VF", "forced": True},
    ])
    assert 'FORCE VF' not in out[0], out
    assert out[0] == 'French — video.subtitles_forced', out


def test_an_untagged_track_is_numbered_not_called_unknown(tmp_path, app):
    """A file with no language tags still needs distinguishable entries."""
    out = _label_cases(tmp_path, app, [{"i": 3, "lang": None}, {"i": 4, "lang": "und"}])
    assert out == ["video.subtitle_track_n:4", "video.subtitle_track_n:5"], out


# ── The shape of the feature, read from the source ────────────────────────────

def test_the_selector_comes_from_the_node_list_and_not_from_a_version(app):
    """Discovery from the answer. There is no version check in the player, and
    an empty list is the whole of "this node has no subtitles for you"."""
    assert "msg.subtitle_tracks" in app
    assert "subtitleTracks.length > 0 &&" in app, (
        "the selector must be drawn from the node's list")
    for forbidden in ("MNP_VERSION", "v >= '3.3'", "'3.3'"):
        assert forbidden not in app, (
            f"the player decided on {forbidden!r} instead of on the answer")


def test_the_ordinal_is_passed_through_rather_than_renumbered(app):
    """`track.i` is the node's ordinal; the list index is not it."""
    assert "requestSubtitle(entry.id, track.i)" in app, (
        "the request must carry the ordinal the node published")
    assert not re.search(r"requestSubtitle\([^)]*\bindex\b", app), (
        "a list position was sent where a stream ordinal belongs")


def test_a_seek_does_not_take_the_subtitles_down(app):
    """The payoff of extracting whole-file rather than per-seek.

    `stream_init` arrives again on every seek and on every audio-language
    change. It may restate the available tracks; it must not clear the chosen
    one or revoke the blob, or every seek would re-extract and the track would
    blink out mid-film.
    """
    body = _block(app, "transport.onStreamInit = (msg) => {")
    assert "setSubtitleTracks(" in body
    for forbidden in ("setSubtitleUrl(", "setSubtitleTrack(", "subtitleUrlRef.current ="):
        assert forbidden not in body, (
            f"stream_init touches {forbidden!r}, so a seek disturbs the track")


def test_the_blob_is_pulled_through_the_ordinary_chunk_path(app):
    """Not a new transfer mechanism — the same indirection as a poster."""
    assert "pipelinedDownload(" in app
    assert "info.hash" in app, "the cache hash from the reply is what is fetched"


def test_a_stale_reply_cannot_install_its_track(app):
    """Two extractions in flight, and the last answered is not the last asked.

    Without the generation the older reply overwrites the newer choice, and
    the menu then shows a tick against a language that is not on screen.
    """
    body = _player(app)
    assert "subtitleGenRef.current !== gen" in body, (
        "nothing stops an out-of-date extraction from installing its blob")
    assert "URL.revokeObjectURL(url)" in body, (
        "the superseded blob must be released, not merely ignored")


def test_the_reply_is_matched_on_the_file_and_the_track(transport):
    """One film's two tracks are exactly the pair that can be in flight
    together, so the file id alone cannot route the reply."""
    assert "`subtitle:${obj.file_id}:${obj.track}`" in transport
    assert "`subtitle:${msg.file_id}:${msg.track}`" in transport


def test_a_failed_extraction_does_not_take_the_film_down(app):
    """Subtitles are an addition to a film that is already playing.

    `setError` is the player's fatal path — it replaces the picture. A text
    track that could not be read must not reach it.
    """
    body = _player(app)
    start = body.index("const selectSubtitle = useCallback(")
    end = body.index("}, [entry, transportRef, gekRef]);", start)
    assert "setError(" not in body[start:end], (
        "a subtitle failure takes the whole player down")
    assert "setSubtitleError(true)" in body[start:end]