115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
import tempfile
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
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_finished_at_roundtrip(storage: SqliteStorage):
|
|
d = Download(url="https://example.com/a.zip")
|
|
await storage.create(d)
|
|
assert (await storage.get(d.id)).finished_at is None
|
|
|
|
ts = datetime.now(timezone.utc)
|
|
snap = await storage.update(d.id, status=DownloadStatus.DONE, finished_at=ts)
|
|
assert snap is not None and snap.finished_at == ts
|
|
|
|
|
|
async def test_migrates_legacy_db_without_finished_at():
|
|
# БД, созданная до появления колонки finished_at.
|
|
path = Path(tempfile.mkdtemp()) / "legacy.db"
|
|
db = await aiosqlite.connect(path)
|
|
await db.execute(
|
|
"""
|
|
CREATE TABLE downloads (
|
|
id TEXT PRIMARY KEY, url TEXT NOT NULL, filename TEXT,
|
|
status TEXT NOT NULL, progress REAL NOT NULL DEFAULT 0,
|
|
size_bytes INTEGER, downloaded_bytes INTEGER NOT NULL DEFAULT 0,
|
|
speed REAL, eta REAL, error TEXT, created_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO downloads (id, url, status, created_at) VALUES (?, ?, ?, ?)",
|
|
("x1", "https://e.com/a.zip", "done", datetime.now(timezone.utc).isoformat()),
|
|
)
|
|
await db.commit()
|
|
await db.close()
|
|
|
|
st = SqliteStorage(path)
|
|
await st.init() # должен добавить колонку, не уронив существующие данные
|
|
got = await st.get("x1")
|
|
assert got is not None and got.status == DownloadStatus.DONE
|
|
assert got.finished_at is None
|
|
await st.close()
|
|
|
|
|
|
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
|
|
assert got.finished_at is not None # прерванная загрузка тоже «завершилась»
|
|
await st2.close()
|