fix: add verification and playback safeguards

This commit is contained in:
ChindanaiNaKub 2026-06-12 23:14:11 +07:00
parent 312d5d6df0
commit 50a954c585
16 changed files with 1556 additions and 43 deletions

View file

@ -0,0 +1,37 @@
from fastapi.testclient import TestClient
import stream_server
def muted_entry(video):
return {
"video": video,
"mode": 1,
"vol": 0,
"pixel": False,
"cols": 200,
"rows": 0,
}
def test_indexed_audio_does_not_use_global_current_index():
client = TestClient(stream_server.app)
stream_server.app.state.queue = [
muted_entry("missing-a.mp4"),
muted_entry("missing-b.mp4"),
]
stream_server.app.state.current_index = 1
response = client.get("/audio/0")
assert response.status_code == 204
def test_indexed_audio_rejects_out_of_range_index():
client = TestClient(stream_server.app)
stream_server.app.state.queue = [muted_entry("missing-a.mp4")]
stream_server.app.state.current_index = 0
response = client.get("/audio/99")
assert response.status_code == 404

View file

@ -0,0 +1,18 @@
import subprocess
from pathlib import Path
APP_JS = Path(__file__).resolve().parents[1] / "app.js"
def test_app_js_has_idempotent_render_loop_guard():
source = APP_JS.read_text(encoding="utf-8")
assert "let renderLoopId = null;" in source
assert "if (renderLoopId !== null) return;" in source
assert "renderLoopId = requestAnimationFrame(renderFrame);" in source
assert "cancelAnimationFrame(renderLoopId);" in source
def test_app_js_syntax_is_valid():
subprocess.run(["node", "--check", str(APP_JS)], check=True)

View file

@ -0,0 +1,61 @@
import json
from pathlib import Path
from types import SimpleNamespace
import stream_server
def make_args(**overrides):
values = {
"video": "video.mp4",
"playlist": None,
"folder": None,
"mode": 1,
"vol": 1,
"pixel": False,
"cols": None,
"rows": 0,
}
values.update(overrides)
return SimpleNamespace(**values)
def test_calc_auto_rows_preserves_video_aspect_for_text_and_pixel():
text_rows = stream_server.calc_auto_rows(240, 1920, 1080, pixel_mode=False)
pixel_rows = stream_server.calc_auto_rows(450, 1920, 1080, pixel_mode=True)
assert text_rows == round(240 / (1920 / 1080) / 2)
assert pixel_rows == round(450 / (1920 / 1080))
assert pixel_rows > text_rows
def test_load_folder_includes_only_supported_video_files(tmp_path):
for name in ["intro.mp4", "clip.MOV", "notes.txt", "image.png", "scene.webm"]:
(tmp_path / name).write_text("", encoding="utf-8")
items = stream_server.load_folder(str(tmp_path), default_mode=3, default_vol=2)
names = {Path(item["video"]).name for item in items}
assert names == {"intro.mp4", "clip.MOV", "scene.webm"}
assert {item["mode"] for item in items} == {3}
assert {item["vol"] for item in items} == {2}
def test_build_queue_fills_playlist_defaults(tmp_path):
playlist = tmp_path / "playlist.json"
playlist.write_text(json.dumps([{"video": "clip.mp4"}]), encoding="utf-8")
queue = stream_server.build_queue(
make_args(playlist=str(playlist), mode=3, vol=2, pixel=False, cols=220)
)
assert queue == [
{
"video": "clip.mp4",
"mode": 3,
"vol": 2,
"pixel": False,
"cols": 220,
"rows": 0,
}
]

View file

@ -0,0 +1,88 @@
import json
from types import SimpleNamespace
import pytest
import stream_server
def make_args(**overrides):
values = {
"video": "video.mp4",
"playlist": None,
"folder": None,
"mode": 1,
"vol": 1,
"pixel": False,
"cols": None,
"rows": 0,
}
values.update(overrides)
return SimpleNamespace(**values)
def assert_rejected(args, message_part):
with pytest.raises(ValueError, match=message_part):
stream_server.build_queue(args)
def test_valid_single_video_args_produce_normalized_queue_entry():
queue = stream_server.build_queue(
make_args(video="movie.mp4", mode=5, vol=3, pixel=True, cols=520, rows=240)
)
assert queue == [
{
"video": "movie.mp4",
"mode": 5,
"vol": 3,
"pixel": True,
"cols": 520,
"rows": 240,
}
]
@pytest.mark.parametrize("vol", [-1, 6])
def test_rejects_volume_outside_supported_range(vol):
assert_rejected(make_args(vol=vol), "vol must be between 0 and 5")
@pytest.mark.parametrize("cols", [0, -1, stream_server.MAX_COLS + 1])
def test_rejects_invalid_columns(cols):
assert_rejected(make_args(cols=cols), "cols must be between")
@pytest.mark.parametrize("rows", [-1, stream_server.MAX_ROWS + 1])
def test_rejects_invalid_rows(rows):
assert_rejected(make_args(rows=rows), "rows must be between")
def test_rejects_playlist_entry_missing_video(tmp_path):
playlist = tmp_path / "playlist.json"
playlist.write_text(json.dumps([{"mode": 3}]), encoding="utf-8")
assert_rejected(make_args(playlist=str(playlist)), "playlist entry 1: video")
def test_rejects_pixel_playlist_entry_in_text_mode(tmp_path):
playlist = tmp_path / "playlist.json"
playlist.write_text(
json.dumps([{"video": "clip.mp4", "mode": 1, "pixel": True}]),
encoding="utf-8",
)
assert_rejected(make_args(playlist=str(playlist)), "pixel mode requires color mode 2-5")
def test_rejects_explicit_grid_larger_than_cell_limit():
assert_rejected(
make_args(mode=5, pixel=True, cols=1200, rows=800),
"grid size 1200x800 exceeds",
)
def test_accepts_zero_rows_for_auto_scaling():
queue = stream_server.build_queue(make_args(cols=200, rows=0))
assert queue[0]["rows"] == 0

View file

@ -0,0 +1,10 @@
import py_compile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_python_entrypoints_compile():
py_compile.compile(str(ROOT / "stream_server.py"), doraise=True)
py_compile.compile(str(ROOT / "ascii_video_player2.py"), doraise=True)