224 lines
8.1 KiB
Python
224 lines
8.1 KiB
Python
"""Tests for QueueStore - JSON persistence, corruption recovery, and CRUD operations."""
|
|
|
|
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 make_item(item_id="test1", video_id="abc123", title="Test Video", status="pending", **kwargs):
|
|
return QueueItem(
|
|
id=item_id,
|
|
video_id=video_id,
|
|
title=title,
|
|
url=f"https://youtube.com/watch?v={video_id}",
|
|
thumbnail="https://img.youtube.com/vi/" + video_id + "/hqdefault.jpg",
|
|
status=status,
|
|
progress=kwargs.get("progress", 0.0),
|
|
category=kwargs.get("category", "General"),
|
|
added_at=kwargs.get("added_at", "2026-01-01T00:00:00+00:00"),
|
|
quality=kwargs.get("quality", "1080"),
|
|
)
|
|
|
|
|
|
def test_create_empty_store():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
assert store.get_all() == []
|
|
print("PASS: test_create_empty_store")
|
|
|
|
|
|
def test_add_and_get_item():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
item = make_item()
|
|
store.add_item(item)
|
|
items = store.get_all()
|
|
assert len(items) == 1
|
|
assert items[0].id == "test1"
|
|
assert items[0].title == "Test Video"
|
|
print("PASS: test_add_and_get_item")
|
|
|
|
|
|
def test_update_status():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item())
|
|
store.update_status("test1", "completed", download_path="/foo/bar.mp4", file_size="100MB")
|
|
item = store.get_item("test1")
|
|
assert item.status == "completed"
|
|
assert item.download_path == "/foo/bar.mp4"
|
|
assert item.file_size == "100MB"
|
|
assert item.progress == 100.0
|
|
print("PASS: test_update_status")
|
|
|
|
|
|
def test_update_progress():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item())
|
|
store.update_progress("test1", 45.5, speed="1.2MB/s", eta="5m")
|
|
item = store.get_item("test1")
|
|
assert item.progress == 45.5
|
|
assert item.speed == "1.2MB/s"
|
|
assert item.eta == "5m"
|
|
print("PASS: test_update_progress")
|
|
|
|
|
|
def test_remove_item():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item())
|
|
assert store.remove_item("test1") is True
|
|
assert store.get_all() == []
|
|
assert store.remove_item("nonexistent") is False
|
|
print("PASS: test_remove_item")
|
|
|
|
|
|
def test_clear_completed():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item(item_id="done1", status="completed"))
|
|
store.add_item(make_item(item_id="done2", status="completed"))
|
|
store.add_item(make_item(item_id="pend1", status="pending"))
|
|
removed = store.clear_completed()
|
|
assert removed == 2
|
|
assert len(store.get_all()) == 1
|
|
assert store.get_item("pend1").status == "pending"
|
|
print("PASS: test_clear_completed")
|
|
|
|
|
|
def test_clear_failed():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item(item_id="fail1", status="failed"))
|
|
store.add_item(make_item(item_id="pend1", status="pending"))
|
|
removed = store.clear_failed()
|
|
assert removed == 1
|
|
assert len(store.get_all()) == 1
|
|
print("PASS: test_clear_failed")
|
|
|
|
|
|
def test_clear_all():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item(item_id="a"))
|
|
store.add_item(make_item(item_id="b"))
|
|
removed = store.clear_all()
|
|
assert removed == 2
|
|
assert len(store.get_all()) == 0
|
|
print("PASS: test_clear_all")
|
|
|
|
|
|
def test_get_stats():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item(item_id="p1", status="pending"))
|
|
store.add_item(make_item(item_id="d1", status="downloading"))
|
|
store.add_item(make_item(item_id="c1", status="completed"))
|
|
store.add_item(make_item(item_id="f1", status="failed"))
|
|
stats = store.get_stats()
|
|
assert stats["total"] == 4
|
|
assert stats["pending"] == 1
|
|
assert stats["downloading"] == 1
|
|
assert stats["completed"] == 1
|
|
assert stats["failed"] == 1
|
|
print("PASS: test_get_stats")
|
|
|
|
|
|
def test_corrupted_json_recovery():
|
|
"""Test that corrupted JSON file is handled gracefully."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
with open(path, "w") as f:
|
|
f.write('{"id": "test1", "title": "Video with invalid char: \x01\x02\x03"}')
|
|
store = QueueStore(store_path=path)
|
|
items = store.get_all()
|
|
assert items == []
|
|
assert store.get_item("test1") is None
|
|
print("PASS: test_corrupted_json_recovery")
|
|
|
|
|
|
def test_corrupted_json_with_control_chars():
|
|
"""Test recovery from control character corruption (the actual bug)."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
with open(path, "w") as f:
|
|
f.write('{"test1": {"id": "test1", "title": "Error: Some long message\nwith\ncontrol\x00chars"}}')
|
|
store = QueueStore(store_path=path)
|
|
items = store.get_all()
|
|
assert items == []
|
|
print("PASS: test_corrupted_json_with_control_chars")
|
|
|
|
|
|
def test_empty_file_recovery():
|
|
"""Test recovery from empty file."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
with open(path, "w") as f:
|
|
f.write("")
|
|
store = QueueStore(store_path=path)
|
|
items = store.get_all()
|
|
assert items == []
|
|
print("PASS: test_empty_file_recovery")
|
|
|
|
|
|
def test_persistence_across_instances():
|
|
"""Test that data persists when creating new QueueStore instance."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store1 = QueueStore(store_path=path)
|
|
store1.add_item(make_item())
|
|
del store1
|
|
store2 = QueueStore(store_path=path)
|
|
items = store2.get_all()
|
|
assert len(items) == 1
|
|
assert items[0].id == "test1"
|
|
print("PASS: test_persistence_across_instances")
|
|
|
|
|
|
def test_error_message_with_special_chars():
|
|
"""Test that error messages with special characters don't corrupt the file."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = os.path.join(tmpdir, "test_queue.json")
|
|
store = QueueStore(store_path=path)
|
|
store.add_item(make_item())
|
|
special_msg = "Error: Connection timeout\nRetrying...\nFailed after 3 attempts"
|
|
store.update_status("test1", "failed", error_message=special_msg)
|
|
item = store.get_item("test1")
|
|
assert item.error_message == special_msg
|
|
assert item.status == "failed"
|
|
print("PASS: test_error_message_with_special_chars")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_create_empty_store()
|
|
test_add_and_get_item()
|
|
test_update_status()
|
|
test_update_progress()
|
|
test_remove_item()
|
|
test_clear_completed()
|
|
test_clear_failed()
|
|
test_clear_all()
|
|
test_get_stats()
|
|
test_corrupted_json_recovery()
|
|
test_corrupted_json_with_control_chars()
|
|
test_empty_file_recovery()
|
|
test_persistence_across_instances()
|
|
test_error_message_with_special_chars()
|
|
print("\nAll tests passed!")
|