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

"""
Subtitles on a cast receiver.

Casting takes a different road than the player does, and the difference that
matters is the clock. The node extracts a subtitle whole, so its cues carry the
film's own timeline; the player feeds its SourceBuffer a `timestampOffset`
equal to the stream's start, which puts the element's `currentTime` on that
same timeline and lets the cues be used as they arrive.

A receiver has no such offset. The relay hands it the node's fragments
untouched, and those are rebased to zero at the seek point. Film time and
stream time therefore differ by exactly the start, and a document sent across
unshifted is wrong by however far the viewer had seeked — an hour into a film,
an hour wrong, with nothing in any log to say so. `shiftWebVtt` is the whole
correction and the first half of this file tests it by running it.

The second half runs the relay itself. Unlike the Electron shell it is plain
Node with no dependencies, so it can be started, served from and stopped here.
Three of its properties are load-bearing and invisible when broken: the
subtitle sits behind the same token as the stream, it carries CORS headers
because a receiver fetches it with XHR rather than handing it to a media
element, and its URL changes when its content does — a side-loaded track is
cached by address, so a fixed URL would leave the old language on screen.
"""

import json
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"
CLIENT = Path(__file__).resolve().parents[2] / "meshbay-client"
RELAY = CLIENT / "src" / "cast-relay.js"
CHROMECAST = CLIENT / "src" / "cast-chromecast.js"

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


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 _run(tmp_path, body: str, *args: str):
    script = tmp_path / "cast.mjs"
    app = APP.read_text(encoding="utf-8")
    script.write_text(
        _lift(app, "shiftWebVtt")
        + "\n" + _lift(app, "castSubtitleFor")
        + "\n" + body, encoding="utf-8")
    proc = subprocess.run(
        ["node", str(script), *args],
        capture_output=True, text=True, timeout=30)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


VTT = """WEBVTT

NOTE this file was converted from an embedded track

1
00:00:10.000 --> 00:00:12.500
Before the seek.

2
01:23:45.000 --> 01:23:47.250 line:90%
Hello.

3
01:30:00.000 --> 01:30:02.000
Goodbye.
"""


def _timings(vtt: str):
    return [line.strip() for line in vtt.splitlines() if "-->" in line]


# ── The clock ────────────────────────────────────────────────────────────────

def test_cues_move_back_by_the_streams_start(tmp_path):
    """
    A stream that begins at 01:20:00 makes a cue at 01:23:45 land at 00:03:45.

    This is the correction the whole feature rests on. Without it the cues keep
    the film's timeline while the receiver counts from zero, and the error is
    the size of the seek rather than a small drift — invisible in code review,
    unmistakable on screen.
    """
    out = _run(tmp_path, """
      const shifted = shiftWebVtt(process.argv[2], -4800);
      console.log(JSON.stringify(shifted));
    """, VTT)
    assert "00:03:45.000 --> 00:03:47.250 line:90%" in out
    assert "00:10:00.000 --> 00:10:02.000" in out


def test_a_stream_from_the_top_leaves_every_cue_where_it_was(tmp_path):
    """No seek means no shift, and the document must come back unchanged."""
    out = _run(tmp_path, """
      console.log(JSON.stringify(shiftWebVtt(process.argv[2], 0)));
    """, VTT)
    assert _timings(out) == _timings(VTT)


def test_cues_before_the_stream_are_dropped_not_clamped(tmp_path):
    """
    A cue that has already finished when the stream starts must disappear.

    Clamping it to zero instead would print a line from before the seek over
    the first frames after it — the one failure mode that looks like a bug in
    the extraction rather than in the arithmetic.
    """
    out = _run(tmp_path, """
      console.log(JSON.stringify(shiftWebVtt(process.argv[2], -4800)));
    """, VTT)
    assert "Before the seek." not in out
    assert len(_timings(out)) == 2


def test_a_cue_straddling_the_seek_survives_and_starts_at_zero(tmp_path):
    """
    Someone is mid-sentence when the viewer lands. The line is still owed to
    them, so the cue is kept with its start pulled up to the stream's own.
    """
    out = _run(tmp_path, """
      const vtt = 'WEBVTT\\n\\n1\\n00:00:08.000 --> 00:00:14.000\\nMid-sentence.\\n';
      console.log(JSON.stringify(shiftWebVtt(vtt, -10)));
    """)
    assert "00:00:00.000 --> 00:00:04.000" in out
    assert "Mid-sentence." in out


