myYouTube/tests/test_channel_activity.py

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}