"""
Whole files on a cast receiver — photos and music tracks — and the television
a session casts to.

A file takes a different road to the television than a film: no stream, one
file at a time, handed to the relay in pieces (begin, writes, end), served at
`/file` with its cover at `/cover`, and loaded by the default media receiver
as a picture or as a music track. The page scales and re-encodes a photo
first (cast-session.js); a track goes as it is stored.

The first half runs the desktop relay itself, as test_cast_subtitles.py does:
it is plain Node with no dependencies. The file is behind the same token as
the stream, its address changes with every file — a receiver handed the same
URL twice shows what it already has — ranges are honoured, because that is how
a receiver seeks in a track, and only a picture or a sound is served as one.

The second half holds the wiring: what the receiver loads is the shell's to
say, never the page's; Android accepts no URL but the relay's own; a
television chosen anywhere is the one a film opens on, with the player as its
remote from the start; and the music bar drives the television once one is
chosen.
"""

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

import pytest

PACKAGES = Path(__file__).resolve().parents[2]
STATIC = PACKAGES / "meshbay-hub" / "src" / "meshbay_hub" / "static"
DESKTOP = PACKAGES / "meshbay-client" / "src"
RELAY = DESKTOP / "cast-relay.js"
CHROMECAST = DESKTOP / "cast-chromecast.js"
ANDROID_MAIN = PACKAGES / "meshbay-android" / "app" / "src" / "main"
ANDROID_SRC = ANDROID_MAIN / "kotlin"
CLIENT = ANDROID_SRC / "org" / "meshbay" / "client"
CAST = CLIENT / "cast"
SHIM = ANDROID_MAIN / "assets" / "bridge" / "meshbay-bridge.js"


def _read(path: Path) -> str:
    return path.read_text(encoding="utf-8")


RELAY_SCRIPT = r"""
const http = require('node:http');
const CastRelay = require(process.argv[2]);
const get = (url, headers = {}) => new Promise((resolve, reject) => {
  http.get(url, { headers }, (res) => {
    const parts = [];
    res.on('data', (c) => parts.push(c));
    res.on('end', () => resolve({ status: res.statusCode, headers: res.headers,
                                  body: Buffer.concat(parts).toString('hex') }));
  }).on('error', reject);
});
const refused = async (f) => { try { await f(); return false; } catch { return true; } };
const send = async (relay, bytes, opts) => {
  await relay.beginFile({ size: bytes.length, ...opts });
  for (let at = 0; at < bytes.length; at += 3) relay.writeFile(bytes.subarray(at, at + 3));
  return relay.endFile();
};
(async () => {
  const relay = new CastRelay();
  const out = {};
  out.activeBefore = relay.active;
  out.htmlRefused = await refused(() => relay.beginFile({ type: 'text/html', size: 3 }));
  out.activeAfterRefusal = relay.active;

  const jpeg = Buffer.from([0xff, 0xd8, 1, 2, 0xff, 0xd9]);
  const first = await send(relay, jpeg, { type: 'image/jpeg' });
  out.activeAfterPhoto = relay.active;
  out.firstUrl = first.url;
  out.photo = await get(first.url);
  out.badToken = (await get(first.url.replace(/t=[0-9a-f]+/, 't=x'))).status;
  out.photoMeta = relay.fileMeta;

  // A track, its cover and what to show with it; read back whole and by range.
  const track = Buffer.from([...Array(20).keys()]);
  const second = await send(relay, track, {
    type: 'audio/flac', title: 'A Song', artist: 'Some Band', album: 'An Album',
    cover: { bytes: Buffer.from([0xff, 0xd8, 7, 0xff, 0xd9]), type: 'image/jpeg' } });
  out.secondUrl = second.url;
  out.trackType = relay.fileType;
  out.trackMeta = relay.fileMeta;
  out.whole = await get(second.url);
  out.range = await get(second.url, { Range: 'bytes=5-9' });
  out.tail = await get(second.url, { Range: 'bytes=15-' });
  out.beyond = (await get(second.url, { Range: 'bytes=50-60' })).status;
  out.cover = await get(relay.fileMeta.cover);

  // Fewer bytes than announced is not a file; more is refused as they come.
  await relay.beginFile({ type: 'audio/mpeg', size: 10 });
  relay.writeFile(Buffer.from([1, 2, 3]));
  out.shortRefused = await refused(() => relay.endFile());
  await relay.beginFile({ type: 'audio/mpeg', size: 2 });
  out.longRefused = await refused(() => relay.writeFile(Buffer.from([1, 2, 3])));
  out.stillServed = (await get(second.url)).status;

  await relay.start({});
  out.fileAfterFilm = relay.fileUrl;
  await send(relay, jpeg, { type: 'image/jpeg' });
  await relay.stop();
  out.fileAfterStop = relay.fileUrl;
  console.log(JSON.stringify(out));
})().catch((e) => { console.error(e); process.exit(1); });
"""


