89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from app.pipeline import youtube
|
|
|
|
|
|
@pytest.fixture
|
|
def fresh_db(monkeypatch, tmp_path):
|
|
from app import db
|
|
monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "music.db"))
|
|
db.init_db()
|
|
return db
|
|
|
|
|
|
def _setup_token(db_module):
|
|
db_module.upsert_oauth_token(
|
|
channel_id="UC1", channel_title="t", avatar_url=None,
|
|
refresh_token="r1", access_token="a1", expires_at="2099-01-01T00:00:00",
|
|
)
|
|
|
|
|
|
@patch("app.pipeline.youtube._build_youtube_client")
|
|
def test_upload_succeeds_after_resumable(mock_client, fresh_db, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("YOUTUBE_OAUTH_CLIENT_ID", "cid")
|
|
monkeypatch.setenv("YOUTUBE_OAUTH_CLIENT_SECRET", "sec")
|
|
_setup_token(fresh_db)
|
|
|
|
yt = MagicMock()
|
|
insert = MagicMock()
|
|
# next_chunk: first call returns (None, None), second returns (None, response with id)
|
|
insert.next_chunk.side_effect = [(None, None), (None, {"id": "VID123"})]
|
|
yt.videos().insert.return_value = insert
|
|
mock_client.return_value = yt
|
|
|
|
video_path = tmp_path / "v.mp4"
|
|
video_path.write_bytes(b"\x00" * 100)
|
|
out = youtube.upload_video(
|
|
video_path=str(video_path),
|
|
thumbnail_path=None,
|
|
metadata={"title": "T", "description": "D", "tags": ["x"], "category_id": 10},
|
|
privacy="private",
|
|
)
|
|
assert out["video_id"] == "VID123"
|
|
|
|
|
|
def test_upload_no_token_raises(fresh_db, tmp_path):
|
|
video_path = tmp_path / "v.mp4"
|
|
video_path.write_bytes(b"\x00")
|
|
with pytest.raises(youtube.NotAuthenticatedError):
|
|
youtube.upload_video(
|
|
video_path=str(video_path), thumbnail_path=None,
|
|
metadata={"title":"T","description":"D","tags":[],"category_id":10},
|
|
privacy="private",
|
|
)
|
|
|
|
|
|
@patch("app.pipeline.youtube._build_youtube_client")
|
|
def test_upload_quota_exceeded_marks_quota(mock_client, fresh_db, tmp_path, monkeypatch):
|
|
from googleapiclient.errors import HttpError
|
|
monkeypatch.setenv("YOUTUBE_OAUTH_CLIENT_ID", "cid")
|
|
monkeypatch.setenv("YOUTUBE_OAUTH_CLIENT_SECRET", "sec")
|
|
_setup_token(fresh_db)
|
|
|
|
yt = MagicMock()
|
|
err = HttpError(MagicMock(status=403), b'{"error":{"errors":[{"reason":"quotaExceeded"}]}}')
|
|
insert_call = MagicMock()
|
|
insert_call.next_chunk.side_effect = err
|
|
yt.videos().insert.return_value = insert_call
|
|
mock_client.return_value = yt
|
|
|
|
video_path = tmp_path / "v.mp4"
|
|
video_path.write_bytes(b"\x00")
|
|
with pytest.raises(youtube.QuotaExceededError):
|
|
youtube.upload_video(
|
|
video_path=str(video_path), thumbnail_path=None,
|
|
metadata={"title":"T","description":"D","tags":[],"category_id":10},
|
|
privacy="private",
|
|
)
|
|
|
|
|
|
def test_get_status_returns_none_when_not_connected(fresh_db):
|
|
assert youtube.get_status() is None
|
|
|
|
|
|
def test_get_status_returns_channel_info(fresh_db):
|
|
_setup_token(fresh_db)
|
|
s = youtube.get_status()
|
|
assert s["channel_id"] == "UC1"
|
|
assert s["channel_title"] == "t"
|