From 57758895bc7b0f7f21497194e81172f43d634852 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sat, 3 Oct 2026 09:03:34 +0200 Subject: feat(android): casting through a LAN relay and the platform cast SDK A port of cast-relay.js (backlog also bounded in bytes), discovery and control with the default media receiver, relay calls kept in order, and a foreground service plus a WebView kept visible so a cast survives the screen going off. Co-Authored-By: Claude Opus 5.5 --- packages/meshbay-hub/tests/test_android_cast.py | 119 +++++++++++++++++++++++ packages/meshbay-hub/tests/test_android_shell.py | 5 +- 2 files changed, 122 insertions(+), 2 deletions(-) create mode 100644 packages/meshbay-hub/tests/test_android_cast.py (limited to 'packages/meshbay-hub') diff --git a/packages/meshbay-hub/tests/test_android_cast.py b/packages/meshbay-hub/tests/test_android_cast.py new file mode 100644 index 0000000..0ff383a --- /dev/null +++ b/packages/meshbay-hub/tests/test_android_cast.py @@ -0,0 +1,119 @@ +""" +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 "BACKPRESSURE_HIGH = 8L * 1024 * 1024" in relay + 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_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")) diff --git a/packages/meshbay-hub/tests/test_android_shell.py b/packages/meshbay-hub/tests/test_android_shell.py index ef7355d..e569bb7 100644 --- a/packages/meshbay-hub/tests/test_android_shell.py +++ b/packages/meshbay-hub/tests/test_android_shell.py @@ -134,8 +134,9 @@ def test_the_shim_offers_desktop_channels_and_native_answers_each(): preload_js = _read(CLIENT / "src" / "preload.js") preload = set(re.findall(r"ipcRenderer\.invoke\('([\w:-]+)'", preload_js)) native = set() - for name in ("Channels.kt", "KeyChannels.kt"): - native |= set(re.findall(r'^\s*"([\w:-]+)" ->', _read(SRC / "bridge" / name), flags=re.M)) + for path in (SRC / "bridge" / "Channels.kt", SRC / "bridge" / "KeyChannels.kt", + SRC / "cast" / "CastChannels.kt"): + native |= set(re.findall(r'^\s*"([\w:-]+)" ->', _read(path), flags=re.M)) shim = _shim_channels() assert shim, "no channel found in the shim" assert shim <= preload, f"channels the desktop does not have: {shim - preload}" -- cgit v1.2.3