@pytest.fixture(scope="module")
def relay_run(tmp_path_factory):
    if shutil.which("node") is None or not RELAY.exists():
        pytest.skip("node or the desktop client sources are not available")
    script = tmp_path_factory.mktemp("cf") / "relay.cjs"
    script.write_text(RELAY_SCRIPT, encoding="utf-8")
    proc = subprocess.run(["node", str(script), str(RELAY)],
                          capture_output=True, text=True, encoding="utf-8", timeout=30)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_a_file_starts_the_relay_and_is_served_behind_its_token(relay_run):
    assert relay_run["activeBefore"] is False
    assert relay_run["activeAfterPhoto"] is True
    photo = relay_run["photo"]
    assert photo["status"] == 200
    assert photo["headers"]["content-type"] == "image/jpeg"
    assert photo["headers"]["cache-control"] == "no-store"
    assert photo["body"] == "ffd80102ffd9"
    assert relay_run["badToken"] == 403
    assert relay_run["photoMeta"]["cover"] is None


def test_every_file_has_its_own_address(relay_run):
    assert re.search(r"/file\?t=[0-9a-f]{32}&v=\d+$", relay_run["firstUrl"])
    assert relay_run["secondUrl"] != relay_run["firstUrl"]


def test_a_track_is_served_with_its_cover_and_what_it_is(relay_run):
    assert relay_run["trackType"] == "audio/flac"
    meta = relay_run["trackMeta"]
    assert (meta["title"], meta["artist"], meta["album"]) == ("A Song", "Some Band", "An Album")
    assert re.search(r"/cover\?t=[0-9a-f]{32}&v=\d+$", meta["cover"])
    assert relay_run["cover"]["status"] == 200
    assert relay_run["cover"]["body"] == "ffd807ffd9"
    assert relay_run["whole"]["body"] == bytes(range(20)).hex()
    assert relay_run["whole"]["headers"]["accept-ranges"] == "bytes"


def test_a_receiver_seeks_in_a_track_by_range(relay_run):
    r = relay_run["range"]
    assert r["status"] == 206
    assert r["headers"]["content-range"] == "bytes 5-9/20"
    assert r["body"] == bytes(range(5, 10)).hex()
    tail = relay_run["tail"]
    assert tail["status"] == 206 and tail["body"] == bytes(range(15, 20)).hex()
    assert relay_run["beyond"] == 416


def test_only_a_whole_picture_or_sound_is_served(relay_run):
    assert relay_run["htmlRefused"] is True
    assert relay_run["activeAfterRefusal"] is False, "a refused file starts nothing"
    assert relay_run["shortRefused"] is True
    assert relay_run["longRefused"] is True
    assert relay_run["stillServed"] == 200, "a failed send leaves the shown file alone"


def test_a_film_or_a_stop_takes_the_file_down(relay_run):
    assert relay_run["fileAfterFilm"] is None
    assert relay_run["fileAfterStop"] is None


