"""
Casting in the Android application, pinned by reading the source.

The relay is a port of meshbay-client/src/cast-relay.js and its mitigations are
the same ones; the JVM tests (CastRelayTest) run it on loopback. What is held
here is the wiring around it: the same bridge surface as the desktop, order
kept where order is meaning, the receiver pointed at nothing but the relay, and
what keeps a cast alive with the screen off switched on only while one runs.
"""

import re
from pathlib import Path

import pytest

PACKAGES = Path(__file__).resolve().parents[2]
MAIN = PACKAGES / "meshbay-android" / "app" / "src" / "main"
SRC = MAIN / "kotlin" / "org" / "meshbay" / "client"
CAST = SRC / "cast"
SHIM = MAIN / "assets" / "bridge" / "meshbay-bridge.js"
DESKTOP = PACKAGES / "meshbay-client" / "src"

pytestmark = pytest.mark.skipif(not CAST.exists(), reason="android sources not present")


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


def test_the_cast_channels_are_the_desktops():
    def channels(text: str, call: str) -> set[str]:
        return set(re.findall(rf"{call}\('(cast:[\w:-]+)'", text))

    preload = channels(_read(DESKTOP / "preload.js"), r"ipcRenderer\.invoke")
    shim = channels(_read(SHIM), "call")
    kt = _read(CAST / "CastChannels.kt")
    native = set(re.findall(r'^\s*"(cast:[\w:-]+)" ->', kt, flags=re.M))
    assert preload and shim == preload, shim ^ preload
    assert native == preload, native ^ preload


def test_no_cast_object_where_casting_cannot_work():
    """`platform.cast.available` is whether the object exists."""
    shim = _read(SHIM)
    assert "...(CAST ? { cast: {" in shim
    assert "lanCast: CAST" in shim
    assert "const CAST = ${cast.control.available()}" in _read(SRC / "MainActivity.kt")


def test_the_relay_keeps_the_desktop_mitigations():
    relay = _read(CAST / "CastRelay.kt")
    desktop = _read(DESKTOP / "cast-relay.js")
    for name in ("RING_CAP", "PORT_BASE", "PORT_COUNT"):
        js = re.search(rf"const {name} = (\d+);", desktop).group(1)
        assert re.search(rf"const val {name} = {js}\b", relay), name
    assert "InetSocketAddress(address, PORT_BASE + i)" in relay, "bound to the LAN address"
    # Code, not comments: the comment saying "never 0.0.0.0" is not a bind.
    code = re.sub(r"/\*.*?\*/", "", relay, flags=re.S)
    code = re.sub(r"(?m)//.*$", "", code)
    assert "0.0.0.0" not in code
    assert 'query["t"] != token' in relay
    # The preflight before the token: a refusal there reads as a network error.
    serve = relay.split("private fun serve(", 1)[1]
    assert serve.index('method == "OPTIONS"') < serve.index('query["t"] != token')


def test_what_a_receiver_has_not_read_waits_on_disk():
    """The desktop drops a fragment once 8 MB wait for a receiver; a fragment
    is 5 to 10 MB at a film's bitrate, so the television froze for its length
    (measured: three dropped in the first fifteen seconds). Here the lead waits
    in a spool file per receiver, deleted with it, and is dropped only past a
    disk bound."""
    relay = _read(CAST / "CastRelay.kt")
    assert "BACKPRESSURE_HIGH" not in relay
    assert "RandomAccessFile(file, \"rw\").channel" in relay
    assert "c.waiting() > spoolLimit()" in relay
    assert "SPOOL_MAX_BYTES = 2L * 1024 * 1024 * 1024" in relay
    assert "file.delete()" in relay
    assert 'java.io.File(context.cacheDir, "cast-relay")' in _read(CAST / "CastChannels.kt")


def test_the_backlog_is_bounded_in_bytes():
    """A fragment is a segment, megabytes at a film's bitrate: 64 of them killed
    the application with an OutOfMemoryError on the emulator."""
    relay = _read(CAST / "CastRelay.kt")
    assert "RING_MAX_BYTES" in relay and "ringBytes > RING_MAX_BYTES" in relay


def test_the_relay_is_on_wifi_never_the_mobile_network():
    channels = _read(CAST / "CastChannels.kt")
    lan = channels.split("private fun lanAddress()", 1)[1]
    assert "TRANSPORT_WIFI" in lan and "TRANSPORT_ETHERNET" in lan
    assert "TRANSPORT_CELLULAR" not in lan


def test_the_receiver_is_pointed_at_the_relay_and_nothing_else():
    channels = _read(CAST / "CastChannels.kt")
    check = channels.split("private fun relayUrl(", 1)[1].split("\n    }", 1)[0]
    assert "asked != ours" in check and "throw Refused" in check
    assert channels.count("relayUrl(o)") == 2, "connect and reload both go through the check"


def test_relay_calls_keep_their_order():
    """The page does not await each push; on a pool they could overtake."""
    channels = _read(CAST / "CastChannels.kt")
    assert '"cast:start", "cast:push", "cast:subtitle", "cast:finish", "cast:stop"' in channels
    bridge = _read(SRC / "bridge" / "Bridge.kt")
    assert "Executors.newSingleThreadExecutor()" in bridge
    assert "(if (ordered) serial else work).execute" in bridge
    assert "fun ordered(frame: BinaryFrame) = frame.channel == BinaryFrame.CAST_PUSH" in _read(
        SRC / "bridge" / "Channels.kt")


def test_screen_off_survival_is_switched_on_only_while_casting():
    activity = _read(SRC / "MainActivity.kt")
    casting = activity.split("private fun casting(on: Boolean)", 1)[1].split("\n    }", 1)[0]
    assert "web.keepVisible = on" in casting
    assert "startForegroundService(service) else stopService(service)" in casting
    assert activity.count("keepVisible =") == 1, "the page is kept visible from one place only"
    view = _read(SRC / "shell" / "ShellWebView.kt")
    assert "var keepVisible = false" in view
    manifest = _read(MAIN / "AndroidManifest.xml")
    assert 'android:name=".cast.CastService"' in manifest
    assert 'android:foregroundServiceType="mediaPlayback"' in manifest


def test_the_receiver_is_given_what_the_desktop_gives_it():
    control = _read(CAST / "CastControl.kt")
    assert "DEFAULT_MEDIA_RECEIVER_APPLICATION_ID" in control
    assert "MediaInfo.STREAM_TYPE_LIVE" in control
    assert "TEXT_TRACK_ID = 1L" in control
    assert "SCAN_DURATION_MS = 6000L" in control
    assert re.search(r"const SCAN_DURATION_MS = 6000;", _read(DESKTOP / "cast-chromecast.js"))