import ./make-test-python.nix ({ lib, pkgs, ... }: { name = "jellyfin"; meta.maintainers = with lib.maintainers; [ minijackson ]; nodes.machine = { ... }: { services.jellyfin.enable = true; environment.systemPackages = with pkgs; [ ffmpeg ]; }; # Documentation of the Jellyfin API: https://api.jellyfin.org/ # Beware, this link can be resource intensive testScript = let payloads = { auth = pkgs.writeText "auth.json" (builtins.toJSON { Username = "jellyfin"; }); empty = pkgs.writeText "empty.json" (builtins.toJSON { }); }; in '' import json from urllib.parse import urlencode machine.wait_for_unit("jellyfin.service") machine.wait_for_open_port(8096) machine.succeed("curl --fail http://localhost:8096/") machine.wait_until_succeeds("curl --fail http://localhost:8096/health | grep Healthy") auth_header = 'MediaBrowser Client="NixOS Integration Tests", DeviceId="1337", Device="Apple II", Version="20.09"' def api_get(path): return f"curl --fail 'http://localhost:8096{path}' -H 'X-Emby-Authorization:{auth_header}'" def api_post(path, json_file=None): if json_file: return f"curl --fail -X post 'http://localhost:8096{path}' -d '@{json_file}' -H Content-Type:application/json -H 'X-Emby-Authorization:{auth_header}'" else: return f"curl --fail -X post 'http://localhost:8096{path}' -H 'X-Emby-Authorization:{auth_header}'" with machine.nested("Wizard completes"): machine.wait_until_succeeds(api_get("/Startup/Configuration")) machine.succeed(api_get("/Startup/FirstUser")) machine.succeed(api_post("/Startup/Complete")) with machine.nested("Can login"): auth_result_str = machine.succeed( api_post( "/Users/AuthenticateByName", "${payloads.auth}", ) ) auth_result = json.loads(auth_result_str) auth_token = auth_result["AccessToken"] auth_header += f", Token={auth_token}" sessions_result_str = machine.succeed(api_get("/Sessions")) sessions_result = json.loads(sessions_result_str) this_session = [ session for session in sessions_result if session["DeviceId"] == "1337" ] if len(this_session) != 1: raise Exception("Session not created") me_str = machine.succeed(api_get("/Users/Me")) me = json.loads(me_str)["Id"] with machine.nested("Can add library"): tempdir = machine.succeed("mktemp -d -p /var/lib/jellyfin").strip() machine.succeed(f"chmod 755 '{tempdir}'") # Generate a dummy video that we can test later videofile = f"{tempdir}/Big Buck Bunny (2008) [1080p].mkv" machine.succeed(f"ffmpeg -f lavfi -i testsrc2=duration=5 '{videofile}'") add_folder_query = urlencode( { "name": "My Library", "collectionType": "Movies", "paths": tempdir, "refreshLibrary": "true", } ) machine.succeed( api_post( f"/Library/VirtualFolders?{add_folder_query}", "${payloads.empty}", ) ) def is_refreshed(_): folders_str = machine.succeed(api_get("/Library/VirtualFolders")) folders = json.loads(folders_str) print(folders) return all(folder["RefreshStatus"] == "Idle" for folder in folders) retry(is_refreshed) with machine.nested("Can identify videos"): items = [] # For some reason, having the folder refreshed doesn't mean the # movie was scanned def has_movie(_): global items items_str = machine.succeed( api_get(f"/Users/{me}/Items?IncludeItemTypes=Movie&Recursive=true") ) items = json.loads(items_str)["Items"] return len(items) == 1 retry(has_movie) video = items[0]["Id"] item_info_str = machine.succeed(api_get(f"/Users/{me}/Items/{video}")) item_info = json.loads(item_info_str) if item_info["Name"] != "Big Buck Bunny": raise Exception("Jellyfin failed to properly identify file") with machine.nested("Can read videos"): media_source_id = item_info["MediaSources"][0]["Id"] machine.succeed( "ffmpeg" + f" -headers 'X-Emby-Authorization:{auth_header}'" + f" -i http://localhost:8096/Videos/{video}/master.m3u8?mediaSourceId={media_source_id}" + " /tmp/test.mkv" ) duration = machine.succeed( "ffprobe /tmp/test.mkv" + " -show_entries format=duration" + " -of compact=print_section=0:nokey=1" ) if duration.strip() != "5.000000": raise Exception("Downloaded video has wrong duration") ''; })