#!/usr/bin/env python3 """Run deterministic Multilang browser checks against JF12 and restore its full state. The runner creates a temporary local-NFO movie library, seeds only its real item IDs into Multilang's SQLite database, runs Playwright, then restores and hashes the JF12 config, data, cache, and injected web index. It refuses any target other than the disposable jellyfin12 instance. """ from __future__ import annotations import argparse import base64 import json import os import shlex import subprocess import sys import time import uuid from pathlib import Path from urllib.parse import urlencode from run_jellyfin12_live_tests import JellyfinApi, RemoteState, wait_for_login FIXTURE_LIBRARY = "Multilang Browser Fixtures" FIXTURE_PATH = "/config/multilang-browser-fixtures/media" FIXTURES = ( ("The Anchor", "Ankkuri", "Ankkurin suomenkielinen kuvaus", "Draama"), ("The Beacon", "Majakka", "Majakan suomenkielinen kuvaus", "Komedia"), ("The Clock", None, None, "Draama"), ) def request_empty(api: JellyfinApi, method: str, path: str, payload: dict | None = None) -> None: body = None if payload is None else json.dumps(payload).encode() response = api.request(method, path, body, {"Content-Type": "application/json"} if body else None) if not 200 <= response.status < 300: raise RuntimeError(f"{method} {path} returned HTTP {response.status}: {response.body[:500].decode(errors='replace')}") def write_fixture_files(remote: RemoteState) -> None: host_fixture_path = f"{remote.args.state_root.rstrip('/')}/config/multilang-browser-fixtures/media" files: dict[str, str] = {} for index, (title, _, _, genre) in enumerate(FIXTURES, start=1): directory = title.replace(" ", "_") nfo = f""" {title} {title} {title} English overview for {title}. English tagline for {title}. 2024 2024-01-{index:02d} {'Drama' if genre == 'Draama' else 'Comedy'} en 99000{index} """ files[f"{directory}/movie.nfo"] = nfo files[f"{directory}/{title}.mkv"] = "" commands = [f"mkdir -p {shlex.quote(host_fixture_path)}"] for relative, contents in files.items(): target = f"{host_fixture_path}/{relative}" encoded = base64.b64encode(contents.encode()).decode() commands.append(f"mkdir -p {shlex.quote(str(Path(target).parent))}") commands.append(f"printf %s {shlex.quote(encoded)} | base64 -d > {shlex.quote(target)}") remote.run("bash -lc " + shlex.quote("; ".join(commands))) def wait_for_fixture_items(api: JellyfinApi, user_id: str) -> tuple[str, dict[str, str]]: for _ in range(90): views = api.json("GET", f"/Users/{user_id}/Views").get("Items", []) library = next((view for view in views if view.get("Name") == FIXTURE_LIBRARY), None) if library: items = api.json( "GET", f"/Users/{user_id}/Items?{urlencode({'ParentId': library['Id'], 'Recursive': 'true', 'IncludeItemTypes': 'Movie'})}", ).get("Items", []) ids = {item.get("Name"): item.get("Id") for item in items} if all(title in ids for title, *_ in FIXTURES): return library["Id"], {title: ids[title] for title, *_ in FIXTURES} time.sleep(1) raise RuntimeError("JF12 did not index every deterministic browser fixture within 90 seconds.") def create_fixture_library(api: JellyfinApi, user_id: str, remote: RemoteState) -> tuple[str, dict[str, str]]: write_fixture_files(remote) request_empty( api, "POST", "/Library/VirtualFolders?" + urlencode({ "name": FIXTURE_LIBRARY, "collectionType": "movies", "paths": FIXTURE_PATH, "refreshLibrary": "true", }), {"LibraryOptions": {"EnableRealtimeMonitor": False, "EnableInternetProviders": False}}, ) return wait_for_fixture_items(api, user_id) def seed_translations(remote: RemoteState, item_ids: dict[str, str]) -> None: rows = [] for index, (title, translated, overview, genre) in enumerate(FIXTURES, start=1): rows.append({ "item_id": item_ids[title].replace("-", "").lower(), "tmdb_id": f"99000{index}", "original_title": title, "original_language": "sv" if title == "The Beacon" else "en", "translations": {"title": translated, "overview": overview, "tagline": translated and f"{translated} - suomenkielinen iskulause"}, "genre_id": 18 if genre == "Draama" else 35, "genre": genre, }) encoded = base64.b64encode(json.dumps(rows).encode()).decode() script = r''' import base64, json, os, sqlite3, sys, time rows = json.loads(base64.b64decode(sys.argv[1])) database = sys.argv[2] connection = sqlite3.connect(database) now = int(time.time()) for row in rows: connection.execute(""" INSERT INTO facts(item_id, tmdb_id, kind, original_title, original_language, original_language_all, origin_countries_json, production_countries_json, spoken_languages_json, audio_track_language, genre_tmdb_ids_json, missing_checked_at, full_checked_at) VALUES (?, ?, 'movie', ?, ?, ?, '[\"US\"]', '[\"US\"]', '[\"en\"]', 'en', ?, ?, ?) ON CONFLICT(item_id) DO UPDATE SET tmdb_id=excluded.tmdb_id, kind=excluded.kind, original_title=excluded.original_title, original_language=excluded.original_language, original_language_all=excluded.original_language_all, genre_tmdb_ids_json=excluded.genre_tmdb_ids_json, missing_checked_at=excluded.missing_checked_at, full_checked_at=excluded.full_checked_at """, (row["item_id"], row["tmdb_id"], row["original_title"], row["original_language"], row["original_language"], json.dumps([row["genre_id"]]), now, now)) for field, value in row["translations"].items(): if value: connection.execute(""" INSERT INTO translations(item_id, lang, field, text) VALUES (?, 'fi', ?, ?) ON CONFLICT(item_id, lang, field) DO UPDATE SET text=excluded.text """, (row["item_id"], field, value)) connection.execute(""" INSERT INTO genres(tmdb_id, media, lang, name, name_norm) VALUES (?, 'movie', 'fi', ?, ?) ON CONFLICT(tmdb_id, media, lang) DO UPDATE SET name=excluded.name, name_norm=excluded.name_norm """, (row["genre_id"], row["genre"], row["genre"].lower())) relative = row["item_id"] + "/fixture.png" destination = os.path.join(os.path.dirname(database), "assets", relative) os.makedirs(os.path.dirname(destination), exist_ok=True) with open(destination, "wb") as image: image.write(base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jZ1kAAAAASUVORK5CYII=")) path = "/Multilang/Assets/" + relative connection.execute("INSERT OR REPLACE INTO assets(item_id,lang,kind,path,path_low,updated_at) VALUES (?, 'fi', 'poster', ?, ?, ?)", (row["item_id"], path, path.lower(), now)) connection.commit() ''' remote.stop() try: database = f"{remote.args.state_root.rstrip('/')}/data/multilang/translations.sqlite" remote.run("python3 -c " + shlex.quote(script) + " " + shlex.quote(encoded) + " " + shlex.quote(database)) finally: remote.start() def configure_multilang(api: JellyfinApi) -> None: config = api.json("GET", "/Multilang/AdminConfig") config.update({ "Languages": ["en", "fi"], "ItemsProxyCacheThresholdMs": 0, "ItemsProxyCacheTtlMinutes": 5, "ItemsProxyCacheMaxMiB": 8, }) api.json("POST", "/Multilang/AdminConfig", config) rules = api.json("GET", "/Multilang/UserRules/self") rules.update({ "Enabled": True, "SortLocale": "fi-FI", "Categories": [], "FallbackFieldActions": { "title": ["Language:fi", "Jellyfin"], "overview": ["Language:fi", "Jellyfin"], "tagline": ["Language:fi", "Jellyfin"], "poster": ["Jellyfin"], "logo": ["Jellyfin"], "banner": ["Jellyfin"], "thumb": ["Jellyfin"], "backdrop": ["Jellyfin"], }, }) api.json("POST", "/Multilang/UserRules/self", rules) api.json("POST", "/Multilang/ClearCache") def run_playwright(args: argparse.Namespace, fixture_file: Path) -> None: browser_dir = Path(__file__).resolve().parents[1] / "tests" / "browser" if not (browser_dir / "node_modules" / "@playwright" / "test").exists(): raise RuntimeError("Playwright is not installed. Run: cd tests/browser && npm install && npx playwright install chromium") environment = os.environ | { "JELLYFIN_BASE_URL": args.base_url.rstrip("/"), "MULTILANG_BROWSER_FIXTURE": str(fixture_file), } result = subprocess.run(["npm", "test"], cwd=browser_dir, env=environment) if result.returncode: raise RuntimeError("Playwright browser checks failed.") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", required=True) parser.add_argument("--username", required=True) parser.add_argument("--password", required=True) parser.add_argument("--ssh-host", required=True) parser.add_argument("--ssh-user", default="root") parser.add_argument("--local-docker", action="store_true") parser.add_argument("--container", default="jellyfin12") parser.add_argument("--state-root", required=True) parser.add_argument("--confirm-destructive", action="store_true") args = parser.parse_args() if not args.confirm_destructive: parser.error("--confirm-destructive is required") if args.container != "jellyfin12" or not args.state_root.rstrip("/").endswith("/jellyfin12"): parser.error("This runner only permits the disposable jellyfin12 instance.") if ":8097" not in args.base_url: parser.error("This runner only permits the JF12 URL on port 8097.") args.full_server_state = True return args def main() -> int: args = parse_args() remote = RemoteState(args) fixture_file = Path(__file__).resolve().parents[1] / "tests" / "browser" / ".auth" / f"fixture-{uuid.uuid4().hex}.json" success = False try: remote.verify_target() remote.snapshot() api, user_id = wait_for_login(args.base_url, args.username, args.password) library_id, item_ids = create_fixture_library(api, user_id, remote) seed_translations(remote, item_ids) api, user_id = wait_for_login(args.base_url, args.username, args.password) configure_multilang(api) fixture_file.parent.mkdir(parents=True, exist_ok=True) fixture_file.write_text(json.dumps({ "username": args.username, "password": args.password, "token": api.token, "userId": user_id, "libraryId": library_id, "jellyfinTitles": [title for title, *_ in FIXTURES], "translatedTitles": [translated or title for title, translated, *_ in FIXTURES], "orderedItemIds": [item_ids[title] for title, *_ in FIXTURES], "translatedSortedTitles": ["Ankkuri", "The Clock", "Majakka"], "translatedOrderedItemIds": [item_ids["The Anchor"], item_ids["The Clock"], item_ids["The Beacon"]], })) run_playwright(args, fixture_file) success = True finally: fixture_file.unlink(missing_ok=True) try: if remote.expected_web_hash: remote.restore_and_verify() print("Restoration hash check passed for JF12 config, data, cache, and injected web index.") remote.discard_snapshot() else: remote.discard_snapshot() except Exception as error: print(f"RESTORATION FAILED: {error}", file=sys.stderr) return 2 return 0 if success else 1 if __name__ == "__main__": raise SystemExit(main())