def test_the_header_and_its_notes_travel_unchanged(tmp_path):
    """
    A document that loses its `WEBVTT` line is not a WebVTT document, and a
    receiver rejects it whole rather than complaining about one cue.
    """
    out = _run(tmp_path, """
      console.log(JSON.stringify(shiftWebVtt(process.argv[2], -4800)));
    """, VTT)
    assert out.startswith("WEBVTT")
    assert "NOTE this file was converted" in out


def test_comma_decimals_are_read_and_written_back_as_points(tmp_path):
    """
    SRT writes `00:00:10,000` and some converters leave the comma in place.
    Reading it and emitting the WebVTT spelling costs nothing and saves a
    document that would otherwise be silently dropped cue by cue.
    """
    out = _run(tmp_path, """
      const vtt = 'WEBVTT\\n\\n1\\n00:01:00,500 --> 00:01:02,000\\nComma.\\n';
      console.log(JSON.stringify(shiftWebVtt(vtt, -30)));
    """)
    assert "00:00:30.500 --> 00:00:32.000" in out


def test_the_payload_negates_the_start_it_is_given(tmp_path):
    """
    `castSubtitleFor` takes where the stream *begins* and must subtract it.
    Getting the sign wrong doubles the error instead of cancelling it, and both
    directions produce a plausible-looking document.
    """
    out = _run(tmp_path, """
      const sub = { text: process.argv[2], language: 'fr', label: 'French' };
      const payload = castSubtitleFor(sub, 4800);
      console.log(JSON.stringify(payload));
    """, VTT)
    assert "00:03:45.000 --> 00:03:47.250" in out["vtt"]
    assert out["language"] == "fr"
    assert out["label"] == "French"


def test_no_track_showing_means_no_payload(tmp_path):
    """
    Subtitles off has to reach the relay as null, not as an empty document: an
    empty WebVTT file is a track the receiver will happily display nothing
    from, and its menu would still offer it.
    """
    out = _run(tmp_path, """
      console.log(JSON.stringify([
        castSubtitleFor(null, 0), castSubtitleFor({ text: '' }, 0),
      ]));
    """)
    assert out == [None, None]


# ── The relay ────────────────────────────────────────────────────────────────

relay_only = pytest.mark.skipif(
    not RELAY.exists(), reason="desktop client sources not present")

RELAY_SCRIPT = """
const CastRelay = require(process.argv[2]);

(async () => {
  const relay = new CastRelay();
  const started = await relay.start({
    codec: 'avc1.640028',
    initSegment: Buffer.from([0, 0, 0, 8, 102, 116, 121, 112]),
    subtitle: { vtt: 'WEBVTT\\n\\n1\\n00:00:01.000 --> 00:00:02.000\\nHi.\\n',
                language: 'fr', label: 'French' },
  });

  const out = { streamUrl: started.url, subtitleUrl: started.subtitle.url };

  const get = await fetch(out.subtitleUrl);
  out.status = get.status;
  out.contentType = get.headers.get('content-type');
  out.allowOrigin = get.headers.get('access-control-allow-origin');
  out.allowHeaders = get.headers.get('access-control-allow-headers');
  out.body = await get.text();

  const noToken = await fetch(out.subtitleUrl.replace(/t=[0-9a-f]+/, 't=0'));
  out.withoutToken = noToken.status;

  const preflight = await fetch(out.subtitleUrl, { method: 'OPTIONS' });
  out.preflight = preflight.status;
  out.preflightAllowOrigin =
    preflight.headers.get('access-control-allow-origin');

  const second = relay.setSubtitle({ vtt: 'WEBVTT\\n\\n', language: 'en' });
  out.secondUrl = second.url;

  relay.setSubtitle(null);
  out.afterClear = relay.subtitleUrl;
  out.afterClearStatus = (await fetch(out.secondUrl)).status;
  const stream = await fetch(out.streamUrl);
  out.streamStillServes = stream.status;
  out.streamAllowOrigin = stream.headers.get('access-control-allow-origin');

  await relay.stop();
  console.log(JSON.stringify(out));
  process.exit(0);
})().catch((err) => { console.error(err); process.exit(1); });
"""


@pytest.fixture(scope="module")
def served(tmp_path_factory):
    script = tmp_path_factory.mktemp("relay") / "serve.cjs"
    script.write_text(RELAY_SCRIPT, encoding="utf-8")
    proc = subprocess.run(
        ["node", str(script), str(RELAY)],
        capture_output=True, text=True, timeout=60)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout.strip().splitlines()[-1])


