73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.core.storage import SqliteStorage
|
|
from app.models import Download, DownloadStatus
|
|
|
|
|
|
@pytest.fixture
|
|
async def storage():
|
|
path = Path(tempfile.mkdtemp()) / "test.db"
|
|
st = SqliteStorage(path)
|
|
await st.init()
|
|
yield st
|
|
await st.close()
|
|
|
|
|
|
async def test_create_get_list(storage: SqliteStorage):
|
|
d = Download(url="https://example.com/file.zip")
|
|
await storage.create(d)
|
|
|
|
got = await storage.get(d.id)
|
|
assert got is not None
|
|
assert got.url == d.url
|
|
assert got.status == DownloadStatus.PENDING
|
|
|
|
items = await storage.list()
|
|
assert [i.id for i in items] == [d.id]
|
|
|
|
|
|
async def test_update_returns_snapshot(storage: SqliteStorage):
|
|
d = Download(url="https://example.com/a.zip")
|
|
await storage.create(d)
|
|
|
|
snap = await storage.update(
|
|
d.id, status=DownloadStatus.DOWNLOADING, progress=42.5, downloaded_bytes=100
|
|
)
|
|
assert snap is not None
|
|
assert snap.status == DownloadStatus.DOWNLOADING
|
|
assert snap.progress == 42.5
|
|
assert snap.downloaded_bytes == 100
|
|
|
|
|
|
async def test_update_ignores_unknown_fields(storage: SqliteStorage):
|
|
d = Download(url="https://example.com/a.zip")
|
|
await storage.create(d)
|
|
snap = await storage.update(d.id, bogus="x", progress=10.0)
|
|
assert snap is not None and snap.progress == 10.0
|
|
|
|
|
|
async def test_delete(storage: SqliteStorage):
|
|
d = Download(url="https://example.com/a.zip")
|
|
await storage.create(d)
|
|
assert await storage.delete(d.id) is True
|
|
assert await storage.delete(d.id) is False
|
|
assert await storage.get(d.id) is None
|
|
|
|
|
|
async def test_downloading_marked_failed_on_init():
|
|
path = Path(tempfile.mkdtemp()) / "test.db"
|
|
st = SqliteStorage(path)
|
|
await st.init()
|
|
d = Download(url="https://example.com/a.zip", status=DownloadStatus.DOWNLOADING)
|
|
await st.create(d)
|
|
await st.close()
|
|
|
|
st2 = SqliteStorage(path)
|
|
await st2.init() # должен пометить "качающиеся" записи как FAILED
|
|
got = await st2.get(d.id)
|
|
assert got is not None and got.status == DownloadStatus.FAILED
|
|
await st2.close()
|