test_hls_e2e.py

#!/usr/bin/env python3
"""
E2E tests for HLS video loading and API endpoints.
Tests against the running FastAPI server on port 8090.

Usage:
  python test_hls_e2e.py
"""

import os
import sys
import time
import json
import requests

BASE = os.environ.get("BASE_URL", "http://localhost:8090")
PASS = 0
FAIL = 0


def _run(name, fn):
    global PASS, FAIL
    try:
        fn()
        PASS += 1
        print(f"  ✓ {name}")
    except Exception as e:
        FAIL += 1
        print(f"  ✗ {name}: {e}")


def test_hls_stats():
    """HLS stats endpoint returns valid data."""
    r = requests.get(f"{BASE}/api/hls/stats", timeout=5)
    assert r.status_code == 200, f"Status {r.status_code}"
    d = r.json()
    assert "total_videos" in d
    assert "hls_ready" in d
    assert "hls_pending" in d
    assert d["hls_ready"] > 0, "No HLS videos ready"


def test_media_info_hls_flag():
    """media_info includes hls_available flag."""
    # Find a video with HLS
    stats = requests.get(f"{BASE}/api/hls/stats", timeout=5).json()
    if stats["hls_ready"] == 0:
        raise Exception("No HLS videos to test with")

    # Get a creator with videos
    creators = requests.get(f"{BASE}/api/creators", timeout=5).json()
    for creator in creators:
        if creator["video_count"] > 0:
            posts = requests.get(
                f"{BASE}/api/creators/{creator['id']}/posts?media_type=video&per_page=5",
                timeout=5
            ).json()
            for post in posts.get("posts", []):
                for media in post.get("media", []):
                    if media["file_type"] == "video":
                        info = requests.get(f"{BASE}/api/media/{media['id']}/info", timeout=5).json()
                        assert "hls_available" in info, "Missing hls_available field"
                        return
    raise Exception("No video found to test")


def test_hls_playlist_content_type():
    """HLS playlist served with correct content type."""
    media_id = _find_hls_media_id()
    r = requests.get(f"{BASE}/api/media/{media_id}/hls/stream.m3u8", timeout=5)
    assert r.status_code == 200, f"Status {r.status_code}"
    assert "mpegurl" in r.headers.get("content-type", "").lower(), \
        f"Wrong content-type: {r.headers.get('content-type')}"
    assert "#EXTM3U" in r.text, "Invalid m3u8 content"
    assert "#EXTINF:" in r.text, "No segments in playlist"


def test_hls_segment_content_type():
    """HLS segment served with correct content type and cache headers."""
    media_id = _find_hls_media_id()
    r = requests.get(f"{BASE}/api/media/{media_id}/hls/stream.m3u8", timeout=5)
    # Find first segment
    segments = [l.strip() for l in r.text.splitlines() if l.strip().endswith('.ts')]
    assert len(segments) > 0, "No segments in playlist"

    seg_r = requests.get(f"{BASE}/api/media/{media_id}/hls/{segments[0]}", timeout=5)
    assert seg_r.status_code == 200, f"Segment status {seg_r.status_code}"
    assert "mp2t" in seg_r.headers.get("content-type", "").lower(), \
        f"Wrong segment content-type: {seg_r.headers.get('content-type')}"
    assert "max-age=31536000" in seg_r.headers.get("cache-control", ""), \
        "Segment missing long cache header"


def test_hls_path_traversal_blocked():
    """Path traversal in HLS endpoint is blocked."""
    media_id = _find_hls_media_id()
    r = requests.get(f"{BASE}/api/media/{media_id}/hls/../../etc/passwd", timeout=5)
    assert r.status_code in (400, 404), f"Path traversal not blocked: {r.status_code}"


def test_non_hls_video_returns_404():
    """HLS endpoint returns 404 for videos without HLS."""
    # Find a video without HLS
    creators = requests.get(f"{BASE}/api/creators", timeout=5).json()
    for creator in creators:
        if creator["video_count"] > 0:
            posts = requests.get(
                f"{BASE}/api/creators/{creator['id']}/posts?media_type=video&per_page=50",
                timeout=5
            ).json()
            for post in posts.get("posts", []):
                for media in post.get("media", []):
                    if media["file_type"] == "video":
                        info = requests.get(f"{BASE}/api/media/{media['id']}/info", timeout=5).json()
                        if not info.get("hls_available"):
                            r = requests.get(f"{BASE}/api/media/{media['id']}/hls/stream.m3u8", timeout=5)
                            assert r.status_code == 404, f"Expected 404, got {r.status_code}"
                            return
    # All videos have HLS — that's fine, skip
    print("    (skipped: all videos have HLS)")


