titaniumtown/nixos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tests: extract shared jellyfin test helpers and use real jellyfin in annotations test

Browse Source
This commit is contained in:
Simon Gardling
2026-04-01 02:18:23 -04:00
parent a5206b9ec6
commit 297264a34a
4 changed files with 213 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
tests/jellyfin-annotations.nix
View File

@@ -4,40 +4,8 @@
...
}:
let
jfLib = import ./jellyfin-test-lib.nix { inherit pkgs lib; };
mockGrafana = ./mock-grafana-server.py;
mockJellyfin = pkgs.writeText "mock-jellyfin-server.py" ''
import http.server, json, sys
PORT = int(sys.argv[1])
DATA_FILE = sys.argv[2]
class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
pass
def _json(self, code, body):
data = json.dumps(body).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(data)
def do_GET(self):
if self.path.startswith("/Sessions"):
try:
with open(DATA_FILE) as f:
sessions = json.load(f)
except Exception:
sessions = []
self._json(200, sessions)
else:
self.send_response(404)
self.end_headers()
http.server.HTTPServer(("127.0.0.1", PORT), Handler).serve_forever()
'';
script = ../services/jellyfin-annotations.py;
python = pkgs.python3;
in
@@ -47,6 +15,7 @@ pkgs.testers.runNixOSTest {
nodes.machine =
{ pkgs, ... }:
{
imports = [ jfLib.jellyfinTestConfig ];
environment.systemPackages = [ pkgs.python3 ];
};
@@ -54,39 +23,42 @@ pkgs.testers.runNixOSTest {
import json
import time
JELLYFIN_PORT = 18096
import importlib.util
_spec = importlib.util.spec_from_file_location("jf_helpers", "${jfLib.helpers}")
assert _spec and _spec.loader
_jf = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_jf)
setup_jellyfin = _jf.setup_jellyfin
jellyfin_api = _jf.jellyfin_api
GRAFANA_PORT = 13000
SESSIONS_FILE = "/tmp/sessions.json"
ANNOTS_FILE = "/tmp/annotations.json"
STATE_FILE = "/tmp/annotations-state.json"
CREDS_DIR = "/tmp/test-creds"
PYTHON = "${python}/bin/python3"
MOCK_GRAFANA = "${mockGrafana}"
MOCK_JELLYFIN = "${mockJellyfin}"
SCRIPT = "${script}"
auth_header = 'MediaBrowser Client="Infuse", DeviceId="test-dev-1", Device="iPhone", Version="1.0"'
auth_header2 = 'MediaBrowser Client="Jellyfin Web", DeviceId="test-dev-2", Device="Chrome", Version="1.0"'
def read_annotations():
out = machine.succeed(f"cat {ANNOTS_FILE} 2>/dev/null || echo '[]'")
return json.loads(out.strip())
start_all()
machine.wait_for_unit("multi-user.target")
token, user_id, movie_id, media_source_id = setup_jellyfin(
machine, retry, auth_header,
"${jfLib.payloads.auth}", "${jfLib.payloads.empty}",
)
with subtest("Setup mock credentials and data files"):
machine.succeed(f"mkdir -p {CREDS_DIR} && echo 'fake-api-key' > {CREDS_DIR}/jellyfin-api-key")
machine.succeed(f"echo '[]' > {SESSIONS_FILE}")
with subtest("Setup mock Grafana and credentials"):
machine.succeed(f"mkdir -p {CREDS_DIR}")
machine.succeed(f"echo '{token}' > {CREDS_DIR}/jellyfin-api-key")
machine.succeed(f"echo '[]' > {ANNOTS_FILE}")
with subtest("Start mock Jellyfin and Grafana servers"):
machine.succeed(
f"systemd-run --unit=mock-jellyfin {PYTHON} {MOCK_JELLYFIN} {JELLYFIN_PORT} {SESSIONS_FILE}"
)
machine.succeed(
f"systemd-run --unit=mock-grafana {PYTHON} {MOCK_GRAFANA} {GRAFANA_PORT} {ANNOTS_FILE}"
)
machine.wait_until_succeeds(
f"curl -sf http://127.0.0.1:{JELLYFIN_PORT}/Sessions", timeout=10
)
machine.wait_until_succeeds(
f"curl -sf -X POST http://127.0.0.1:{GRAFANA_PORT}/api/annotations "
f"-H 'Content-Type: application/json' -d '{{\"text\":\"ping\",\"tags\":[]}}' | grep -q id",
@@ -94,10 +66,10 @@ pkgs.testers.runNixOSTest {
)
machine.succeed(f"echo '[]' > {ANNOTS_FILE}")
with subtest("Start annotation service pointing at mock servers"):
with subtest("Start annotation service"):
machine.succeed(
f"systemd-run --unit=annotations-svc "
f"--setenv=JELLYFIN_URL=http://127.0.0.1:{JELLYFIN_PORT} "
f"--setenv=JELLYFIN_URL=http://127.0.0.1:8096 "
f"--setenv=GRAFANA_URL=http://127.0.0.1:{GRAFANA_PORT} "
f"--setenv=CREDENTIALS_DIRECTORY={CREDS_DIR} "
f"--setenv=STATE_FILE={STATE_FILE} "
@@ -106,38 +78,24 @@ pkgs.testers.runNixOSTest {
)
time.sleep(2)
with subtest("No annotations pushed when no streams active"):
with subtest("No annotations when no streams active"):
time.sleep(4)
annots = read_annotations()
assert annots == [], f"Expected no annotations, got: {annots}"
with subtest("Annotation created when stream starts"):
rich_session = json.dumps([{
"Id": "sess-1",
"UserName": "alice",
"Client": "Infuse",
"DeviceName": "iPhone",
"PlayState": {"PlayMethod": "Transcode"},
"NowPlayingItem": {
"Name": "Inception",
"Type": "Movie",
"Bitrate": 20000000,
"MediaStreams": [
{"Type": "Video", "Codec": "h264", "Width": 1920, "Height": 1080},
{"Type": "Audio", "Codec": "dts", "Channels": 6, "IsDefault": True},
],
},
"TranscodingInfo": {
"IsVideoDirect": True,
"IsAudioDirect": False,
"VideoCodec": "h264",
"AudioCodec": "aac",
"AudioChannels": 2,
"Bitrate": 8000000,
"TranscodeReasons": ["AudioCodecNotSupported"],
},
}])
machine.succeed(f"echo {repr(rich_session)} > {SESSIONS_FILE}")
with subtest("Annotation created when playback starts"):
playback_start = json.dumps({
"ItemId": movie_id,
"MediaSourceId": media_source_id,
"PlaySessionId": "test-play-1",
"CanSeek": True,
"IsPaused": False,
})
machine.succeed(
f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing' "
f"-d '{playback_start}' -H 'Content-Type:application/json' "
f"-H 'X-Emby-Authorization:{auth_header}, Token={token}'"
)
machine.wait_until_succeeds(
f"cat {ANNOTS_FILE} | python3 -c \"import sys,json; a=json.load(sys.stdin); exit(0 if a else 1)\"",
timeout=15,
@@ -145,18 +103,24 @@ pkgs.testers.runNixOSTest {
annots = read_annotations()
assert len(annots) == 1, f"Expected 1 annotation, got: {annots}"
text = annots[0]["text"]
assert "alice: Inception (movie)" in text, f"Missing title in: {text}"
assert "Transcode" in text, f"Missing method in: {text}"
assert "H.264" in text, f"Missing video codec in: {text}"
assert "DTS" in text and "AAC" in text, f"Missing audio codec in: {text}"
assert "8.0 Mbps" in text, f"Missing bitrate in: {text}"
assert "AudioCodecNotSupported" in text, f"Missing transcode reason in: {text}"
assert "Infuse" in text and "iPhone" in text, f"Missing client in: {text}"
assert "jellyfin" in annots[0].get("tags", []), f"Missing jellyfin tag: {annots[0]}"
assert "Test Movie" in text, f"Missing title in: {text}"
assert "Infuse" in text, f"Missing client in: {text}"
assert "iPhone" in text, f"Missing device in: {text}"
assert "timeEnd" not in annots[0], f"timeEnd should not be set yet: {annots[0]}"
with subtest("Annotation closed when stream ends"):
machine.succeed(f"echo '[]' > {SESSIONS_FILE}")
with subtest("Annotation closed when playback stops"):
playback_stop = json.dumps({
"ItemId": movie_id,
"MediaSourceId": media_source_id,
"PlaySessionId": "test-play-1",
"PositionTicks": 50000000,
})
machine.succeed(
f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing/Stopped' "
f"-d '{playback_stop}' -H 'Content-Type:application/json' "
f"-H 'X-Emby-Authorization:{auth_header}, Token={token}'"
)
machine.wait_until_succeeds(
f"cat {ANNOTS_FILE} | python3 -c \"import sys,json; a=json.load(sys.stdin); exit(0 if a and 'timeEnd' in a[0] else 1)\"",
timeout=15,
@@ -168,11 +132,37 @@ pkgs.testers.runNixOSTest {
with subtest("Multiple concurrent streams each get their own annotation"):
machine.succeed(f"echo '[]' > {ANNOTS_FILE}")
auth_result2 = json.loads(machine.succeed(
f"curl -sf -X POST 'http://localhost:8096/Users/AuthenticateByName' "
f"-d '@${jfLib.payloads.auth}' -H 'Content-Type:application/json' "
f"-H 'X-Emby-Authorization:{auth_header2}'"
))
token2 = auth_result2["AccessToken"]
playback1 = json.dumps({
"ItemId": movie_id,
"MediaSourceId": media_source_id,
"PlaySessionId": "test-play-multi-1",
"CanSeek": True,
"IsPaused": False,
})
machine.succeed(
f"""echo '[
{{"Id":"sess-2","UserName":"bob","NowPlayingItem":{{"Name":"Breaking Bad","SeriesName":"Breaking Bad","ParentIndexNumber":1,"IndexNumber":1}}}},
{{"Id":"sess-3","UserName":"carol","NowPlayingItem":{{"Name":"Inception","Type":"Movie"}}}}
]' > {SESSIONS_FILE}"""
f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing' "
f"-d '{playback1}' -H 'Content-Type:application/json' "
f"-H 'X-Emby-Authorization:{auth_header}, Token={token}'"
)
playback2 = json.dumps({
"ItemId": movie_id,
"MediaSourceId": media_source_id,
"PlaySessionId": "test-play-multi-2",
"CanSeek": True,
"IsPaused": False,
})
machine.succeed(
f"curl -sf -X POST 'http://localhost:8096/Sessions/Playing' "
f"-d '{playback2}' -H 'Content-Type:application/json' "
f"-H 'X-Emby-Authorization:{auth_header2}, Token={token2}'"
)
machine.wait_until_succeeds(
f"cat {ANNOTS_FILE} | python3 -c \"import sys,json; a=json.load(sys.stdin); exit(0 if len(a)==2 else 1)\"",
@@ -180,16 +170,13 @@ pkgs.testers.runNixOSTest {
)
annots = read_annotations()
assert len(annots) == 2, f"Expected 2 annotations, got: {annots}"
texts = sorted(a["text"] for a in annots)
assert any("Breaking Bad" in t and "S01E01" in t for t in texts), f"Missing Bob's annotation: {texts}"
assert any("carol" in t and "Inception" in t for t in texts), f"Missing Carol's annotation: {texts}"
with subtest("State survives service restart (no duplicate annotations)"):
machine.succeed("systemctl stop annotations-svc || true")
time.sleep(1)
machine.succeed(
f"systemd-run --unit=annotations-svc-2 "
f"--setenv=JELLYFIN_URL=http://127.0.0.1:{JELLYFIN_PORT} "
f"--setenv=JELLYFIN_URL=http://127.0.0.1:8096 "
f"--setenv=GRAFANA_URL=http://127.0.0.1:{GRAFANA_PORT} "
f"--setenv=CREDENTIALS_DIRECTORY={CREDS_DIR} "
f"--setenv=STATE_FILE={STATE_FILE} "