def test_the_receiver_loads_a_photo_as_a_picture_and_a_track_as_music():
    if shutil.which("node") is None or not CHROMECAST.exists():
        pytest.skip("node or the desktop client sources are not available")
    script = (
        "const Module = require('node:module');"
        "const real = Module._load;"
        "Module._load = function (r, ...a) {"
        "  if (r === 'bonjour-service') return { Bonjour: class {} };"
        "  if (r === 'castv2-client') return { Client: class {}, DefaultMediaReceiver: {} };"
        "  return real.call(this, r, ...a); };"
        f"const {{ mediaFor, loadOptionsFor }} = require({json.dumps(str(CHROMECAST))});"
        "const meta = { title: 'A Song', artist: 'Some Band', album: 'An Album',"
        "               cover: 'http://x/cover?t=a&v=2' };"
        "console.log(JSON.stringify({"
        "  photo: mediaFor('http://x/file?t=a&v=1', null,"
        "                  { contentType: 'image/jpeg', meta: { cover: null } }),"
        "  track: mediaFor('http://x/file?t=a&v=2', null, { contentType: 'audio/flac', meta }),"
        "  from: loadOptionsFor(null, 42.5),"
        "  film: mediaFor('http://x/stream.mp4?t=a', null, null),"
        "  filmOptions: loadOptionsFor(null, 0),"
        "  legacy: mediaFor('http://x/stream.mp4?t=a', null) }));")
    proc = subprocess.run(["node", "-e", script], capture_output=True, text=True,
                          encoding="utf-8", timeout=30)
    assert proc.returncode == 0, proc.stderr
    out = json.loads(proc.stdout)
    assert out["photo"]["contentType"] == "image/jpeg"
    assert out["photo"]["streamType"] == "NONE"
    assert "tracks" not in out["photo"]
    track = out["track"]
    assert track["contentType"] == "audio/flac" and track["streamType"] == "BUFFERED"
    assert track["metadata"] == {"metadataType": 3, "title": "A Song", "artist": "Some Band",
                                 "albumName": "An Album",
                                 "images": [{"url": "http://x/cover?t=a&v=2"}]}
    assert out["from"]["currentTime"] == 42.5
    assert "currentTime" not in out["filmOptions"]
    assert out["film"] == out["legacy"], "a film is loaded exactly as before"
    assert out["film"]["streamType"] == "LIVE" and out["film"]["contentType"] == "video/mp4"


def test_the_shell_not_the_page_says_what_the_receiver_loads():
    """The page names a URL; what it is, is the relay's to say. A page that
    could name the content type could have a television load the stream as
    anything at all."""
    main = _read(DESKTOP / "main.js")
    load_of = main.split("const loadOf = ", 1)[1].split("\n\n", 1)[0]
    assert "mediaUrl === castRelay.fileUrl" in load_of
    assert "contentType: castRelay.fileType, meta: castRelay.fileMeta" in load_of
    assert "subtitle: null" in load_of, "a file carries no subtitle"
    assert main.count("loadOf(mediaUrl, subtitle)") == 2, "connect and reload both"
    # A start position means something for a file only; a film starts where
    # its relay does.
    assert main.count("l.file ? startAt : 0") == 2


