252f38597d
- Detect shorts by duration (SHORTS_MAX_DURATION_SECONDS, default 180)
and expose is_short in feed, video and channel-activity DTOs.
- Feed gains type=all|long|short and anchor; lists show a two-mode
'Обычные | Shorts' filter (no 'Все') persisted in the URL.
- Clicking a short opens /shorts/🆔 a vertical scroll-snap feed with
autoplay for the active slide, context-aware endpoints and infinite
loading.
- Keep the sound choice across swipes, syncing with the player's own
mute control and guarding against the widget's stale isMuted() reads.
391 lines
14 KiB
Python
391 lines
14 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.auth_dependency import require_session
|
|
from app.db import get_db
|
|
from app.main import app
|
|
from app.models.category import Category
|
|
from app.models.channel import Channel
|
|
from app.models.channel_category import channel_categories
|
|
from app.models.video import Video
|
|
|
|
|
|
@pytest.fixture
|
|
def client(db_session):
|
|
def _get_db_override():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_db] = _get_db_override
|
|
app.dependency_overrides[require_session] = lambda: None
|
|
# Deliberately not using `with TestClient(app)`: that runs the app's
|
|
# lifespan, which would try to reach the real MeTube instance and DB
|
|
# (see AGENTS.md: MeTube network calls in tests must be opt-in only).
|
|
yield TestClient(app)
|
|
del app.dependency_overrides[get_db]
|
|
del app.dependency_overrides[require_session]
|
|
|
|
|
|
def test_get_video_by_youtube_id(client, db_session):
|
|
channel = Channel(youtube_channel_id="chanA", title="Channel A", subscribed=True)
|
|
db_session.add(channel)
|
|
db_session.commit()
|
|
|
|
category = Category(name="Linux", slug="linux", sort_order=0)
|
|
db_session.add(category)
|
|
db_session.commit()
|
|
db_session.execute(channel_categories.insert().values(channel_id=channel.id, category_id=category.id))
|
|
db_session.commit()
|
|
|
|
video = Video(
|
|
youtube_video_id="vid1",
|
|
channel_id=channel.id,
|
|
title="Video One",
|
|
description="desc",
|
|
published_at=datetime(2026, 9, 10, tzinfo=timezone.utc),
|
|
duration_seconds=125,
|
|
youtube_url="https://www.youtube.com/watch?v=vid1",
|
|
)
|
|
db_session.add(video)
|
|
db_session.commit()
|
|
|
|
resp = client.get("/api/videos/vid1")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["youtube_video_id"] == "vid1"
|
|
assert body["channel"]["id"] == channel.id
|
|
assert body["categories"] == [{"id": category.id, "name": "Linux"}]
|
|
assert body["local"]["available"] is False
|
|
# 125s <= default 180s threshold.
|
|
assert body["is_short"] is True
|
|
|
|
|
|
def test_get_video_is_short_false_for_long_and_unknown_duration(client, db_session, monkeypatch):
|
|
_seed_single_video(db_session, youtube_video_id="vidLong", duration_seconds=3600)
|
|
_seed_single_video(db_session, youtube_video_id="vidUnknown")
|
|
|
|
assert client.get("/api/videos/vidLong").json()["is_short"] is False
|
|
assert client.get("/api/videos/vidUnknown").json()["is_short"] is False
|
|
|
|
# The threshold is read from settings at serialization time.
|
|
monkeypatch.setattr("app.config.settings.shorts_max_duration_seconds", 4000)
|
|
assert client.get("/api/videos/vidLong").json()["is_short"] is True
|
|
assert client.get("/api/videos/vidUnknown").json()["is_short"] is False
|
|
|
|
|
|
def test_get_video_not_found(client):
|
|
resp = client.get("/api/videos/missing")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def _seed_single_video(db_session, youtube_video_id="vid1", duration_seconds=None):
|
|
channel = db_session.query(Channel).filter(Channel.youtube_channel_id == "chanA").first()
|
|
if channel is None:
|
|
channel = Channel(youtube_channel_id="chanA", title="Channel A", subscribed=True)
|
|
db_session.add(channel)
|
|
db_session.commit()
|
|
|
|
video = Video(
|
|
youtube_video_id=youtube_video_id,
|
|
channel_id=channel.id,
|
|
title="Video One",
|
|
published_at=datetime(2026, 9, 10, tzinfo=timezone.utc),
|
|
duration_seconds=duration_seconds,
|
|
youtube_url=f"https://www.youtube.com/watch?v={youtube_video_id}",
|
|
)
|
|
db_session.add(video)
|
|
db_session.commit()
|
|
db_session.refresh(video)
|
|
return video
|
|
|
|
|
|
def test_download_video_enqueues(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.enqueue_video", lambda self, u, p: {"status": "ok"}
|
|
)
|
|
|
|
resp = client.post(f"/api/videos/{video.youtube_video_id}/download")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "queued"
|
|
|
|
|
|
def test_download_video_metube_rejects(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.enqueue_video",
|
|
lambda self, u, p: {"status": "error", "msg": "bad url"},
|
|
)
|
|
|
|
resp = client.post(f"/api/videos/{video.youtube_video_id}/download")
|
|
|
|
assert resp.status_code == 502
|
|
|
|
|
|
def test_download_video_second_click_returns_same_job(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.enqueue_video", lambda self, u, p: {"status": "ok"}
|
|
)
|
|
|
|
r1 = client.post(f"/api/videos/{video.youtube_video_id}/download").json()
|
|
r2 = client.post(f"/api/videos/{video.youtube_video_id}/download").json()
|
|
|
|
assert r1["requested_at"] == r2["requested_at"]
|
|
|
|
from app.models.download_job import DownloadJob
|
|
|
|
assert db_session.query(DownloadJob).count() == 1
|
|
|
|
|
|
def test_download_status_not_downloaded(client, db_session):
|
|
video = _seed_single_video(db_session)
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
assert resp.json()["status"] == "not_downloaded"
|
|
|
|
|
|
def _seed_job_with_age(db_session, video, status="queued", age_minutes=10):
|
|
"""Seeds a job whose requested_at/updated_at are both `age_minutes` in the
|
|
past (aware datetimes -- see AGENTS.md: SQLite stores them naive, the
|
|
production code normalizes them before comparing)."""
|
|
from app.models.download_job import DownloadJob
|
|
|
|
old = datetime.now(timezone.utc) - timedelta(minutes=age_minutes)
|
|
job = DownloadJob(video_id=video.id, status=status, requested_at=old, updated_at=old)
|
|
db_session.add(job)
|
|
db_session.commit()
|
|
return job
|
|
|
|
|
|
def test_download_status_self_heals_stale_job_to_completed(client, db_session, monkeypatch):
|
|
"""Bug 1: a job stuck in 'queued' because the terminal Socket.IO event was
|
|
missed is reconciled against MeTube history on poll."""
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="queued", age_minutes=10)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.fetch_history",
|
|
lambda self: {
|
|
"queue": [],
|
|
"pending": [],
|
|
"done": [{"id": "vid1.vid1", "url": video.youtube_url, "status": "finished", "filename": "f.mp4"}],
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.build_media_url",
|
|
lambda self, filename: f"http://metube.local/download/{filename}",
|
|
)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "completed"
|
|
assert body["media_url"] == "http://metube.local/download/f.mp4"
|
|
|
|
|
|
def test_download_status_self_heals_stale_job_to_failed(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="queued", age_minutes=10)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.fetch_history",
|
|
lambda self: {
|
|
"queue": [],
|
|
"pending": [],
|
|
"done": [{"id": "vid1.vid1", "url": video.youtube_url, "status": "error", "msg": "HTTP 429 bot check"}],
|
|
},
|
|
)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "failed"
|
|
assert body["error_message"] == "HTTP 429 bot check"
|
|
assert body["media_url"] is None
|
|
|
|
|
|
def test_download_status_stale_job_still_in_queue_stays_active(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="queued", age_minutes=10)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.fetch_history",
|
|
lambda self: {
|
|
"queue": [{"id": "vid1.vid1", "url": video.youtube_url, "status": "pending"}],
|
|
"pending": [],
|
|
"done": [],
|
|
},
|
|
)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "queued"
|
|
|
|
|
|
def test_download_status_stale_job_not_in_history_is_unknown(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="queued", age_minutes=10)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.fetch_history",
|
|
lambda self: {"queue": [], "pending": [], "done": []},
|
|
)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "unknown"
|
|
|
|
|
|
def test_download_status_fresh_job_does_not_fetch_history(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="queued", age_minutes=0)
|
|
|
|
def fail_if_called(self):
|
|
raise AssertionError("fetch_history must not be called for a fresh job")
|
|
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.fetch_history", fail_if_called)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "queued"
|
|
|
|
|
|
def test_download_status_terminal_job_does_not_fetch_history(client, db_session, monkeypatch):
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="completed", age_minutes=10)
|
|
|
|
def fail_if_called(self):
|
|
raise AssertionError("fetch_history must not be called for a terminal job")
|
|
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.fetch_history", fail_if_called)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "completed"
|
|
|
|
|
|
def test_download_status_fetch_error_keeps_status(client, db_session, monkeypatch):
|
|
"""A transient history-fetch failure must not bubble up nor change the
|
|
job's status -- the poll response keeps the last known state."""
|
|
video = _seed_single_video(db_session)
|
|
_seed_job_with_age(db_session, video, status="queued", age_minutes=10)
|
|
|
|
def raise_error(self):
|
|
raise RuntimeError("connection refused")
|
|
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.fetch_history", raise_error)
|
|
|
|
resp = client.get(f"/api/videos/{video.youtube_video_id}/download-status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "queued"
|
|
|
|
|
|
def test_recheck_local_downgrades_to_unknown_when_media_missing(client, db_session, monkeypatch):
|
|
from app.models.download_job import DownloadJob
|
|
|
|
video = _seed_single_video(db_session)
|
|
job = DownloadJob(video_id=video.id, status="completed", media_url="http://metube.local/download/f.mp4")
|
|
db_session.add(job)
|
|
db_session.commit()
|
|
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.check_media", lambda self, url: False)
|
|
|
|
resp = client.post(f"/api/videos/{video.youtube_video_id}/recheck-local")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "unknown"
|
|
|
|
|
|
def test_recheck_local_keeps_completed_when_media_reachable(client, db_session, monkeypatch):
|
|
from app.models.download_job import DownloadJob
|
|
|
|
video = _seed_single_video(db_session)
|
|
job = DownloadJob(video_id=video.id, status="completed", media_url="http://metube.local/download/f.mp4")
|
|
db_session.add(job)
|
|
db_session.commit()
|
|
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.check_media", lambda self, url: True)
|
|
|
|
resp = client.post(f"/api/videos/{video.youtube_video_id}/recheck-local")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "completed"
|
|
|
|
|
|
def test_delete_download_requires_completed_job(client, db_session):
|
|
video = _seed_single_video(db_session)
|
|
resp = client.delete(f"/api/videos/{video.youtube_video_id}/download")
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_delete_download_success(client, db_session, monkeypatch):
|
|
from app.models.download_job import DownloadJob
|
|
|
|
video = _seed_single_video(db_session)
|
|
job = DownloadJob(
|
|
video_id=video.id,
|
|
status="completed",
|
|
metube_job_id="vid1.vid1",
|
|
media_url="http://metube.local/download/f.mp4",
|
|
)
|
|
db_session.add(job)
|
|
db_session.commit()
|
|
|
|
calls = {}
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.delete_download",
|
|
lambda self, youtube_url: calls.setdefault("youtube_url", youtube_url) or {"status": "ok"},
|
|
)
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.check_media", lambda self, url: False)
|
|
|
|
resp = client.delete(f"/api/videos/{video.youtube_video_id}/download")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "deleted"
|
|
assert resp.json()["media_url"] is None
|
|
# MeTube's queue/done stores are keyed by URL, not by the id it reports
|
|
# over Socket.IO -- passing the wrong one is exactly the bug this
|
|
# regression test guards against (it silently no-ops on MeTube's side).
|
|
assert calls["youtube_url"] == video.youtube_url
|
|
|
|
db_session.refresh(job)
|
|
assert job.status == "deleted"
|
|
assert job.media_url is None
|
|
|
|
|
|
def test_delete_download_file_still_present_returns_409(client, db_session, monkeypatch):
|
|
"""MeTube can accept /delete and still leave the file on disk if its own
|
|
DELETE_FILE_ON_TRASHCAN config is off -- we must not lie about that."""
|
|
from app.models.download_job import DownloadJob
|
|
|
|
video = _seed_single_video(db_session)
|
|
job = DownloadJob(
|
|
video_id=video.id,
|
|
status="completed",
|
|
metube_job_id="vid1.vid1",
|
|
media_url="http://metube.local/download/f.mp4",
|
|
)
|
|
db_session.add(job)
|
|
db_session.commit()
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.metube_client.MeTubeClient.delete_download", lambda self, youtube_url: {"status": "ok"}
|
|
)
|
|
monkeypatch.setattr("app.services.metube_client.MeTubeClient.check_media", lambda self, url: True)
|
|
|
|
resp = client.delete(f"/api/videos/{video.youtube_video_id}/download")
|
|
|
|
assert resp.status_code == 409
|
|
|
|
db_session.refresh(job)
|
|
assert job.status == "completed"
|
|
assert job.media_url == "http://metube.local/download/f.mp4"
|