246 lines
9.3 KiB
Python
246 lines
9.3 KiB
Python
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.main import app
|
|
|
|
|
|
def _mock_discovery_live(video_id: str = "dQw4w9WgXcQ", thumb: str = "http://thumb.png"):
|
|
return AsyncMock(return_value=(video_id, thumb))
|
|
|
|
|
|
def _mock_discovery_none():
|
|
return AsyncMock(return_value=(None, None))
|
|
|
|
|
|
def _mock_extractor_hls(video_id: str = "dQw4w9WgXcQ"):
|
|
return MagicMock(
|
|
return_value={
|
|
"videoId": video_id,
|
|
"url": "https://manifest.hls.tv/pl.m3u8",
|
|
"streamType": "hls",
|
|
"title": "Live Stream",
|
|
"channel": "Lofi Girl",
|
|
"duration": None,
|
|
"isLive": True,
|
|
}
|
|
)
|
|
|
|
|
|
def _mock_extractor_none():
|
|
return MagicMock(return_value=None)
|
|
|
|
|
|
def _mock_validator_true():
|
|
return MagicMock(return_value=True)
|
|
|
|
|
|
def _mock_validator_false():
|
|
return MagicMock(return_value=False)
|
|
|
|
|
|
class TestListChannels:
|
|
def test_returns_all_channels(self) -> None:
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 99
|
|
assert all("id" in c for c in data)
|
|
assert all("name" in c for c in data)
|
|
assert all("isLive" in c for c in data)
|
|
assert all(c["isLive"] is True for c in data)
|
|
assert all(c["videoId"] is None for c in data)
|
|
assert all("thumbnail" in c for c in data)
|
|
assert any(c["thumbnail"] is not None for c in data)
|
|
|
|
def test_marks_live_channels(self) -> None:
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(c["isLive"] for c in data)
|
|
|
|
|
|
class TestCheckChannelLive:
|
|
def test_returns_live_status(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_live()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels/UCSJ4gkVC6NrvII8umztf0Ow/live")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["channelId"] == "UCSJ4gkVC6NrvII8umztf0Ow"
|
|
assert data["isLive"] is True
|
|
assert data["videoId"] == "dQw4w9WgXcQ"
|
|
assert data["thumbnail"] == "http://thumb.png"
|
|
|
|
def test_returns_not_live(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_none()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels/UCSJ4gkVC6NrvII8umztf0Ow/live")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["isLive"] is False
|
|
assert data["videoId"] is None
|
|
|
|
def test_returns_404_for_unknown_channel(self) -> None:
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels/UC_NOTEXIST/live")
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestGetStream:
|
|
def test_returns_proxied_stream_url(self) -> None:
|
|
with patch("src.main.extract_audio_stream", _mock_extractor_hls()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/stream/dQw4w9WgXcQ")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["url"].startswith("/api/proxy/hls?video=")
|
|
assert data["streamType"] == "hls"
|
|
|
|
def test_returns_503_when_no_stream(self) -> None:
|
|
with patch("src.main.extract_audio_stream", _mock_extractor_none()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/stream/dQw4w9WgXcQ")
|
|
|
|
assert response.status_code == 503
|
|
|
|
|
|
class TestNowPlaying:
|
|
def test_returns_active_stream(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_live()):
|
|
with patch("src.main.extract_audio_stream", _mock_extractor_hls()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/now-playing")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["channel"] is not None
|
|
assert data["videoId"] == "dQw4w9WgXcQ"
|
|
|
|
def test_returns_none_when_no_live_channels(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_none()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/now-playing")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["channel"] is None
|
|
assert data["videoId"] is None
|
|
assert data["url"] is None
|
|
|
|
def test_now_playing_returns_proxied_url(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_live()):
|
|
with patch("src.main.extract_audio_stream", _mock_extractor_hls()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/now-playing")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["url"].startswith("/api/proxy/hls?video=")
|
|
|
|
|
|
class TestProxyStream:
|
|
def test_proxy_hls_returns_playlist(self) -> None:
|
|
"""Verify HLS proxy returns m3u8 playlist with rewritten segments."""
|
|
mock_playlist = "#EXTM3U\n#EXTINF:5.0\nseg0.ts\n#EXTINF:5.0\nseg1.ts"
|
|
with patch("src.main._get_cached_stream", return_value=("http://example.com/playlist.m3u8", "hls")):
|
|
with patch("src.main._fetch_playlist", return_value=mock_playlist):
|
|
client = TestClient(app)
|
|
response = client.get("/api/proxy/hls?video=test123")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/x-mpegURL"
|
|
body = response.text
|
|
assert "#EXTM3U" in body
|
|
assert "/api/proxy/segment?video=test123&idx=0" in body
|
|
assert "/api/proxy/segment?video=test123&idx=1" in body
|
|
|
|
def test_proxy_segment_returns_audio_data(self) -> None:
|
|
"""Verify segment proxy returns actual audio content."""
|
|
mock_segment = b"\x00\x01\x02\x03" * 1000
|
|
mock_playlist = "#EXTM3U\n#EXTINF:5.0\nhttp://example.com/seg0.ts"
|
|
with patch("src.main._get_cached_stream", return_value=("http://example.com/pl.m3u8", "hls")):
|
|
with patch("src.main._fetch_playlist", return_value=mock_playlist):
|
|
with patch("httpx.get", return_value=MagicMock(status_code=200, content=mock_segment)):
|
|
client = TestClient(app)
|
|
response = client.get("/api/proxy/segment?video=test123&idx=0")
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == mock_segment
|
|
assert len(response.content) > 0
|
|
|
|
def test_proxy_hls_503_when_no_stream(self) -> None:
|
|
"""HLS proxy should return 503 when video stream can't be extracted."""
|
|
with patch("src.main._get_cached_stream", side_effect=HTTPException(status_code=503)):
|
|
client = TestClient(app)
|
|
response = client.get("/api/proxy/hls?video=test123")
|
|
|
|
assert response.status_code == 503
|
|
|
|
def test_proxy_hls_direct_stream_returns_m3u8(self) -> None:
|
|
"""HLS proxy returns synthetic m3u8 for direct streams."""
|
|
with patch("src.main._get_cached_stream", return_value=("http://example.com/audio.mp4", "direct")):
|
|
client = TestClient(app)
|
|
response = client.get("/api/proxy/hls?video=test123")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/x-mpegURL"
|
|
body = response.text
|
|
assert "#EXTM3U" in body
|
|
assert "/api/proxy/segment?video=test123&idx=0" in body
|
|
|
|
def test_proxy_audio_direct_stream(self) -> None:
|
|
"""Audio proxy streams direct audio content."""
|
|
mock_chunk = b"\x00\x01\x02\x03" * 1000
|
|
with patch("src.main._get_cached_stream", return_value=("http://example.com/audio.mp4", "direct")):
|
|
with patch("urllib.request.urlopen") as mock_urlopen:
|
|
mock_resp = MagicMock()
|
|
mock_resp.read.side_effect = [mock_chunk, b""]
|
|
mock_urlopen.return_value = mock_resp
|
|
client = TestClient(app)
|
|
response = client.get("/api/proxy/audio?video=test123")
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == mock_chunk
|
|
|
|
def test_proxy_audio_rejects_hls(self) -> None:
|
|
"""Audio proxy should reject HLS streams."""
|
|
with patch("src.main._get_cached_stream", return_value=("http://example.com/playlist.m3u8", "hls")):
|
|
client = TestClient(app)
|
|
response = client.get("/api/proxy/audio?video=test123")
|
|
|
|
assert response.status_code == 503
|
|
|
|
|
|
class TestCORS:
|
|
def test_allows_frontend_origin(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_none()):
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/channels",
|
|
headers={"origin": "http://localhost:5173"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "access-control-allow-origin" in response.headers
|
|
|
|
def test_allows_docker_frontend_origin(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_none()):
|
|
client = TestClient(app)
|
|
response = client.get(
|
|
"/api/channels",
|
|
headers={"origin": "http://frontend:80"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "access-control-allow-origin" in response.headers
|