Senko-san
105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
"""Unit tests for LocalFileStorage."""
|
|
|
|
import pytest
|
|
from app.infrastructure.storage.local import LocalFileStorage
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
async def test_save_and_stat(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
src = tmp_path / "src.mp3"
|
|
src.write_bytes(b"test audio data")
|
|
|
|
size = await storage.save_file("tracks/te/test.mp3", src)
|
|
assert size == 15
|
|
|
|
stat = await storage.stat("tracks/te/test.mp3")
|
|
assert stat.size == 15
|
|
assert stat.content_type == "audio/mpeg"
|
|
|
|
|
|
async def test_save_creates_parent_dirs(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
src = tmp_path / "src.flac"
|
|
src.write_bytes(b"x")
|
|
|
|
await storage.save_file("tracks/ab/cdef.flac", src)
|
|
assert (tmp_path / "tracks" / "ab" / "cdef.flac").exists()
|
|
|
|
|
|
async def test_open_range_full(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
data = b"hello world" * 100
|
|
src = tmp_path / "src.flac"
|
|
src.write_bytes(data)
|
|
await storage.save_file("tracks/he/hello.flac", src)
|
|
|
|
stream, total = await storage.open_range("tracks/he/hello.flac", 0, None)
|
|
result = b"".join([chunk async for chunk in stream])
|
|
assert result == data
|
|
assert total == len(data)
|
|
|
|
|
|
async def test_open_range_partial(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
data = b"0123456789"
|
|
src = tmp_path / "src.mp3"
|
|
src.write_bytes(data)
|
|
await storage.save_file("tracks/sr/src.mp3", src)
|
|
|
|
stream, total = await storage.open_range("tracks/sr/src.mp3", 3, 7)
|
|
result = b"".join([chunk async for chunk in stream])
|
|
assert result == b"34567"
|
|
assert total == 10
|
|
|
|
|
|
async def test_open_range_from_offset_to_end(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
data = b"abcdefghij"
|
|
src = tmp_path / "src.wav"
|
|
src.write_bytes(data)
|
|
await storage.save_file("tracks/sr/src.wav", src)
|
|
|
|
stream, total = await storage.open_range("tracks/sr/src.wav", 5, None)
|
|
result = b"".join([chunk async for chunk in stream])
|
|
assert result == b"fghij"
|
|
assert total == 10
|
|
|
|
|
|
async def test_exists_and_delete(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
src = tmp_path / "src.ogg"
|
|
src.write_bytes(b"ogg data")
|
|
await storage.save_file("tracks/sr/src.ogg", src)
|
|
|
|
assert await storage.exists("tracks/sr/src.ogg") is True
|
|
await storage.delete("tracks/sr/src.ogg")
|
|
assert await storage.exists("tracks/sr/src.ogg") is False
|
|
|
|
|
|
async def test_delete_missing_is_noop(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
await storage.delete("tracks/no/nope.mp3")
|
|
|
|
|
|
async def test_as_local_path(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
src = tmp_path / "src.mp3"
|
|
src.write_bytes(b"local bytes")
|
|
await storage.save_file("tracks/lo/local.mp3", src)
|
|
|
|
async with storage.as_local_path("tracks/lo/local.mp3") as path:
|
|
assert path.read_bytes() == b"local bytes"
|
|
|
|
|
|
async def test_stat_unknown_extension(tmp_path):
|
|
storage = LocalFileStorage(tmp_path)
|
|
src = tmp_path / "src.xyz"
|
|
src.write_bytes(b"mystery")
|
|
await storage.save_file("tracks/my/mystery.xyz", src)
|
|
|
|
stat = await storage.stat("tracks/my/mystery.xyz")
|
|
assert stat.size == 7
|
|
assert stat.content_type is None
|