@relay_only
def test_the_subtitle_is_served_as_webvtt(served):
    assert served["status"] == 200
    assert served["contentType"].startswith("text/vtt")
    assert served["body"].startswith("WEBVTT")


@relay_only
def test_the_subtitle_sits_behind_the_same_token_as_the_stream(served):
    """
    The relay's only defence is that its URL cannot be guessed. A subtitle
    path exempt from the token would hand the film's dialogue — often the
    whole script — to anything on the Wi-Fi.
    """
    assert served["withoutToken"] == 403


@relay_only
def test_the_receiver_is_allowed_to_read_it(served):
    """
    A side-loaded track is fetched with XHR from the receiver's own origin, so
    without CORS it fails as a network error and the film plays on with no
    subtitles and no message. `Range` is named because the receiver sends it
    even for a document it reads whole.
    """
    assert served["allowOrigin"] == "*"
    for header in ("Content-Type", "Accept-Encoding", "Range"):
        assert header in served["allowHeaders"]


@relay_only
def test_the_stream_carries_the_same_headers_as_its_subtitle(served):
    """
    A receiver given a side-loaded track reads the media through the same
    CORS-checked path, so headers on one and not the other fails the load
    entirely rather than losing the subtitles alone. They widen nothing the
    token does not already govern: a page holding the URL could put it in a
    media element with or without them.
    """
    assert served["streamAllowOrigin"] == "*"


@relay_only
def test_the_preflight_is_answered_before_the_token_is_checked(served):
    """
    A browser sends `OPTIONS` without the credentials that would let it pass a
    token check, so refusing it there would deny every well-formed request.
    """
    assert served["preflight"] == 204
    assert served["preflightAllowOrigin"] == "*"


@relay_only
def test_a_new_subtitle_gets_a_new_address(served):
    """
    A receiver caches a side-loaded track by its URL. Serving different cues
    from a fixed address leaves the previous language on screen, which reads
    as the switch having been ignored.
    """
    assert served["secondUrl"] != served["subtitleUrl"]


@relay_only
def test_turning_subtitles_off_takes_the_file_away_but_not_the_film(served):
    """
    Clearing the track must not disturb playback: the stream is the reason the
    relay exists, and losing the picture to a subtitle change would be a far
    worse failure than the one being fixed.
    """
    assert served["afterClear"] is None
    assert served["afterClearStatus"] == 404
    assert served["streamStillServes"] == 200


# ── What the receiver is told ────────────────────────────────────────────────

chromecast_only = pytest.mark.skipif(
    not CHROMECAST.exists(), reason="desktop client sources not present")

CHROMECAST_SCRIPT = """
const m = require(process.argv[2]);
const sub = { url: 'http://10.0.0.2:19550/subs.vtt?t=ab&v=3',
              language: 'fr', label: 'French — Forced' };
console.log(JSON.stringify({
  with: { media: m.mediaFor('http://10.0.0.2:19550/stream.mp4?t=ab', sub),
          options: m.loadOptionsFor(sub) },
  without: { media: m.mediaFor('http://10.0.0.2:19550/stream.mp4?t=ab', null),
             options: m.loadOptionsFor(null) },
}));
"""


@pytest.fixture(scope="module")
def loaded(tmp_path_factory):
    script = tmp_path_factory.mktemp("cc") / "media.cjs"
    script.write_text(CHROMECAST_SCRIPT, encoding="utf-8")
    proc = subprocess.run(
        ["node", str(script), str(CHROMECAST)],
        capture_output=True, text=True, timeout=60)
    if proc.returncode != 0:
        pytest.skip(f"cast-chromecast.js is not loadable here: {proc.stderr}")
    return json.loads(proc.stdout)


@chromecast_only
def test_the_track_is_declared_and_switched_on(loaded):
    """
    Declaring a track without naming it in `activeTrackIds` loads it and shows
    nothing, which is the same symptom as not declaring it at all.
    """
    track = loaded["with"]["media"]["tracks"][0]
    assert track["type"] == "TEXT"
    assert track["subtype"] == "SUBTITLES"
    assert track["trackContentType"] == "text/vtt"
    assert track["language"] == "fr"
    assert loaded["with"]["options"]["activeTrackIds"] == [track["trackId"]]


@chromecast_only
def test_a_stream_without_subtitles_declares_none(loaded):
    """
    A stale id in `activeTrackIds` is a load error on the receiver, and the
    load error takes the film with it.
    """
    assert "tracks" not in loaded["without"]["media"]
    assert loaded["without"]["options"]["activeTrackIds"] == []