ajp_anton/jellyfin-multilang
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Expand integration coverage and fix settings UI

Browse Source
This commit is contained in:
ajp_anton
2026-07-21 00:42:44 +00:00
parent 25088ce91d
commit 094d5ad9f0
26 changed files with 1665 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/run_jellyfin12_live_tests.py
+405
View File
@@ -0,0 +1,405 @@
#!/usr/bin/env python3
"""Destructive Multilang integration checks with verified JF12 state restoration.
The script intentionally accepts all deployment details as arguments. It refuses
to run unless both the container and persistent state path are the JF12 test
instance. It snapshots only Multilang-owned state and the injected web index,
then restores and hashes those paths in a finally block. Run it from a separate
machine through SSH, or use --local-docker when it runs on the Docker host.
"""
from __future__ import annotations
import argparse
import json
import shlex
import subprocess
import sys
import time
import uuid
import zipfile
from dataclasses import dataclass
from io import BytesIO
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode, urlparse
from urllib.request import Request, urlopen
PLUGIN_CONFIG = "config/plugins/configurations/Jellyfin.Plugin.Multilang.xml"
PLUGIN_DATA = "data/multilang"
WEB_INDEX = "/jellyfin/jellyfin-web/index.html"
@dataclass(frozen=True)
class Response:
status: int
body: bytes
content_type: str
class JellyfinApi:
def __init__(self, base_url: str, token: str | None = None) -> None:
self.base_url = base_url.rstrip("/")
self.token = token
def request(self, method: str, path: str, body: bytes | None = None, headers: dict[str, str] | None = None) -> Response:
request_headers = dict(headers or {})
if self.token:
request_headers.setdefault("Authorization", f'MediaBrowser Token="{self.token}"')
request_headers.setdefault("X-Emby-Token", self.token)
request_headers.setdefault("X-MediaBrowser-Token", self.token)
request = Request(self.base_url + path, body, request_headers, method=method)
try:
with urlopen(request, timeout=60) as response:
return Response(response.status, response.read(), response.headers.get_content_type())
except HTTPError as error:
return Response(error.code, error.read(), error.headers.get_content_type())
def json(self, method: str, path: str, payload: Any | None = None) -> Any:
body = None if payload is None else json.dumps(payload).encode()
headers = {} if body is None else {"Content-Type": "application/json"}
response = self.request(method, path, body, headers)
if response.status < 200 or response.status >= 300:
raise RuntimeError(f"{method} {path} returned HTTP {response.status}: {response.body[:500].decode(errors='replace')}")
return json.loads(response.body)
def multipart(self, path: str, fields: dict[str, str], filename: str, contents: bytes) -> Any:
boundary = "----multilang-live-test-" + uuid.uuid4().hex
chunks: list[bytes] = []
for key, value in fields.items():
chunks.extend([
f"--{boundary}\r\n".encode(),
f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode(),
value.encode(),
b"\r\n",
])
chunks.extend([
f"--{boundary}\r\n".encode(),
f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'.encode(),
b"Content-Type: application/zip\r\n\r\n",
contents,
b"\r\n",
f"--{boundary}--\r\n".encode(),
])
response = self.request("POST", path, b"".join(chunks), {"Content-Type": f"multipart/form-data; boundary={boundary}"})
if response.status < 200 or response.status >= 300:
raise RuntimeError(f"POST {path} returned HTTP {response.status}: {response.body[:500].decode(errors='replace')}")
return json.loads(response.body)
class RemoteState:
def __init__(self, args: argparse.Namespace) -> None:
self.args = args
self.snapshot_dir = f"/tmp/multilang-live-test-{uuid.uuid4().hex}"
self.expected_hashes = ""
self.expected_web_hash = ""
def run(self, command: str, check: bool = True) -> subprocess.CompletedProcess[str]:
if self.args.local_docker:
result = subprocess.run(["bash", "-lc", command], text=True, capture_output=True)
else:
target = f"{self.args.ssh_user}@{self.args.ssh_host}"
result = subprocess.run(["ssh", target, command], text=True, capture_output=True)
if check and result.returncode:
raise RuntimeError(f"Remote command failed: {command}\n{result.stderr.strip()}")
return result
def docker(self, command: str, check: bool = True) -> subprocess.CompletedProcess[str]:
return self.run("docker " + command, check)
def stop(self) -> None:
self.docker("stop " + shlex.quote(self.args.container))
def start(self) -> None:
self.docker("start " + shlex.quote(self.args.container))
def verify_target(self) -> None:
mounts = self.docker("inspect -f '{{ range .Mounts }}{{ println .Source }}{{ end }}' " + shlex.quote(self.args.container)).stdout
root = self.args.state_root.rstrip("/") + "/"
mounted_paths = mounts.splitlines()
if not any(path.startswith(root) for path in mounted_paths):
raise RuntimeError("The requested JF12 state root does not contain any mounts for the requested container.")
for required in ("config", "data"):
if root + required not in mounted_paths:
raise RuntimeError(f"The requested JF12 state root does not mount {required} into the requested container.")
if self.args.full_server_state and root + "cache" not in mounted_paths:
raise RuntimeError("Full-state verification requires the JF12 cache mount.")
def state_paths(self) -> tuple[str, ...]:
return ("config", "data", "cache") if self.args.full_server_state else (PLUGIN_DATA, PLUGIN_CONFIG)
def hash_plugin_state(self) -> str:
root = shlex.quote(self.args.state_root)
paths = " ".join(shlex.quote(path) for path in self.state_paths())
command = (
f"cd {root} && "
f"find {paths} -type f -print0 2>/dev/null "
"| sort -z | xargs -0 -r sha256sum"
)
return self.run(command).stdout
def snapshot(self) -> None:
self.stop()
try:
root = shlex.quote(self.args.state_root)
snapshot = shlex.quote(self.snapshot_dir)
paths = " ".join(shlex.quote(path) for path in self.state_paths())
self.run(f"mkdir -p {snapshot}")
self.expected_hashes = self.hash_plugin_state()
self.run(
f"tar -C {root} --ignore-failed-read -czf {snapshot}/plugin-state.tar.gz {paths}"
)
self.docker(
"cp " + shlex.quote(self.args.container + ":" + WEB_INDEX) + " " + shlex.quote(self.snapshot_dir + "/index.html")
)
self.expected_web_hash = self.run("sha256sum " + shlex.quote(self.snapshot_dir + "/index.html")).stdout.split()[0]
finally:
self.start()
def restore_and_verify(self) -> None:
self.stop()
try:
root = shlex.quote(self.args.state_root)
snapshot = shlex.quote(self.snapshot_dir)
removals = " ".join(f"{root}/{shlex.quote(path)}" for path in self.state_paths())
self.run(
f"rm -rf {removals} && "
f"tar -C {root} -xzf {snapshot}/plugin-state.tar.gz"
)
self.docker(
"cp " + shlex.quote(self.snapshot_dir + "/index.html") + " " + shlex.quote(self.args.container + ":" + WEB_INDEX)
)
actual_hashes = self.hash_plugin_state()
restored_index = self.snapshot_dir + "/restored-index.html"
self.docker(
"cp " + shlex.quote(self.args.container + ":" + WEB_INDEX) + " " + shlex.quote(restored_index)
)
actual_web_hash = self.run("sha256sum " + shlex.quote(restored_index)).stdout.split()[0]
if actual_hashes != self.expected_hashes or actual_web_hash != self.expected_web_hash:
raise RuntimeError(
"State restoration hash mismatch. The remote snapshot was retained at " + self.snapshot_dir
)
finally:
self.start()
def discard_snapshot(self) -> None:
self.run("rm -rf " + shlex.quote(self.snapshot_dir))
def login(base_url: str, username: str, password: str) -> tuple[JellyfinApi, str]:
api = JellyfinApi(base_url)
headers = {
"Content-Type": "application/json",
"Authorization": 'MediaBrowser Client="Multilang live tests", Device="Codex", DeviceId="multilang-live-tests", Version="1.0"',
}
response = api.request("POST", "/Users/AuthenticateByName", json.dumps({"Username": username, "Pw": password}).encode(), headers)
if response.status != 200:
raise RuntimeError(f"Authentication failed: HTTP {response.status}: {response.body[:500].decode(errors='replace')}")
payload = json.loads(response.body)
return JellyfinApi(base_url, payload["AccessToken"]), payload["User"]["Id"]
def wait_for_login(base_url: str, username: str, password: str) -> tuple[JellyfinApi, str]:
for _ in range(60):
try:
return login(base_url, username, password)
except (OSError, URLError):
time.sleep(1)
continue
except RuntimeError as error:
if "HTTP 5" not in str(error):
raise
time.sleep(1)
raise RuntimeError("Jellyfin12 did not accept authentication within 60 seconds.")
def assert_equal(actual: Any, expected: Any, name: str) -> None:
if actual != expected:
raise AssertionError(f"{name}: expected {expected!r}, got {actual!r}")
def export_zip(api: JellyfinApi, query: dict[str, str]) -> bytes:
response = api.request("GET", "/Multilang/Export?" + urlencode(query))
if response.status != 200:
raise RuntimeError(f"Export failed with HTTP {response.status}: {response.body[:500].decode(errors='replace')}")
with zipfile.ZipFile(BytesIO(response.body)) as archive:
if "manifest.json" not in archive.namelist():
raise AssertionError("Admin export did not contain manifest.json.")
return response.body
def choose_library(api: JellyfinApi, user_id: str) -> tuple[str, str]:
views = api.json("GET", f"/Users/{user_id}/Views")
for view in views.get("Items", []):
collection_type = str(view.get("CollectionType", "")).lower()
if collection_type in {"movies", "tvshows"}:
return view["Id"], "Movie" if collection_type == "movies" else "Series"
raise RuntimeError("No movie or TV-show library was available for the live proxy test.")
def proxy_path(user_id: str, library_id: str, item_type: str) -> str:
upstream = (
f"/Users/{user_id}/Items?ParentId={library_id}&IncludeItemTypes={item_type}"
"&Recursive=true&SortBy=SortName&SortOrder=Ascending&StartIndex=0&Limit=100"
)
return "/Multilang/ItemsProxy?" + urlencode({"url": upstream, "mlLocale": "en-US"})
def run_checks(api: JellyfinApi, user_id: str, include_refresh: bool) -> None:
print("[1/7] Checking configuration and user-rule persistence")
config = api.json("GET", "/Multilang/AdminConfig")
rules = api.json("GET", "/Multilang/UserRules/self")
baseline_export = export_zip(api, {
"pluginSettings": "true",
"userSettings": "true",
"translationsDatabase": "false",
"downloadedAssets": "false",
})
changed_config = dict(config)
changed_config["ItemsProxyCacheThresholdMs"] = 0 if config.get("ItemsProxyCacheThresholdMs") else 1
saved_config = api.json("POST", "/Multilang/AdminConfig", changed_config)
assert_equal(saved_config["ItemsProxyCacheThresholdMs"], changed_config["ItemsProxyCacheThresholdMs"], "admin configuration save")
changed_rules = dict(rules)
changed_rules["Enabled"] = True
changed_rules["SortLocale"] = "fi-FI"
saved_rules = api.json("POST", "/Multilang/UserRules/self", changed_rules)
assert_equal(saved_rules["Enabled"], True, "user rule enable")
assert_equal(saved_rules["SortLocale"], "fi-FI", "user sort locale save")
print("[2/7] Checking user and admin backup/import endpoints")
user_export = api.request("GET", "/Multilang/UserRules/self/export")
if user_export.status != 200:
raise RuntimeError(f"User export failed with HTTP {user_export.status}")
inspected = api.multipart("/Multilang/UserRules/self/import/inspect", {}, "user-rules.zip", user_export.body)
if not inspected.get("ContainsCurrentUser"):
raise AssertionError("User export inspection did not identify the exporting user.")
imported_user = api.multipart("/Multilang/UserRules/self/import", {}, "user-rules.zip", user_export.body)
if imported_user.get("UserSettingsImported") != 1:
raise AssertionError("User export did not import exactly one rule set.")
api.multipart(
"/Multilang/Import",
{"pluginSettings": "true", "userSettings": "true", "translationsDatabase": "false", "downloadedAssets": "false"},
"admin-export.zip",
baseline_export,
)
restored_config = api.json("GET", "/Multilang/AdminConfig")
assert_equal(restored_config["ItemsProxyCacheThresholdMs"], config["ItemsProxyCacheThresholdMs"], "admin export/import round trip")
print("[3/7] Checking transformed proxy responses, cache entries, and pre-cache triggering")
enabled_rules = dict(rules)
enabled_rules["Enabled"] = True
api.json("POST", "/Multilang/UserRules/self", enabled_rules)
cached_config = dict(config)
cached_config["ItemsProxyCacheThresholdMs"] = 0
cached_config["ItemsProxyCacheTtlMinutes"] = max(1, int(cached_config.get("ItemsProxyCacheTtlMinutes", 1)))
cached_config["ItemsProxyCacheMaxMiB"] = max(1, int(cached_config.get("ItemsProxyCacheMaxMiB", 1)))
api.json("POST", "/Multilang/AdminConfig", cached_config)
api.json("POST", "/Multilang/ClearCache")
library_id, item_type = choose_library(api, user_id)
path = proxy_path(user_id, library_id, item_type)
first = api.request("GET", path)
second = api.request("GET", path)
if first.status != 200 or second.status != 200:
raise RuntimeError(f"ItemsProxy returned {first.status} then {second.status}")
first_payload = json.loads(first.body)
if not isinstance(first_payload.get("Items"), list):
raise AssertionError("ItemsProxy response did not contain an Items list.")
if not first_payload["Items"]:
raise RuntimeError("The selected library had no items for the cleanup probe.")
probe_item_id = first_payload["Items"][0]["Id"]
requests = api.json("GET", "/Multilang/ItemsProxyRequests")
if not any(request.get("CacheHit") for request in requests):
raise AssertionError("Second proxy request did not produce a cache hit.")
entries = api.json("GET", "/Multilang/CacheEntries")
if not entries:
raise AssertionError("ItemsProxy did not store a cache entry at a zero-millisecond threshold.")
print("[4/7] Checking concurrent proxy requests remain valid")
api.json("POST", "/Multilang/ClearCache")
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
responses = list(pool.map(lambda _: api.request("GET", path), range(2)))
if any(response.status != 200 for response in responses):
raise AssertionError("Concurrent proxy request did not return HTTP 200.")
if include_refresh:
print("[5/7] Checking one-item refresh and debug data")
refreshed = api.json("POST", f"/Multilang/RefreshItem/{probe_item_id}?includeChildren=false")
assert_equal(refreshed["ItemId"].lower(), probe_item_id.replace("-", "").lower(), "single-item refresh")
debug = api.json("GET", f"/Multilang/Debug/{probe_item_id}")
assert_equal(debug["ItemId"].lower(), probe_item_id.replace("-", "").lower(), "debug item identity")
else:
print("[5/7] Skipping the provider-backed refresh probe (pass --include-refresh to run it)")
print("[6/7] Checking cleanup endpoint and default-state recovery")
api.json("POST", "/Multilang/CleanupAll", {"Confirm": True})
cleanup_info = api.json("GET", "/Multilang/ExportInfo")
if cleanup_info["AssetsBytes"] != 0:
raise AssertionError("CleanupAll did not remove local assets.")
default_rules = api.json("GET", "/Multilang/UserRules/self")
if default_rules.get("Enabled"):
raise AssertionError("CleanupAll did not remove the user rule document.")
if api.json("GET", f"/Multilang/Debug/{probe_item_id}").get("Facts") is not None:
raise AssertionError("CleanupAll did not remove stored item facts.")
print("[7/7] Live checks passed; raw restoration will now run")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-url", required=True, help="JF12 base URL, expected to use port 8097")
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", help="Run Docker and filesystem commands on this host instead of through SSH")
parser.add_argument("--container", default="jellyfin12")
parser.add_argument("--state-root", required=True, help="JF12 persistent volume root")
parser.add_argument("--confirm-destructive", action="store_true", help="Required: tests modify JF12 Multilang state before restoring it")
parser.add_argument("--include-refresh", action="store_true", help="Also run a provider-backed single-item refresh; this can take longer than the core suite")
parser.add_argument("--full-server-state", action="store_true", help="Snapshot and restore the JF12 config, data, and cache mounts instead of only Multilang-owned paths")
args = parser.parse_args()
parsed = urlparse(args.base_url)
if not args.confirm_destructive:
parser.error("--confirm-destructive is required")
if args.container != "jellyfin12":
parser.error("This runner only permits the jellyfin12 container")
if parsed.port != 8097:
parser.error("This runner only permits the JF12 test URL on port 8097")
if parsed.hostname in {"127.0.0.1", "::1", "localhost"}:
parser.error("Use a LAN host name or address reachable from inside the Jellyfin container, not loopback.")
if not args.state_root.rstrip("/").endswith("/jellyfin12"):
parser.error("This runner only permits a state root ending in /jellyfin12")
return args
def main() -> int:
args = parse_args()
remote = RemoteState(args)
success = False
try:
remote.verify_target()
remote.snapshot()
api, user_id = wait_for_login(args.base_url, args.username, args.password)
run_checks(api, user_id, args.include_refresh)
success = True
finally:
try:
if remote.expected_web_hash:
remote.restore_and_verify()
scope = "JF12 config, data, cache, and injected web index" if args.full_server_state else "Multilang state and injected web index"
print(f"Restoration hash check passed for {scope}.")
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())