Jarian Cottingham
06877fb2fc
- New TUI using Textual framework with responsive interface - Search, results, download, and queue screens for YouTube video management - Download queue system with sequential background downloads - Status bar with queue count and download progress indicators - Comprehensive test suite (97 tests) with 100% pass rate - Modern Python packaging with pyproject.toml and uv support - Added requirements-tui.txt for TUI-specific dependencies - Updated setup.py and install.sh for TUI integration - Enhanced README.md with TUI usage documentation
325 lines
11 KiB
Python
325 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Integration test for the download queue system
|
|
"""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from youtube_tui.models.queue_item import QueueStatus
|
|
from youtube_tui.models.video import Video
|
|
from youtube_tui.services.queue import DownloadQueue
|
|
|
|
|
|
class TestDownloadQueue:
|
|
"""Tests for the DownloadQueue service"""
|
|
|
|
def test_queue_initialization(self):
|
|
"""Test queue initialization"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create a queue without loading from file
|
|
queue = DownloadQueue.__new__(DownloadQueue)
|
|
queue._queue = []
|
|
queue._archive_file = Path(tmpdir) / "download_queue.json"
|
|
|
|
assert queue.get_stats()["total"] == 0
|
|
|
|
def test_add_video_to_queue(self):
|
|
"""Test adding a video to the queue"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create a test video
|
|
video = Video(
|
|
video_id="test123",
|
|
title="Test Video",
|
|
channel="Test Channel",
|
|
channel_id="channel123",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test description",
|
|
)
|
|
|
|
# Add to queue
|
|
item = queue.add_video(video, category="Tech")
|
|
assert item.status == QueueStatus.PENDING
|
|
assert item.category == "Tech"
|
|
assert item.video.video_id == "test123"
|
|
|
|
# Check stats
|
|
stats = queue.get_stats()
|
|
assert stats["total"] == 1
|
|
assert stats["pending"] == 1
|
|
|
|
def test_remove_video_from_queue(self):
|
|
"""Test removing a video from the queue"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create a test video
|
|
video = Video(
|
|
video_id="test123",
|
|
title="Test Video",
|
|
channel="Test Channel",
|
|
channel_id="channel123",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test description",
|
|
)
|
|
|
|
# Add to queue
|
|
queue.add_video(video)
|
|
|
|
# Remove from queue
|
|
removed = queue.remove_video("test123")
|
|
assert removed is True
|
|
|
|
# Check stats
|
|
stats = queue.get_stats()
|
|
assert stats["total"] == 0
|
|
|
|
def test_update_status_and_progress(self):
|
|
"""Test updating status and progress"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create a test video
|
|
video = Video(
|
|
video_id="test123",
|
|
title="Test Video",
|
|
channel="Test Channel",
|
|
channel_id="channel123",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test description",
|
|
)
|
|
|
|
# Add to queue
|
|
queue.add_video(video)
|
|
|
|
# Update status to downloading
|
|
queue.update_status(
|
|
"test123", QueueStatus.DOWNLOADING, progress=50
|
|
)
|
|
|
|
# Get the item and check status
|
|
items = queue.get_queue()
|
|
assert len(items) == 1
|
|
assert items[0].status == QueueStatus.DOWNLOADING
|
|
assert items[0].progress == 50
|
|
|
|
# Update status to completed
|
|
queue.update_status(
|
|
"test123", QueueStatus.COMPLETED, progress=100
|
|
)
|
|
|
|
items = queue.get_queue()
|
|
assert items[0].status == QueueStatus.COMPLETED
|
|
assert items[0].progress == 100
|
|
|
|
def test_cancel_video(self):
|
|
"""Test cancelling a video in the queue"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create a test video
|
|
video = Video(
|
|
video_id="test123",
|
|
title="Test Video",
|
|
channel="Test Channel",
|
|
channel_id="channel123",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test description",
|
|
)
|
|
|
|
# Add to queue
|
|
queue.add_video(video)
|
|
|
|
# Cancel the video
|
|
cancelled = queue.cancel_video("test123")
|
|
assert cancelled is True
|
|
|
|
# Check status
|
|
items = queue.get_queue()
|
|
assert items[0].status == QueueStatus.CANCELLED
|
|
|
|
def test_get_next_pending(self):
|
|
"""Test getting the next pending item"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create test videos
|
|
video1 = Video(
|
|
video_id="test1",
|
|
title="Test Video 1",
|
|
channel="Test Channel",
|
|
channel_id="channel1",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test",
|
|
)
|
|
video2 = Video(
|
|
video_id="test2",
|
|
title="Test Video 2",
|
|
channel="Test Channel",
|
|
channel_id="channel2",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test",
|
|
)
|
|
|
|
# Add to queue
|
|
queue.add_video(video1)
|
|
queue.add_video(video2)
|
|
|
|
# Get next pending
|
|
next_item = queue.get_next_pending()
|
|
assert next_item is not None
|
|
assert next_item.video.video_id == "test1"
|
|
|
|
# Update first item to downloading
|
|
queue.update_status("test1", QueueStatus.DOWNLOADING)
|
|
|
|
# Get next pending - should be test2
|
|
next_item = queue.get_next_pending()
|
|
assert next_item.video.video_id == "test2"
|
|
|
|
# Update test2 to downloading
|
|
queue.update_status("test2", QueueStatus.DOWNLOADING)
|
|
|
|
# No more pending items
|
|
next_item = queue.get_next_pending()
|
|
assert next_item is None
|
|
|
|
def test_clear_completed(self):
|
|
"""Test clearing completed and cancelled items"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create test videos
|
|
video1 = Video(
|
|
video_id="test1",
|
|
title="Test Video 1",
|
|
channel="Test Channel",
|
|
channel_id="channel1",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test",
|
|
)
|
|
video2 = Video(
|
|
video_id="test2",
|
|
title="Test Video 2",
|
|
channel="Test Channel",
|
|
channel_id="channel2",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test",
|
|
)
|
|
video3 = Video(
|
|
video_id="test3",
|
|
title="Test Video 3",
|
|
channel="Test Channel",
|
|
channel_id="channel3",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test",
|
|
)
|
|
|
|
# Add to queue
|
|
queue.add_video(video1) # pending
|
|
queue.add_video(video2) # pending
|
|
|
|
# Mark test1 as completed
|
|
queue.update_status("test1", QueueStatus.COMPLETED)
|
|
|
|
# Mark test2 as cancelled
|
|
queue.cancel_video("test2")
|
|
|
|
# Mark test3 as completed
|
|
queue.add_video(video3)
|
|
queue.update_status("test3", QueueStatus.COMPLETED)
|
|
|
|
# Clear completed and cancelled
|
|
removed = queue.clear_completed()
|
|
assert removed == 3 # All three should be removed
|
|
|
|
# Check stats
|
|
stats = queue.get_stats()
|
|
assert stats["total"] == 0
|
|
|
|
def test_clear_failed(self):
|
|
"""Test clearing failed items"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch.object(
|
|
DownloadQueue,
|
|
"_archive_file",
|
|
Path(tmpdir) / "download_queue.json",
|
|
):
|
|
queue = DownloadQueue()
|
|
|
|
# Create test videos
|
|
video1 = Video(
|
|
video_id="test1",
|
|
title="Test Video 1",
|
|
channel="Test Channel",
|
|
channel_id="channel1",
|
|
duration="10:00",
|
|
view_count="1000",
|
|
upload_date="20240101",
|
|
description="Test",
|
|
)
|
|
|
|
# Add to queue and fail it
|
|
queue.add_video(video1)
|
|
queue.update_status("test1", QueueStatus.FAILED)
|
|
|
|
# Clear failed
|
|
removed = queue.clear_failed()
|
|
assert removed == 1
|
|
|
|
# Check stats
|
|
stats = queue.get_stats()
|
|
assert stats["total"] == 0
|
|
assert stats["failed"] == 0
|