ytstream/tests/test_web.py

"""The admin UI: rendering, auth gating and CSRF.

The templates are f-strings over a dict, so a renamed key is a KeyError at render
time rather than a type error at import time — which is exactly the kind of
breakage that only shows up when somebody opens the page. These tests render every
page for real.
"""

from __future__ import annotations

import pytest

from ytstream import videos
from ytstream.web import auth, templates

from conftest import add_channel, add_video


@pytest.fixture()
def channel_row(conn, settings, channel):
    """The dict shape server.py builds for the channel table."""
    add_video(conn, channel["id"], "vid00000001", state=videos.MATERIALISED)
    return {
        "id": channel["id"],
        "title": channel["title"],
        "handle": channel["handle"],
        "channel_id": channel["channel_id"],
        "retention_days": None,
        "global_retention": 30,
        "last_polled_at": "2026-08-12T14:00:00+00:00",
        "last_poll_ok": 1,
        "consecutive_poll_failures": 0,
        "episodes": 1,
        "source": "youtube",
        "missing_syncs": 0,
        "missing_threshold": 3,
        "latest": "2026-08-12",
    }


def test_login_page_renders(settings):
    html = templates.login_page().decode()
    assert "ytstream" in html
    assert "youtube-automate" not in html
    assert 'type="password"' in html


def test_login_page_shows_an_error():
    assert "Wrong" in templates.login_page("Wrong password").decode()


def test_index_page_renders(channel_row):
    html = templates.index_page(
        channels=[channel_row], settings_values={"retention_days": "30"},
        settings_errors={}, csrf="tok", add_error=None, queue_depth=0,
    ).decode()

    assert "clabretro" in html
    assert "ytstream" in html
    # The provenance column replaced the meaningless size column.
    assert "youtube" in html
    assert "Size" not in html


def test_index_page_renders_with_no_channels():
    html = templates.index_page(
        channels=[], settings_values={}, settings_errors={}, csrf="tok",
        add_error=None, queue_depth=0,
    ).decode()
    assert "0 channel(s)" in html


def test_absence_badge_appears_before_the_channel_disappears(channel_row):
    """A channel counting towards removal must be visible while it still exists."""
    channel_row["missing_syncs"] = 2
    html = templates.index_page(
        channels=[channel_row], settings_values={}, settings_errors={}, csrf="tok",
        add_error=None, queue_depth=0,
    ).decode()
    assert "absent 2/3 syncs" in html


def test_no_absence_badge_when_healthy(channel_row):
    html = templates.index_page(
        channels=[channel_row], settings_values={}, settings_errors={}, csrf="tok",
        add_error=None, queue_depth=0,
    ).decode()
    assert "absent" not in html


def test_hostile_channel_title_is_escaped(channel_row):
    channel_row["title"] = '<script>alert("x")</script>'
    html = templates.index_page(
        channels=[channel_row], settings_values={}, settings_errors={}, csrf="tok",
        add_error=None, queue_depth=0,
    ).decode()
    assert "<script>alert" not in html
    assert "&lt;script&gt;" in html


def test_settings_errors_are_rendered_inline(channel_row):
    html = templates.index_page(
        channels=[channel_row], settings_values={"retention_days": "abc"},
        settings_errors={"retention_days": "must be a whole number"},
        csrf="tok", add_error=None, queue_depth=0,
    ).decode()
    assert "must be a whole number" in html


def test_masked_settings_are_not_rendered_in_plaintext(channel_row):
    """Both API keys are secrets; neither belongs in the HTML."""
    html = templates.index_page(
        channels=[channel_row],
        settings_values={"youtube_api_key": "AIzaSECRETVALUE",
                         "jellyfin_api_key": "JELLYSECRET"},
        settings_errors={}, csrf="tok", add_error=None, queue_depth=0,
    ).decode()
    assert "AIzaSECRETVALUE" not in html
    assert "JELLYSECRET" not in html


# ------------------------------------------------------------------------ csrf


def test_csrf_token_round_trips():
    secret = auth.new_secret()
    session = "session-token"
    token = auth.csrf_token(secret, session)
    assert auth.verify_csrf(secret, session, token)


def test_csrf_token_is_bound_to_the_session():
    secret = auth.new_secret()
    token = auth.csrf_token(secret, "session-a")
    assert not auth.verify_csrf(secret, "session-b", token)


def test_csrf_token_rejects_tampering():
    secret = auth.new_secret()
    token = auth.csrf_token(secret, "s")
    assert not auth.verify_csrf(secret, "s", token[:-1] + "x")


def test_csrf_rejects_an_empty_token():
    secret = auth.new_secret()
    assert not auth.verify_csrf(secret, "s", "")


# --------------------------------------------------- sources / approval queue


def _source(**kw):
    base = {"key": "youtube:UCbrother", "label": "C Flux",
            "channel_id": "UCPcTWaLV8zwx4WP4QExHj4Q", "enabled": 1, "imported": 1,
            "last_sync_at": "2026-08-12T16:00:00+00:00", "last_sync_ok": 1,
            "consecutive_failures": 0, "last_error": None}
    base.update(kw)
    return base


def _pending(n=3):
    return [{"id": i, "source": "youtube:UCbrother", "title": f"Channel {i}",
             "channel_id": f"UC{i:022d}", "seen_at": "2026-08-12T16:00:00+00:00"}
            for i in range(1, n + 1)]


def test_pending_page_renders_the_queue():
    html = templates.pending_page(pending=_pending(3), sources=[_source()],
                                 csrf="tok").decode()
    assert "3 channel(s) waiting" in html
    assert "Channel 1" in html and "Channel 3" in html
    assert "Approve selected" in html


def test_every_queue_row_shares_the_id_field_name():
    """The approve handler reads a repeated `id` field; if the template numbered
    them uniquely the multi-select would silently approve nothing."""
    html = templates.pending_page(pending=_pending(3), sources=[_source()],
                                 csrf="tok").decode()
    assert html.count('name="id"') == 3


def test_pending_page_with_an_empty_queue():
    html = templates.pending_page(pending=[], sources=[_source()], csrf="tok").decode()
    assert "Nothing awaiting approval" in html


def test_pending_page_with_no_sources():
    html = templates.pending_page(pending=[], sources=[], csrf="tok").decode()
    assert "No sources yet" in html


def test_never_synced_source_is_labelled():
    html = templates.pending_page(pending=[], sources=[_source(last_sync_ok=None)],
                                 csrf="tok").decode()
    assert "never synced" in html


def test_failing_source_shows_the_actionable_error_in_full():
    """The useful errors say exactly what to do; truncating them defeats the point."""
    message = ('subscriptions are private (subscriptionForbidden) — nothing '
               'changed. Fix: YouTube → Settings → Privacy → uncheck "Keep all '
               'my subscriptions private".')
    html = templates.pending_page(
        pending=[], sources=[_source(last_sync_ok=0, consecutive_failures=3,
                                     last_error=message)],
        csrf="tok").decode()
    assert "failing (3)" in html
    assert "Keep all" in html


def test_hostile_pending_title_is_escaped():
    items = _pending(1)
    items[0]["title"] = '<img src=x onerror=alert(1)>'
    html = templates.pending_page(pending=items, sources=[_source()],
                                 csrf="tok").decode()
    assert "<img src=x" not in html
    assert "&lt;img" in html