2 files changed, 122 insertions, 2 deletions
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}"