92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
# tests/test_media_indexing.py — Audio/video in the tree & index (roadmap #109-A/F2)
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
from backend.indexer import _index_single_file_sync, _scan_vault
|
|
from backend.services.vaults import browse_directory
|
|
from backend.watcher import VaultEventHandler
|
|
|
|
MEDIA_BYTES = b"ID3\x03\x00\x00\x00\x00\x00\x00" + bytes(range(256)) * 16
|
|
|
|
|
|
def _make_vault(tmp_path: Path) -> Path:
|
|
vault = tmp_path / "Vault"
|
|
vault.mkdir()
|
|
(vault / "note.md").write_text("# Note\n", encoding="utf-8")
|
|
(vault / "note-vocale.mp3").write_bytes(MEDIA_BYTES)
|
|
(vault / "clip.mp4").write_bytes(MEDIA_BYTES)
|
|
assets = vault / "assets"
|
|
assets.mkdir()
|
|
(assets / "ambiance.flac").write_bytes(MEDIA_BYTES)
|
|
return vault
|
|
|
|
|
|
class TestScanVaultMedia:
|
|
def test_audio_video_indexed_without_content(self, tmp_path):
|
|
vault = _make_vault(tmp_path)
|
|
result = _scan_vault("V", str(vault))
|
|
by_path = {f["path"]: f for f in result["files"]}
|
|
for path in ("note-vocale.mp3", "clip.mp4", "assets/ambiance.flac"):
|
|
assert path in by_path, path
|
|
entry = by_path[path]
|
|
assert entry["content"] == "" # never read the bytes (#109-A1)
|
|
assert entry["content_preview"] == ""
|
|
assert entry["size"] > 0
|
|
|
|
def test_media_present_in_paths(self, tmp_path):
|
|
vault = _make_vault(tmp_path)
|
|
result = _scan_vault("V", str(vault))
|
|
files = {p["path"] for p in result["paths"] if p["type"] == "file"}
|
|
assert "note-vocale.mp3" in files
|
|
assert "clip.mp4" in files
|
|
assert "assets/ambiance.flac" in files
|
|
|
|
def test_no_binary_content_leaks(self, tmp_path):
|
|
vault = _make_vault(tmp_path)
|
|
result = _scan_vault("V", str(vault))
|
|
for f in result["files"]:
|
|
if f["extension"] in (".mp3", ".mp4", ".flac"):
|
|
assert f["content"] == ""
|
|
|
|
|
|
class TestSingleFileMedia:
|
|
def test_metadata_only(self, tmp_path):
|
|
vault = _make_vault(tmp_path)
|
|
info = _index_single_file_sync("V", str(vault), str(vault / "note-vocale.mp3"))
|
|
assert info is not None
|
|
assert info["content"] == ""
|
|
assert info["extension"] == ".mp3"
|
|
assert info["path"] == "note-vocale.mp3"
|
|
|
|
|
|
class TestWatcherRelevance:
|
|
def test_media_extensions_relevant(self, tmp_path):
|
|
handler = VaultEventHandler("V", asyncio.Queue(), asyncio.new_event_loop())
|
|
for name in ("a.mp3", "a.flac", "a.opus", "a.mp4", "a.webm", "a.mov"):
|
|
assert handler._is_relevant(str(tmp_path / name)) is True, name
|
|
assert handler._is_relevant(str(tmp_path / "a.mkv")) is False
|
|
|
|
|
|
class TestBrowseAndSearch:
|
|
def test_browse_lists_media(self, client):
|
|
resp = client.get("/api/browse/TestVault")
|
|
assert resp.status_code == 200
|
|
items = resp.json()["items"]
|
|
names = {i["name"]: i for i in items}
|
|
assert "sample-audio.mp3" in names
|
|
assert names["sample-audio.mp3"]["type"] == "file"
|
|
assert names["sample-audio.mp3"]["extension"] == ".mp3"
|
|
|
|
def test_ext_filter_returns_audio(self, client):
|
|
resp = client.get("/api/search/advanced", params={"q": "ext:mp3"})
|
|
assert resp.status_code == 200
|
|
paths = [r["path"] for r in resp.json()["results"]]
|
|
assert any(p.endswith(".mp3") for p in paths)
|
|
|
|
|
|
class TestBrowseService:
|
|
def test_browse_directory_service(self, app_with_vault, test_vault_dir):
|
|
data = browse_directory("TestVault", "")
|
|
names = [i["name"] for i in data["items"]]
|
|
assert "sample-audio.mp3" in names
|