169 lines
8.3 KiB
Python
169 lines
8.3 KiB
Python
# tests/test_media_stream.py — Audio/video streaming & HTTP Range (roadmap #109-A/F1)
|
|
import base64
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# 4 KiB of pseudo-ID3 bytes — the endpoint never decodes the payload, it only
|
|
# has to serve bytes with the right MIME and honour Range.
|
|
MEDIA_BYTES = b"ID3\x03\x00\x00\x00\x00\x00\x00" + bytes(range(256)) * 16
|
|
PDF_BYTES = b"%PDF-1.4\n" + b"x" * 2048 + b"\n%%EOF"
|
|
|
|
|
|
def _write(vault_dir: str, name: str, content: bytes) -> None:
|
|
(Path(vault_dir) / name).write_bytes(content)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# /api/file — audio/video metadata returned before read_text (#109-A4)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestMediaFileView:
|
|
def test_audio_metadata(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "note-vocale.mp3", MEDIA_BYTES)
|
|
resp = client.get("/api/file/TestVault", params={"path": "note-vocale.mp3"})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["is_audio"] is True
|
|
assert data["is_video"] is not True
|
|
assert data["media_mime"] == "audio/mpeg"
|
|
assert data["stream_url"] == "/api/media/TestVault?path=note-vocale.mp3"
|
|
assert data["size_bytes"] == len(MEDIA_BYTES)
|
|
assert data["unsupported"] is not True
|
|
|
|
def test_video_metadata(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "clip.mp4", MEDIA_BYTES)
|
|
resp = client.get("/api/file/TestVault", params={"path": "clip.mp4"})
|
|
data = resp.json()
|
|
assert data["is_video"] is True
|
|
assert data["is_audio"] is not True
|
|
assert data["media_mime"] == "video/mp4"
|
|
assert data["stream_url"] == "/api/media/TestVault?path=clip.mp4"
|
|
|
|
def test_large_media_falls_back_to_download(self, client, test_vault_dir, monkeypatch):
|
|
_write(test_vault_dir, "huge.mp3", MEDIA_BYTES)
|
|
monkeypatch.setenv("OBSIGATE_MEDIA_MAX_INLINE_MB", "0.000001")
|
|
resp = client.get("/api/file/TestVault", params={"path": "huge.mp3"})
|
|
data = resp.json()
|
|
assert data["unsupported"] is True
|
|
assert data["media_too_large"] is True
|
|
assert not data.get("is_audio")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# /api/media — byte-range streaming (#109-A2/F1)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestMediaStreamRange:
|
|
def test_no_range_returns_200_and_accept_ranges(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "song.mp3", MEDIA_BYTES)
|
|
resp = client.get("/api/media/TestVault", params={"path": "song.mp3"})
|
|
assert resp.status_code == 200
|
|
assert resp.headers["accept-ranges"] == "bytes"
|
|
assert resp.headers["content-type"].startswith("audio/mpeg")
|
|
assert resp.content == MEDIA_BYTES
|
|
|
|
def test_first_kibibyte_returns_206(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "song.mp3", MEDIA_BYTES)
|
|
resp = client.get(
|
|
"/api/media/TestVault",
|
|
params={"path": "song.mp3"},
|
|
headers={"Range": "bytes=0-1023"},
|
|
)
|
|
assert resp.status_code == 206
|
|
assert resp.headers["accept-ranges"] == "bytes"
|
|
assert resp.headers["content-range"] == f"bytes 0-1023/{len(MEDIA_BYTES)}"
|
|
assert len(resp.content) == 1024
|
|
assert resp.content == MEDIA_BYTES[:1024]
|
|
|
|
def test_suffix_range(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "song.mp3", MEDIA_BYTES)
|
|
resp = client.get(
|
|
"/api/media/TestVault",
|
|
params={"path": "song.mp3"},
|
|
headers={"Range": "bytes=-100"},
|
|
)
|
|
assert resp.status_code == 206
|
|
assert resp.content == MEDIA_BYTES[-100:]
|
|
|
|
def test_unsatisfiable_range_returns_416(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "song.mp3", MEDIA_BYTES)
|
|
resp = client.get(
|
|
"/api/media/TestVault",
|
|
params={"path": "song.mp3"},
|
|
headers={"Range": f"bytes={len(MEDIA_BYTES) + 10}-"},
|
|
)
|
|
assert resp.status_code == 416
|
|
assert resp.headers["content-range"] == f"bytes */{len(MEDIA_BYTES)}"
|
|
|
|
def test_mov_mime_override(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "clip.mov", MEDIA_BYTES)
|
|
resp = client.get("/api/media/TestVault", params={"path": "clip.mov"})
|
|
assert resp.headers["content-type"].startswith("video/quicktime")
|
|
|
|
def test_rejects_non_media(self, client, test_vault_dir):
|
|
resp = client.get("/api/media/TestVault", params={"path": "note1.md"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_missing_media_404(self, client):
|
|
resp = client.get("/api/media/TestVault", params={"path": "nope.mp3"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_unknown_vault_404(self, client):
|
|
resp = client.get("/api/media/NoSuchVault", params={"path": "song.mp3"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_path_traversal_rejected(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "song.mp3", MEDIA_BYTES)
|
|
resp = client.get("/api/media/TestVault", params={"path": "../../../etc/passwd.mp3"})
|
|
assert resp.status_code == 403
|
|
|
|
def test_media_too_large_returns_413(self, client, test_vault_dir, monkeypatch):
|
|
_write(test_vault_dir, "huge.mp3", MEDIA_BYTES)
|
|
monkeypatch.setenv("OBSIGATE_MEDIA_MAX_INLINE_MB", "0.000001")
|
|
resp = client.get("/api/media/TestVault", params={"path": "huge.mp3"})
|
|
assert resp.status_code == 413
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Vault access control (#109-A3)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestMediaAccess:
|
|
def test_normaluser_cannot_stream_other_vault(self, admin_client):
|
|
resp = admin_client.post(
|
|
"/api/auth/login", json={"username": "normaluser", "password": "normal123"}
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
token = resp.json()["access_token"]
|
|
resp = admin_client.get(
|
|
"/api/media/OtherVault",
|
|
params={"path": "song.mp3"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Regression: pdf/stream unchanged after extracting the shared helper (#109-A2)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestPdfStreamRegression:
|
|
def test_pdf_range_still_works(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "doc.pdf", PDF_BYTES)
|
|
resp = client.get(
|
|
"/api/file/TestVault/pdf/stream",
|
|
params={"path": "doc.pdf"},
|
|
headers={"Range": "bytes=0-9"},
|
|
)
|
|
assert resp.status_code == 206
|
|
assert resp.headers["content-type"].startswith("application/pdf")
|
|
assert resp.headers["content-range"] == f"bytes 0-9/{len(PDF_BYTES)}"
|
|
assert resp.content == PDF_BYTES[:10]
|
|
|
|
def test_pdf_without_range_returns_200(self, client, test_vault_dir):
|
|
_write(test_vault_dir, "doc.pdf", PDF_BYTES)
|
|
resp = client.get("/api/file/TestVault/pdf/stream", params={"path": "doc.pdf"})
|
|
assert resp.status_code == 200
|
|
assert resp.headers["accept-ranges"] == "bytes"
|