def test_hls_playlist_load_time():
    """HLS playlist loads within 500ms."""
    media_id = _find_hls_media_id()
    start = time.time()
    r = requests.get(f"{BASE}/api/media/{media_id}/hls/stream.m3u8", timeout=5)
    elapsed = time.time() - start
    assert r.status_code == 200
    assert elapsed < 0.5, f"Playlist load too slow: {elapsed:.3f}s (limit 500ms)"


def test_hls_segment_load_time():
    """First HLS segment loads within 2s."""
    media_id = _find_hls_media_id()
    r = requests.get(f"{BASE}/api/media/{media_id}/hls/stream.m3u8", timeout=5)
    segments = [l.strip() for l in r.text.splitlines() if l.strip().endswith('.ts')]
    assert len(segments) > 0

    start = time.time()
    seg_r = requests.get(f"{BASE}/api/media/{media_id}/hls/{segments[0]}", timeout=10)
    elapsed = time.time() - start
    assert seg_r.status_code == 200
    assert elapsed < 2.0, f"Segment load too slow: {elapsed:.3f}s (limit 2s)"


def test_media_info_load_time():
    """media_info API responds within 200ms."""
    media_id = _find_hls_media_id()
    start = time.time()
    r = requests.get(f"{BASE}/api/media/{media_id}/info", timeout=5)
    elapsed = time.time() - start
    assert r.status_code == 200
    assert elapsed < 0.2, f"media_info too slow: {elapsed:.3f}s (limit 200ms)"


def test_like_endpoint():
    """Like check endpoint works at GET /api/posts/{id}/like."""
    creators = requests.get(f"{BASE}/api/creators", timeout=5).json()
    for creator in creators:
        if creator["post_count"] > 0:
            posts = requests.get(
                f"{BASE}/api/creators/{creator['id']}/posts?per_page=1",
                timeout=5
            ).json()
            if posts.get("posts"):
                post_id = posts["posts"][0]["id"]
                r = requests.get(f"{BASE}/api/posts/{post_id}/like", timeout=5)
                assert r.status_code == 200, f"Like check failed: {r.status_code}"
                d = r.json()
                assert "liked" in d, "Missing 'liked' field"
                return
    raise Exception("No posts to test like endpoint")


def test_mp4_fallback_still_works():
    """Raw mp4 stream endpoint still works for all videos."""
    media_id = _find_hls_media_id()
    r = requests.get(f"{BASE}/api/media/{media_id}/stream", timeout=10,
                     headers={"Range": "bytes=0-1023"})
    assert r.status_code == 206, f"Range request failed: {r.status_code}"
    assert len(r.content) > 0, "Empty response"


# --- Helpers ---

_hls_media_cache = None

def _find_hls_media_id():
    global _hls_media_cache
    if _hls_media_cache:
        return _hls_media_cache

    creators = requests.get(f"{BASE}/api/creators", timeout=5).json()
    for creator in creators:
        if creator["video_count"] > 0:
            posts = requests.get(
                f"{BASE}/api/creators/{creator['id']}/posts?media_type=video&per_page=10",
                timeout=5
            ).json()
            for post in posts.get("posts", []):
                for media in post.get("media", []):
                    if media["file_type"] == "video":
                        info = requests.get(f"{BASE}/api/media/{media['id']}/info", timeout=5).json()
                        if info.get("hls_available"):
                            _hls_media_cache = media["id"]
                            return media["id"]
    raise Exception("No HLS video found")


if __name__ == "__main__":
    print(f"\nHLS E2E Tests against {BASE}\n")

    # Check server is running
    try:
        requests.get(f"{BASE}/api/hls/stats", timeout=3)
    except Exception:
        print("ERROR: Server not running at", BASE)
        sys.exit(1)

    tests = [
        ("HLS stats endpoint", test_hls_stats),
        ("media_info includes hls_available", test_media_info_hls_flag),
        ("HLS playlist content type", test_hls_playlist_content_type),
        ("HLS segment content type + cache", test_hls_segment_content_type),
        ("HLS path traversal blocked", test_hls_path_traversal_blocked),
        ("Non-HLS video returns 404", test_non_hls_video_returns_404),
        ("HLS playlist loads < 500ms", test_hls_playlist_load_time),
        ("HLS first segment loads < 2s", test_hls_segment_load_time),
        ("media_info responds < 200ms", test_media_info_load_time),
        ("Like endpoint works", test_like_endpoint),
        ("MP4 fallback still works", test_mp4_fallback_still_works),
    ]

    for name, fn in tests:
        _run(name, fn)

    print(f"\n{'=' * 40}")
    print(f"Results: {PASS} passed, {FAIL} failed")
    print(f"{'=' * 40}")
    sys.exit(1 if FAIL > 0 else 0)