237 lines
6.9 KiB
Python
237 lines
6.9 KiB
Python
"""
|
|
Tests for database models
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.core.database import Base
|
|
from app.models.playlist import PlaylistSubscription
|
|
from app.models.video import VideoRecord, VideoStatus
|
|
|
|
|
|
@pytest.fixture
|
|
def test_db():
|
|
"""Create a test database session"""
|
|
# Create in-memory SQLite database
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
db = SessionLocal()
|
|
|
|
yield db
|
|
|
|
db.close()
|
|
|
|
|
|
def test_playlist_creation(test_db):
|
|
"""Test playlist creation"""
|
|
playlist = PlaylistSubscription(
|
|
url="https://www.youtube.com/playlist?list=TEST123",
|
|
title="Test Playlist",
|
|
check_interval=60,
|
|
quality="best",
|
|
format="mp4",
|
|
enabled=True
|
|
)
|
|
|
|
test_db.add(playlist)
|
|
test_db.commit()
|
|
test_db.refresh(playlist)
|
|
|
|
assert playlist.id is not None
|
|
assert playlist.url == "https://www.youtube.com/playlist?list=TEST123"
|
|
assert playlist.title == "Test Playlist"
|
|
assert playlist.check_interval == 60
|
|
assert playlist.enabled is True
|
|
assert playlist.created_at is not None
|
|
assert playlist.updated_at is not None
|
|
|
|
|
|
def test_video_creation(test_db):
|
|
"""Test video creation"""
|
|
# First create a playlist
|
|
playlist = PlaylistSubscription(
|
|
url="https://www.youtube.com/playlist?list=TEST123",
|
|
title="Test Playlist"
|
|
)
|
|
test_db.add(playlist)
|
|
test_db.commit()
|
|
test_db.refresh(playlist)
|
|
|
|
# Create a video
|
|
video = VideoRecord(
|
|
playlist_id=playlist.id,
|
|
video_url="https://www.youtube.com/watch?v=TESTVIDEO",
|
|
video_id="TESTVIDEO",
|
|
title="Test Video",
|
|
playlist_index=1,
|
|
status=VideoStatus.PENDING
|
|
)
|
|
|
|
test_db.add(video)
|
|
test_db.commit()
|
|
test_db.refresh(video)
|
|
|
|
assert video.id is not None
|
|
assert video.playlist_id == playlist.id
|
|
assert video.video_id == "TESTVIDEO"
|
|
assert video.title == "Test Video"
|
|
assert video.status == VideoStatus.PENDING
|
|
assert video.created_at is not None
|
|
|
|
|
|
def test_playlist_video_relationship(test_db):
|
|
"""Test playlist-video relationship"""
|
|
# Create playlist
|
|
playlist = PlaylistSubscription(
|
|
url="https://www.youtube.com/playlist?list=TEST123",
|
|
title="Test Playlist"
|
|
)
|
|
test_db.add(playlist)
|
|
test_db.commit()
|
|
test_db.refresh(playlist)
|
|
|
|
# Create videos
|
|
video1 = VideoRecord(
|
|
playlist_id=playlist.id,
|
|
video_url="https://www.youtube.com/watch?v=VIDEO1",
|
|
video_id="VIDEO1",
|
|
title="Video 1",
|
|
playlist_index=1
|
|
)
|
|
|
|
video2 = VideoRecord(
|
|
playlist_id=playlist.id,
|
|
video_url="https://www.youtube.com/watch?v=VIDEO2",
|
|
video_id="VIDEO2",
|
|
title="Video 2",
|
|
playlist_index=2
|
|
)
|
|
|
|
test_db.add_all([video1, video2])
|
|
test_db.commit()
|
|
|
|
# Test relationship
|
|
assert len(playlist.videos) == 2
|
|
assert playlist.videos[0].title == "Video 1"
|
|
assert playlist.videos[1].title == "Video 2"
|
|
|
|
# Test cascade delete
|
|
test_db.delete(playlist)
|
|
test_db.commit()
|
|
|
|
# Videos should be deleted due to cascade
|
|
remaining_videos = test_db.query(VideoRecord).all()
|
|
assert len(remaining_videos) == 0
|
|
|
|
|
|
def test_video_status_methods(test_db):
|
|
"""Test video status management methods"""
|
|
# Create playlist and video
|
|
playlist = PlaylistSubscription(url="https://www.youtube.com/playlist?list=TEST123")
|
|
test_db.add(playlist)
|
|
test_db.commit()
|
|
test_db.refresh(playlist)
|
|
|
|
video = VideoRecord(
|
|
playlist_id=playlist.id,
|
|
video_url="https://www.youtube.com/watch?v=TESTVIDEO",
|
|
video_id="TESTVIDEO",
|
|
status=VideoStatus.PENDING
|
|
)
|
|
test_db.add(video)
|
|
test_db.commit()
|
|
test_db.refresh(video)
|
|
|
|
# Test downloading
|
|
video.mark_as_downloading("metube_123")
|
|
assert video.status == VideoStatus.DOWNLOADING
|
|
assert video.metube_download_id == "metube_123"
|
|
assert video.download_requested_at is not None
|
|
|
|
# Test completion
|
|
video.mark_as_completed("test_video.mp4")
|
|
assert video.status == VideoStatus.COMPLETED
|
|
assert video.download_completed_at is not None
|
|
assert video.original_filename == "test_video.mp4"
|
|
assert video.retry_count == 0
|
|
|
|
# Test failure
|
|
video.mark_as_failed("Network error")
|
|
assert video.status == VideoStatus.FAILED
|
|
assert video.error_message == "Network error"
|
|
assert video.last_error_at is not None
|
|
assert video.retry_count == 1
|
|
|
|
# Test reset
|
|
video.reset_to_pending()
|
|
assert video.status == VideoStatus.PENDING
|
|
assert video.metube_download_id is None
|
|
assert video.error_message is None
|
|
assert video.retry_count == 0
|
|
|
|
|
|
def test_playlist_should_check(test_db):
|
|
"""Test playlist check logic"""
|
|
playlist = PlaylistSubscription(
|
|
url="https://www.youtube.com/playlist?list=TEST123",
|
|
check_interval=60, # 60 minutes
|
|
enabled=True
|
|
)
|
|
|
|
# Should check if never checked
|
|
assert playlist.should_check() is True
|
|
|
|
# Should check if last checked was long ago
|
|
playlist.last_checked = datetime.utcnow() - timedelta(minutes=61)
|
|
assert playlist.should_check() is True
|
|
|
|
# Should not check if recently checked
|
|
playlist.last_checked = datetime.utcnow() - timedelta(minutes=30)
|
|
assert playlist.should_check() is False
|
|
|
|
# Should not check if disabled
|
|
playlist.enabled = False
|
|
assert playlist.should_check() is False
|
|
|
|
|
|
def test_video_properties(test_db):
|
|
"""Test video computed properties"""
|
|
playlist = PlaylistSubscription(url="https://www.youtube.com/playlist?list=TEST123")
|
|
test_db.add(playlist)
|
|
test_db.commit()
|
|
test_db.refresh(playlist)
|
|
|
|
video = VideoRecord(
|
|
playlist_id=playlist.id,
|
|
video_url="https://www.youtube.com/watch?v=TESTVIDEO",
|
|
video_id="TESTVIDEO",
|
|
status=VideoStatus.PENDING
|
|
)
|
|
test_db.add(video)
|
|
test_db.commit()
|
|
test_db.refresh(video)
|
|
|
|
# Test downloadable property
|
|
assert video.is_downloadable is True
|
|
|
|
video.status = VideoStatus.COMPLETED
|
|
assert video.is_downloadable is False
|
|
|
|
# Test completed property
|
|
assert video.is_completed is True
|
|
|
|
video.status = VideoStatus.PENDING
|
|
assert video.is_completed is False
|
|
|
|
# Test can_retry property
|
|
video.status = VideoStatus.FAILED
|
|
video.retry_count = 0
|
|
assert video.can_retry() is True
|
|
|
|
video.retry_count = 3
|
|
assert video.can_retry() is False |