diff options
Diffstat (limited to 'packages/meshbay-hub/tests/test_cast_photos.py')
| -rw-r--r-- | packages/meshbay-hub/tests/test_cast_photos.py | 240 |
1 files changed, 0 insertions, 240 deletions
diff --git a/packages/meshbay-hub/tests/test_cast_photos.py b/packages/meshbay-hub/tests/test_cast_photos.py deleted file mode 100644 index 699e663..0000000 --- a/packages/meshbay-hub/tests/test_cast_photos.py +++ /dev/null @@ -1,240 +0,0 @@ -""" -Photos on a cast receiver, and the television a session casts to. - -A photo takes a different road to the television than a film: no stream, one -picture at a time, served by the relay at `/image` and loaded as a picture by -the default media receiver. The page scales and re-encodes it first -(cast-session.js), so what reaches the relay is a JPEG the receiver can show. - -The first half runs the desktop relay itself, as test_cast_subtitles.py does: -it is plain Node with no dependencies. The photo is behind the same token as -the stream, its address changes with every photo — a receiver handed the same -URL twice shows what it already has — and only a picture is served as one. - -The second half holds the wiring: the receiver is told a photo is a photo by -the shell, never by the page; Android accepts no URL but the relay's own; and -a television chosen anywhere is the one a film opens on, with the player as -its remote from the start. -""" - -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_SRC = PACKAGES / "meshbay-android" / "app" / "src" / "main" / "kotlin" -CAST = ANDROID_SRC / "org" / "meshbay" / "client" / "cast" - - -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) => new Promise((resolve, reject) => { - http.get(url, (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); -}); -(async () => { - const relay = new CastRelay(); - const out = {}; - out.activeBefore = relay.active; - try { - await relay.showImage({ bytes: Buffer.from('<p>'), type: 'text/html' }); - out.htmlRefused = false; - } catch { out.htmlRefused = true; } - out.activeAfterRefusal = relay.active; - - const jpeg = Buffer.from([0xff, 0xd8, 1, 2, 0xff, 0xd9]); - const first = await relay.showImage({ bytes: jpeg, type: 'image/jpeg' }); - out.activeAfterPhoto = relay.active; - out.firstUrl = first.url; - out.imageType = relay.imageType; - out.photo = await get(first.url); - out.badToken = (await get(first.url.replace(/t=[0-9a-f]+/, 't=x'))).status; - const second = await relay.showImage({ bytes: Buffer.from([9]), type: 'image/jpeg' }); - out.secondUrl = second.url; - out.second = await get(second.url); - - await relay.start({}); - out.imageAfterFilm = relay.imageUrl; - await relay.stop(); - out.imageAfterStop = relay.imageUrl; - 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("cp") / "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_photo_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["imageType"] == "image/jpeg" - - -def test_every_photo_has_its_own_address(relay_run): - assert re.search(r"/image\?t=[0-9a-f]{32}&v=\d+$", relay_run["firstUrl"]) - assert relay_run["secondUrl"] != relay_run["firstUrl"] - assert relay_run["second"]["body"] == "09" - - -def test_only_a_picture_is_served_as_one(relay_run): - assert relay_run["htmlRefused"] is True - assert relay_run["activeAfterRefusal"] is False, "a refused photo starts nothing" - - -def test_a_film_or_a_stop_takes_the_photo_down(relay_run): - assert relay_run["imageAfterFilm"] is None - assert relay_run["imageAfterStop"] is None - - -def test_the_receiver_loads_a_photo_as_a_picture(): - 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 }} = require({json.dumps(str(CHROMECAST))});" - "console.log(JSON.stringify({" - " photo: mediaFor('http://x/image?t=a&v=1', null, 'image/jpeg')," - " film: mediaFor('http://x/stream.mp4?t=a', null, 'video/mp4')," - " 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"] - 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; whether it is a photo is the relay's to say. A - page that could name the content type could have a television load the - stream URL 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.imageUrl" in load_of - assert "contentType: castRelay.imageType" in load_of - assert "contentType: 'video/mp4'" in load_of - assert "subtitle: null" in load_of, "a photo carries no subtitle" - assert main.count("loadOf(mediaUrl, subtitle)") == 2, "connect and reload both" - - -def test_android_accepts_the_relays_photo_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.imageUrl" in check - assert "if (url == relay.imageUrl) relay.imageType" in channels - assert "url == relay.imageUrl -> null" in channels, "a photo carries no subtitle" - # A photo is not a cast that has to survive the screen going off. - image = channels.split('"cast:image" ->', 1)[1].split('\n "', 1)[0] - assert "onCasting" not in image - relay = _read(CAST / "CastRelay.kt") - assert '"/image" -> serveImage(out, socket)' in relay - serve = relay.split("private fun serve(", 1)[1] - assert serve.index('query["t"] != token') < serve.index('"/image"'), "behind the token" - control = _read(CAST / "CastControl.kt") - assert "MediaInfo.STREAM_TYPE_NONE" in control - assert "MediaMetadata.MEDIA_TYPE_PHOTO" 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_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 - - -def test_the_cast_button_is_where_the_media_are(): - """Videos' toolbar, the Photos landing and a photo album's bar are drawn by - the group page and the search page alike, so one button in each covers - both. On the landing the button keeps a toolbar Search would otherwise - hide.""" - 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 - # 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): - assert "VideoApp" in page and "PhotosApp" in page - - -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 "'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}" |