119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
"""Tests for download recovery after server crash."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from models import QueueItem
|
|
from models.queue_store import QueueStore
|
|
|
|
|
|
def test_recovery_resets_downloading_to_pending():
|
|
"""Test that downloads stuck in 'downloading' status are reset to 'pending' on restart."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
# Simulate a download that was in progress when server crashed
|
|
item = QueueItem(
|
|
id="crash1",
|
|
video_id="vid1",
|
|
title="Crashed Download",
|
|
url="https://youtube.com/watch?v=vid1",
|
|
category="General",
|
|
status="downloading",
|
|
progress=45.0,
|
|
)
|
|
store.add_item(item)
|
|
# Simulate recovery
|
|
items = store.get_all()
|
|
recovered = 0
|
|
for item in items:
|
|
if item.status == "downloading":
|
|
store.update_status(item.id, "pending")
|
|
store.update_progress(item.id, 0.0)
|
|
recovered += 1
|
|
assert recovered == 1
|
|
item = store.get_item("crash1")
|
|
assert item.status == "pending"
|
|
assert item.progress == 0.0
|
|
print("PASS: test_recovery_resets_downloading_to_pending")
|
|
|
|
|
|
def test_recovery_keeps_pending_items():
|
|
"""Test that pending downloads are not affected by recovery."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
item = QueueItem(
|
|
id="pending1",
|
|
video_id="vid1",
|
|
title="Pending Download",
|
|
url="https://youtube.com/watch?v=vid1",
|
|
category="General",
|
|
status="pending",
|
|
progress=0.0,
|
|
)
|
|
store.add_item(item)
|
|
# Simulate recovery
|
|
items = store.get_all()
|
|
for item in items:
|
|
if item.status == "downloading":
|
|
store.update_status(item.id, "pending")
|
|
store.update_progress(item.id, 0.0)
|
|
item = store.get_item("pending1")
|
|
assert item.status == "pending"
|
|
assert item.progress == 0.0
|
|
print("PASS: test_recovery_keeps_pending_items")
|
|
|
|
|
|
def test_recovery_handles_multiple_downloads():
|
|
"""Test that multiple in-progress downloads are recovered."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(QueueItem(id="a", video_id="1", title="A", url="https://y.com/1", status="downloading", progress=30.0))
|
|
store.add_item(QueueItem(id="b", video_id="2", title="B", url="https://y.com/2", status="downloading", progress=60.0))
|
|
store.add_item(QueueItem(id="c", video_id="3", title="C", url="https://y.com/3", status="pending", progress=0.0))
|
|
# Simulate recovery
|
|
items = store.get_all()
|
|
recovered = 0
|
|
for item in items:
|
|
if item.status == "downloading":
|
|
store.update_status(item.id, "pending")
|
|
store.update_progress(item.id, 0.0)
|
|
recovered += 1
|
|
assert recovered == 2
|
|
assert store.get_item("a").status == "pending"
|
|
assert store.get_item("b").status == "pending"
|
|
assert store.get_item("c").status == "pending"
|
|
print("PASS: test_recovery_handles_multiple_downloads")
|
|
|
|
|
|
def test_recovery_preserves_completed():
|
|
"""Test that completed downloads are not affected by recovery."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(QueueItem(id="done1", video_id="1", title="Done", url="https://y.com/1", status="completed", progress=100.0))
|
|
# Simulate recovery
|
|
items = store.get_all()
|
|
for item in items:
|
|
if item.status == "downloading":
|
|
store.update_status(item.id, "pending")
|
|
store.update_progress(item.id, 0.0)
|
|
item = store.get_item("done1")
|
|
assert item.status == "completed"
|
|
assert item.progress == 100.0
|
|
print("PASS: test_recovery_preserves_completed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_recovery_resets_downloading_to_pending()
|
|
test_recovery_keeps_pending_items()
|
|
test_recovery_handles_multiple_downloads()
|
|
test_recovery_preserves_completed()
|
|
print("\nAll recovery tests passed!") |