def test_android_accepts_the_relays_file_and_nothing_else():
    if not CAST.exists():
        pytest.skip("android sources not present")
    channels = _read(CAST / "CastChannels.kt")
    check = channels.split("private fun relayUrl(", 1)[1].split("\n    }", 1)[0]
    assert "asked != ours && asked != relay.fileUrl" in check
    assert 'put("contentType", relay.fileType).put("meta", relay.fileMeta()' in channels
    assert "url == relay.fileUrl -> null" in channels, "a file carries no subtitle"
    assert 'if (url == relay.fileUrl) maxOf(0.0, o.optDouble("startAt", 0.0)) else 0.0' in channels
    # A file is not a cast that has to survive the screen going off.
    begin = channels.split('"cast:file:begin" ->', 1)[1].split('"cast:file:end"', 1)[0]
    assert "onCasting" not in begin
    relay = _read(CAST / "CastRelay.kt")
    assert '"/file" -> serveFile(out, socket, range)' in relay
    assert '"/cover" -> serveCover(out, socket)' in relay
    serve = relay.split("private fun serve(", 1)[1]
    assert serve.index('query["t"] != token') < serve.index('"/file"'), "behind the token"
    # On disk, never held: a long lossless track is a few hundred megabytes.
    assert 'File(spoolDir, "file-${fileSeq.incrementAndGet()}.bin")' in relay
    assert "java.io.FileOutputStream(slot.path)" in relay
    control = _read(CAST / "CastControl.kt")
    assert "MediaInfo.STREAM_TYPE_NONE" in control
    assert "MediaMetadata.MEDIA_TYPE_PHOTO" in control
    assert "MediaMetadata.MEDIA_TYPE_MUSIC_TRACK" in control
    assert "MediaInfo.STREAM_TYPE_BUFFERED" in control
    assert "setCurrentTime((startAt * 1000).toLong())" in control
    assert "MediaSeekOptions.Builder().setPosition(" in control
    # Loaded is shown: a photo never reaches PLAYING, and waiting for it
    # would hold every photo for the full fifteen seconds.
    load = control.split("private fun load(", 1)[1]
    assert load.index("if (photo) return") < load.index("PLAY_WAIT_MS")


def test_a_file_crosses_the_android_bridge_in_binary_pieces():
    """One message the size of a lossless track would be the WebView's to
    refuse; a file goes in pieces of a megabyte, as binary frames where the
    WebView has them."""
    if not SHIM.exists():
        pytest.skip("android sources not present")
    shim = _read(SHIM)
    assert "const CAST_FILE = 3;" in shim
    assert "const FILE_PIECE = 1024 * 1024;" in shim
    assert ("BINARY ? sendBinary(CAST_FILE, 0, piece) : call('cast:file:write', base64Of(piece))"
            in shim)
    assert "const val CAST_FILE = 3" in _read(CLIENT / "save" / "SaveNames.kt")
    channels = _read(CLIENT / "bridge" / "Channels.kt")
    assert "BinaryFrame.CAST_FILE -> " in channels and ".writeFile(frame.bytes" in channels


def test_a_television_chosen_anywhere_takes_the_next_film():
    """The player opens as the remote of a television chosen before it, the
    cast pending until the first segment exactly as a restart is; choosing one
    in the player keeps it for the session, and stopping there forgets it."""
    player = _read(STATIC / "video-player.js")
    assert ("const [autoCast] = useState(() => "
            "(castSession.supported ? castSession.device : null));") in player
    assert "useState(Boolean(autoCast))" in player
    assert "castActiveRef = useRef(Boolean(autoCast))" in player
    assert "castRestartPendingRef = useRef(Boolean(autoCast))" in player
    assert "castDeviceRef = useRef(autoCast)" in player
    assert "useState(autoCast ? autoCast.name : null)" in player
    stop = player.split("const stopCast = async () => {", 1)[1].split("};", 1)[0]
    assert "castSession.forget()" in stop
    assert "castSession.choose(d);" in player
    assert "castSession.claim('video')" in player


