293 lines
9.0 KiB
Python
293 lines
9.0 KiB
Python
"""
|
|
Tests for the /api/sources/ router.
|
|
|
|
Covers:
|
|
- Listing sources
|
|
- Creating local and YouTube-type sources
|
|
- Syncing a YouTube source (mocked yt-dlp call)
|
|
- Deleting a source
|
|
- Attempting to sync a non-YouTube source (should 400)
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from django.test import Client
|
|
from core.models import AppUser, Library, MediaSource, MediaItem
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return Client()
|
|
|
|
|
|
@pytest.fixture
|
|
def user(db):
|
|
return AppUser.objects.create_user(
|
|
username="srcuser", email="src@pytv.local", password="password"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def library(db, user):
|
|
return Library.objects.create(
|
|
owner_user=user,
|
|
name="Source Test Library",
|
|
visibility="public",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def local_source(db, library):
|
|
return MediaSource.objects.create(
|
|
library=library,
|
|
name="Local Movies",
|
|
source_type=MediaSource.SourceType.LOCAL_DIRECTORY,
|
|
uri="/mnt/movies",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def youtube_source(db, library):
|
|
return MediaSource.objects.create(
|
|
library=library,
|
|
name="ABC News",
|
|
source_type=MediaSource.SourceType.YOUTUBE_CHANNEL,
|
|
uri="https://www.youtube.com/@ABCNews",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Listing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.django_db
|
|
def test_list_sources_empty(client):
|
|
"""Listing when no sources exist returns an empty array."""
|
|
response = client.get("/api/sources/")
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_list_sources_returns_all(client, local_source, youtube_source):
|
|
response = client.get("/api/sources/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
names = {s["name"] for s in data}
|
|
assert names == {"Local Movies", "ABC News"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Creation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_local_source(client, library):
|
|
payload = {
|
|
"library_id": library.id,
|
|
"name": "My Movies",
|
|
"source_type": "local_directory",
|
|
"uri": "/mnt/media/movies",
|
|
"is_active": True,
|
|
}
|
|
response = client.post(
|
|
"/api/sources/", data=payload, content_type="application/json"
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["name"] == "My Movies"
|
|
assert data["source_type"] == "local_directory"
|
|
assert MediaSource.objects.filter(id=data["id"]).exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_youtube_channel_source(client, library):
|
|
"""Registering a YouTube channel URL as a media source."""
|
|
payload = {
|
|
"library_id": library.id,
|
|
"name": "ABC News",
|
|
"source_type": "youtube_channel",
|
|
"uri": "https://www.youtube.com/@ABCNews",
|
|
"is_active": True,
|
|
}
|
|
response = client.post(
|
|
"/api/sources/", data=payload, content_type="application/json"
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["source_type"] == "youtube_channel"
|
|
assert "@ABCNews" in data["uri"]
|
|
|
|
# Verify it's in the DB with correct type
|
|
src = MediaSource.objects.get(id=data["id"])
|
|
assert src.source_type == MediaSource.SourceType.YOUTUBE_CHANNEL
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_youtube_playlist_source(client, library):
|
|
"""Registering a YouTube playlist URL as a media source."""
|
|
payload = {
|
|
"library_id": library.id,
|
|
"name": "Tech Talks",
|
|
"source_type": "youtube_playlist",
|
|
"uri": "https://www.youtube.com/playlist?list=PLFgquLnL59akA2PflFpeQG9L01VFg90wS",
|
|
"is_active": True,
|
|
}
|
|
response = client.post(
|
|
"/api/sources/", data=payload, content_type="application/json"
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["source_type"] == "youtube_playlist"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_create_source_invalid_library(client):
|
|
"""Creating a source with a non-existent library should 404."""
|
|
payload = {
|
|
"library_id": 99999,
|
|
"name": "Orphan",
|
|
"source_type": "local_directory",
|
|
"uri": "/nope",
|
|
}
|
|
response = client.post(
|
|
"/api/sources/", data=payload, content_type="application/json"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Retrieval
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.django_db
|
|
def test_get_source(client, youtube_source):
|
|
response = client.get(f"/api/sources/{youtube_source.id}")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == youtube_source.id
|
|
assert data["source_type"] == "youtube_channel"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_get_source_not_found(client):
|
|
response = client.get("/api/sources/99999")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Deletion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.django_db
|
|
def test_delete_source(client, local_source):
|
|
source_id = local_source.id
|
|
response = client.delete(f"/api/sources/{source_id}")
|
|
assert response.status_code == 204
|
|
assert not MediaSource.objects.filter(id=source_id).exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_delete_source_not_found(client):
|
|
response = client.delete("/api/sources/99999")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sync (mocked yt-dlp)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
MOCK_YT_ENTRIES = [
|
|
{
|
|
"id": "abc123",
|
|
"title": "Breaking News: Test Story",
|
|
"duration": 180,
|
|
"thumbnail": "https://i.ytimg.com/vi/abc123/hqdefault.jpg",
|
|
"description": "A breaking news segment.",
|
|
"upload_date": "20240315",
|
|
"url": "https://www.youtube.com/watch?v=abc123",
|
|
"uploader": "ABC News",
|
|
},
|
|
{
|
|
"id": "def456",
|
|
"title": "Weather Report: Sunny with a chance of clouds",
|
|
"duration": 90,
|
|
"thumbnail": "https://i.ytimg.com/vi/def456/hqdefault.jpg",
|
|
"description": "Evening weather.",
|
|
"upload_date": "20240316",
|
|
"url": "https://www.youtube.com/watch?v=def456",
|
|
"uploader": "ABC News",
|
|
},
|
|
]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_sync_youtube_channel_creates_media_items(client, youtube_source):
|
|
"""
|
|
Syncing a YouTube channel should call yt-dlp (mocked) and upsert
|
|
MediaItem rows for each discovered video.
|
|
"""
|
|
with patch("core.services.youtube._extract_playlist_info", return_value=MOCK_YT_ENTRIES):
|
|
response = client.post(f"/api/sources/{youtube_source.id}/sync")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Two new videos → created=2
|
|
assert data["created"] == 2
|
|
assert data["updated"] == 0
|
|
assert data["skipped"] == 0
|
|
|
|
# MediaItems should now exist in the DB
|
|
items = MediaItem.objects.filter(media_source=youtube_source)
|
|
assert items.count() == 2
|
|
|
|
titles = {i.title for i in items}
|
|
assert "Breaking News: Test Story" in titles
|
|
assert "Weather Report: Sunny with a chance of clouds" in titles
|
|
|
|
# Verify youtube_video_id is populated
|
|
item = items.get(youtube_video_id="abc123")
|
|
assert item.runtime_seconds == 180
|
|
assert item.release_year == 2024
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_sync_youtube_channel_updates_existing(client, youtube_source):
|
|
"""Re-syncing the same source updates existing rows rather than duplicating."""
|
|
with patch("core.services.youtube._extract_playlist_info", return_value=MOCK_YT_ENTRIES):
|
|
client.post(f"/api/sources/{youtube_source.id}/sync")
|
|
|
|
# Second sync — same entries, should update not create
|
|
with patch(
|
|
"core.services.youtube._extract_playlist_info",
|
|
return_value=[{**MOCK_YT_ENTRIES[0], "title": "Updated Title"}],
|
|
):
|
|
response = client.post(f"/api/sources/{youtube_source.id}/sync")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["created"] == 0
|
|
assert data["updated"] == 1
|
|
|
|
item = MediaItem.objects.get(youtube_video_id="abc123")
|
|
assert item.title == "Updated Title"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_sync_non_youtube_source_returns_400(client, local_source):
|
|
"""Syncing a non-YouTube source type should return HTTP 400."""
|
|
response = client.post(f"/api/sources/{local_source.id}/sync")
|
|
assert response.status_code == 400
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_sync_source_not_found(client):
|
|
response = client.post("/api/sources/99999/sync")
|
|
assert response.status_code == 404
|