d4950f6979
New GET /api/channels/activity returns subscribed channels sorted by their latest video (no-video channels last) with the 3 newest videos per channel and cursor pagination. The feed page gains a 'Лента | Каналы' toggle (?view=channels) with compact video cards per channel row; category filters and the sidebar preserve the view.
301 lines
11 KiB
Python
301 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",
|
|
"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}
|