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.
302 lines
11 KiB
Python
302 lines
11 KiB
Python
import base64
|
|
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.
|
|
yield TestClient(app)
|
|
del app.dependency_overrides[get_db]
|
|
del app.dependency_overrides[require_session]
|
|
|
|
|
|
BASE = datetime(2026, 9, 10, tzinfo=timezone.utc)
|
|
|
|
|
|
def _add_channel(db_session, youtube_channel_id, title, subscribed=True):
|
|
channel = Channel(youtube_channel_id=youtube_channel_id, title=title, subscribed=subscribed)
|
|
db_session.add(channel)
|
|
db_session.commit()
|
|
db_session.refresh(channel)
|
|
return channel
|
|
|
|
|
|
def _add_video(db_session, channel, youtube_video_id, published_at):
|
|
video = Video(
|
|
youtube_video_id=youtube_video_id,
|
|
channel_id=channel.id,
|
|
title=f"Video {youtube_video_id}",
|
|
published_at=published_at,
|
|
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 _add_category(db_session, name="Linux"):
|
|
category = Category(name=name, slug=name.lower(), sort_order=0)
|
|
db_session.add(category)
|
|
db_session.commit()
|
|
db_session.refresh(category)
|
|
return category
|
|
|
|
|
|
def _assign_category(db_session, channel, category):
|
|
db_session.execute(channel_categories.insert().values(channel_id=channel.id, category_id=category.id))
|
|
db_session.commit()
|
|
|
|
|
|
def _get(client, **params):
|
|
return client.get("/api/channels/activity", params=params)
|
|
|
|
|
|
def test_activity_orders_by_latest_video_freshness(client, db_session):
|
|
channel_a = _add_channel(db_session, "chanA", "Channel A")
|
|
channel_b = _add_channel(db_session, "chanB", "Channel B")
|
|
channel_c = _add_channel(db_session, "chanC", "Channel C")
|
|
|
|
# Interleave videos so insertion order differs from freshness.
|
|
_add_video(db_session, channel_c, "cFresh", BASE + timedelta(hours=5))
|
|
_add_video(db_session, channel_b, "bFresh", BASE + timedelta(hours=20))
|
|
_add_video(db_session, channel_a, "aFresh", BASE + timedelta(hours=10))
|
|
_add_video(db_session, channel_c, "cOld", BASE + timedelta(hours=1))
|
|
_add_video(db_session, channel_b, "bOld", BASE + timedelta(hours=8))
|
|
_add_video(db_session, channel_a, "aOld", BASE + timedelta(hours=6))
|
|
|
|
resp = _get(client)
|
|
assert resp.status_code == 200
|
|
payload = resp.json()
|
|
|
|
assert [item["channel"]["id"] for item in payload["items"]] == [
|
|
channel_b.id,
|
|
channel_a.id,
|
|
channel_c.id,
|
|
]
|
|
assert payload["next_cursor"] is None
|
|
|
|
|
|
def test_activity_puts_channels_without_videos_last(client, db_session):
|
|
channel_with = _add_channel(db_session, "chanWith", "With videos")
|
|
channel_empty_low = _add_channel(db_session, "chanEmptyLow", "Empty low")
|
|
_add_video(db_session, channel_with, "w1", BASE + timedelta(hours=1))
|
|
channel_empty_high = _add_channel(db_session, "chanEmptyHigh", "Empty high")
|
|
|
|
resp = _get(client)
|
|
payload = resp.json()
|
|
|
|
ids = [item["channel"]["id"] for item in payload["items"]]
|
|
assert ids[0] == channel_with.id
|
|
# No-video channels follow all video channels, ordered by id desc.
|
|
assert ids[1] == channel_empty_high.id
|
|
assert ids[2] == channel_empty_low.id
|
|
# No-video channel rows carry no videos.
|
|
assert [len(item["videos"]) for item in payload["items"]] == [1, 0, 0]
|
|
|
|
|
|
def test_activity_returns_exactly_three_latest_videos_per_channel(client, db_session):
|
|
channel = _add_channel(db_session, "chanA", "Channel A")
|
|
for i in range(1, 5):
|
|
_add_video(db_session, channel, f"v{i}", BASE + timedelta(hours=i))
|
|
|
|
payload = _get(client).json()
|
|
|
|
item = payload["items"][0]
|
|
assert [v["youtube_video_id"] for v in item["videos"]] == ["v4", "v3", "v2"]
|
|
assert "v1" not in [v["youtube_video_id"] for v in item["videos"]]
|
|
|
|
|
|
def test_activity_filters_by_category(client, db_session):
|
|
category = _add_category(db_session)
|
|
channel_in = _add_channel(db_session, "chanIn", "In category")
|
|
channel_out = _add_channel(db_session, "chanOut", "Out of category")
|
|
_assign_category(db_session, channel_in, category)
|
|
_add_video(db_session, channel_in, "inVid", BASE + timedelta(hours=3))
|
|
_add_video(db_session, channel_out, "outVid", BASE + timedelta(hours=4))
|
|
|
|
payload = _get(client, category_id=category.id).json()
|
|
|
|
assert [item["channel"]["id"] for item in payload["items"]] == [channel_in.id]
|
|
|
|
payload_uncat = _get(client, uncategorized="true").json()
|
|
assert [item["channel"]["id"] for item in payload_uncat["items"]] == [channel_out.id]
|
|
|
|
|
|
def test_activity_includes_only_subscribed_channels(client, db_session):
|
|
subscribed = _add_channel(db_session, "chanSub", "Subscribed")
|
|
unsubscribed = _add_channel(db_session, "chanUnsub", "Unsubscribed", subscribed=False)
|
|
_add_video(db_session, subscribed, "sVid", BASE + timedelta(hours=1))
|
|
_add_video(db_session, unsubscribed, "uVid", BASE + timedelta(hours=10))
|
|
|
|
payload = _get(client).json()
|
|
|
|
assert [item["channel"]["id"] for item in payload["items"]] == [subscribed.id]
|
|
|
|
|
|
def test_activity_pagination_across_videos_to_no_videos_boundary(client, db_session):
|
|
# Three channels with videos (distinct freshness), two without.
|
|
c3 = _add_channel(db_session, "chanC3", "C3")
|
|
c2 = _add_channel(db_session, "chanC2", "C2")
|
|
c1 = _add_channel(db_session, "chanC1", "C1")
|
|
_add_video(db_session, c1, "c1Vid", BASE + timedelta(hours=1))
|
|
_add_video(db_session, c2, "c2Vid", BASE + timedelta(hours=2))
|
|
_add_video(db_session, c3, "c3Vid", BASE + timedelta(hours=3))
|
|
empty_high = _add_channel(db_session, "chanEHigh", "Empty high")
|
|
empty_low = _add_channel(db_session, "chanELow", "Empty low")
|
|
|
|
page1 = _get(client, limit=2).json()
|
|
assert [i["channel"]["id"] for i in page1["items"]] == [c3.id, c2.id]
|
|
assert page1["next_cursor"] is not None
|
|
|
|
page2 = _get(client, limit=2, cursor=page1["next_cursor"]).json()
|
|
assert [i["channel"]["id"] for i in page2["items"]] == [c1.id, empty_low.id]
|
|
assert page2["next_cursor"] is not None
|
|
|
|
page3 = _get(client, limit=2, cursor=page2["next_cursor"]).json()
|
|
assert [i["channel"]["id"] for i in page3["items"]] == [empty_high.id]
|
|
assert page3["next_cursor"] is None
|
|
|
|
all_ids = [
|
|
i["channel"]["id"] for page in (page1, page2, page3) for i in page["items"]
|
|
]
|
|
assert len(all_ids) == len(set(all_ids)) == 5
|
|
assert set(all_ids) == {c1.id, c2.id, c3.id, empty_high.id, empty_low.id}
|
|
|
|
|
|
def test_activity_pagination_boundary_exact_at_last_video_channel(client, db_session):
|
|
# The page boundary lands exactly on the last channel that has videos,
|
|
# so the next page must be the whole no-videos tail.
|
|
c2 = _add_channel(db_session, "chanC2", "C2")
|
|
c1 = _add_channel(db_session, "chanC1", "C1")
|
|
_add_video(db_session, c1, "c1Vid", BASE + timedelta(hours=1))
|
|
_add_video(db_session, c2, "c2Vid", BASE + timedelta(hours=2))
|
|
empty_high = _add_channel(db_session, "chanEHigh", "Empty high")
|
|
empty_low = _add_channel(db_session, "chanELow", "Empty low")
|
|
|
|
page1 = _get(client, limit=2).json()
|
|
assert [i["channel"]["id"] for i in page1["items"]] == [c2.id, c1.id]
|
|
assert page1["next_cursor"] is not None
|
|
|
|
page2 = _get(client, limit=2, cursor=page1["next_cursor"]).json()
|
|
assert [i["channel"]["id"] for i in page2["items"]] == [empty_low.id, empty_high.id]
|
|
assert page2["next_cursor"] is None
|
|
|
|
|
|
def test_activity_cursor_tiebreak_on_equal_latest_published_at(client, db_session):
|
|
channel_a = _add_channel(db_session, "chanA", "Channel A")
|
|
channel_b = _add_channel(db_session, "chanB", "Channel B")
|
|
tied = BASE + timedelta(hours=10)
|
|
_add_video(db_session, channel_a, "aTied", tied)
|
|
_add_video(db_session, channel_b, "bTied", tied)
|
|
|
|
page1 = _get(client, limit=1).json()
|
|
assert [i["channel"]["id"] for i in page1["items"]] == [channel_b.id]
|
|
assert page1["next_cursor"] is not None
|
|
|
|
page2 = _get(client, limit=1, cursor=page1["next_cursor"]).json()
|
|
assert [i["channel"]["id"] for i in page2["items"]] == [channel_a.id]
|
|
assert page2["next_cursor"] is None
|
|
|
|
|
|
def test_activity_item_shape(client, db_session):
|
|
channel = _add_channel(db_session, "chanA", "Channel A")
|
|
_add_video(db_session, channel, "shapeVid", BASE + timedelta(hours=1))
|
|
|
|
item = _get(client).json()["items"][0]
|
|
|
|
assert set(item.keys()) == {"channel", "videos"}
|
|
assert set(item["channel"].keys()) == {
|
|
"id",
|
|
"youtube_channel_id",
|
|
"title",
|
|
"description",
|
|
"thumbnail_url",
|
|
"uploads_playlist_id",
|
|
"subscriber_count",
|
|
"subscribed",
|
|
"last_synced_at",
|
|
"category_ids",
|
|
"new_videos_count",
|
|
}
|
|
assert item["channel"]["id"] == channel.id
|
|
assert item["channel"]["subscribed"] is True
|
|
assert len(item["videos"]) == 1
|
|
video = item["videos"][0]
|
|
assert set(video.keys()) == {
|
|
"youtube_video_id",
|
|
"title",
|
|
"description",
|
|
"channel",
|
|
"thumbnail_url",
|
|
"published_at",
|
|
"duration_seconds",
|
|
"is_short",
|
|
"youtube_url",
|
|
"categories",
|
|
"local",
|
|
}
|
|
assert set(video["channel"].keys()) == {"id", "youtube_channel_id", "title", "thumbnail_url"}
|
|
assert video["local"] == {
|
|
"available": False,
|
|
"status": "not_downloaded",
|
|
"progress_percent": None,
|
|
"media_url": None,
|
|
}
|
|
assert isinstance(video["categories"], list)
|
|
|
|
|
|
def test_activity_route_not_captured_by_channel_path_param(client, db_session):
|
|
channel = _add_channel(db_session, "chanA", "Channel A")
|
|
|
|
resp = _get(client)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["items"] != []
|
|
|
|
single = client.get(f"/api/channels/{channel.id}")
|
|
assert single.status_code == 200
|
|
assert single.json()["id"] == channel.id
|
|
|
|
|
|
def test_activity_invalid_cursor_returns_400(client, db_session):
|
|
_add_channel(db_session, "chanA", "Channel A")
|
|
|
|
garbage = base64.urlsafe_b64encode(b"hello|notanumber").decode()
|
|
for bad in ("not-base64!", garbage):
|
|
resp = _get(client, cursor=bad)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["detail"] == "Invalid cursor"
|
|
|
|
# An empty cursor is not a cursor at all (same as the feed endpoint).
|
|
assert _get(client, cursor="").status_code == 200
|
|
|
|
|
|
def test_activity_limit_out_of_bounds_returns_422(client, db_session):
|
|
_add_channel(db_session, "chanA", "Channel A")
|
|
|
|
assert _get(client, limit=0).status_code == 422
|
|
assert _get(client, limit=101).status_code == 422
|
|
assert _get(client, limit=100).status_code == 200
|
|
|
|
|
|
def test_activity_empty_when_no_channels(client):
|
|
payload = _get(client).json()
|
|
|
|
assert payload == {"items": [], "next_cursor": None}
|