def test_the_cast_button_is_where_the_media_are():
    """Videos' toolbar, the Photos landing, a photo album's bar and Music's
    toolbar are drawn by the group page and the search page alike, so one
    button in each covers both. On the Photos landing the button keeps a
    toolbar Search would otherwise hide. The music bar has its own, opening
    upward from the bottom of the window."""
    videos = _read(STATIC / "video-app.js")
    assert "<${CastButton} />" in videos.split('<div class="video-toolbar">', 1)[1]
    photos = _read(STATIC / "photos-app.js")
    landing = photos.split("function PhotosApp(", 1)[1]
    assert "(!hideFilter || castSession.supported) && html`" in landing
    assert "<${CastButton} />" in landing.split('<div class="photo-toolbar">', 1)[1]
    album = photos.split("function AlbumView(", 1)[1]
    assert "<${CastButton}" in album
    lightbox = photos.split("function Lightbox(", 1)[1].split("function AlbumView(", 1)[0]
    assert '<${CastButton} variant="overlay"' in lightbox
    assert "if (castDevice && blobUrl && blob) showPhoto(blob);" in lightbox
    music = _read(STATIC / "music-app.js")
    assert "<${CastButton} />" in music.split('<div class="video-toolbar">', 1)[1]
    bar = _read(STATIC / "music-player.js")
    assert '<${CastButton} cls="music-player-btn" up />' in bar
    # The group page draws them through the app registry, the search page
    # imports them: the same components either way.
    registry = _read(STATIC / "apps.js")
    search = _read(STATIC / "search-page.js")
    for page in (registry, search):
        for app in ("VideoApp", "PhotosApp", "MusicApp"):
            assert app in page


def test_the_music_bar_drives_the_television():
    """With a television chosen, the decrypted track goes to the relay instead
    of the element, and the bar's controls and clock are the receiver's. The
    receiver keeps saying FINISHED until it is given something else, so the
    end of a track moves the queue on once; and a television a film or a
    photo has taken is not the bar's to read."""
    bar = _read(STATIC / "music-player.js")
    assert "const remote = Boolean(castDevice) && castSession.canSendFiles;" in bar
    send = bar.split("const sendToTv = useCallback(", 1)[1].split("}, [fetchCover]);", 1)[0]
    assert "owner: 'music', startAt," in send
    assert "type: cached.mime, cover," in send
    assert "sendingRef.current = true;" in send
    load = bar.split("if (remoteRef.current) {\n          // To the television", 1)[1]
    assert "audio.pause();" in load.split("await audio.play();", 1)[0]
    tick = bar.split("const tick = async () => {", 1)[1].split("const id = setInterval", 1)[0]
    assert "castSession.owner !== 'music'" in tick
    assert "sendingRef.current" in tick
    assert "finishedRef.current !== remoteTrackRef.current" in tick
    assert "String(c.idleReason || '').toLowerCase() === 'finished'" in tick
    # The cover is found as the album card finds it, not only on the track.
    cover = bar.split("const fetchCover = useCallback(", 1)[1]
    cover = cover.split("}, [getConnection, tracks]);", 1)[0]
    assert "entry.thumb_hash || (sibling && sibling.thumb_hash)" in cover
    assert "transport.fetchMusicMeta(entry.id)" in cover
    assert "meta.cover_thumb_hash" in cover
    assert "castBridge.chromecastSeek(to)" in bar
    assert "castBridge.chromecastPause() : castBridge.chromecastPlay()" in bar
    # Stopping the cast carries the track on here; a film taking the
    # television pauses the bar.
    assert "if (castSession.device && castSession.owner !== 'music') setPlaying(false);" in bar


def test_a_photo_leaves_scaled_and_upright():
    session = _read(STATIC / "cast-session.js")
    assert "imageOrientation: 'from-image'" in session
    assert "const TV_WIDTH = 1920;" in session and "const TV_HEIGHT = 1080;" in session
    assert "Math.min(1, TV_WIDTH / bitmap.width, TV_HEIGHT / bitmap.height)" in session
    assert "await loadFile({ owner: 'photo', bytes, type: 'image/jpeg' });" in session
    # Only the newest photo asked for is sent.
    assert session.count("gen !== photoGen") >= 2


def test_every_locale_names_the_slideshow():
    for f in sorted((STATIC / "locales").glob("*.js")):
        text = f.read_text(encoding="utf-8")
        for key in ("photo.slideshow", "photo.slideshow_stop"):
            assert f"'{key}':" in text, f"{f.name} lacks {key}"