169 lines
5.5 KiB
Python
169 lines
5.5 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.main import app
|
|
|
|
|
|
def _mock_discovery_live(video_id: str = "dQw4w9WgXcQ"):
|
|
return MagicMock(return_value=video_id)
|
|
|
|
|
|
def _mock_discovery_none():
|
|
return MagicMock(return_value=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:
|
|
with patch("src.main.find_live_video", _mock_discovery_none()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 20
|
|
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 False for c in data)
|
|
assert all(c["videoId"] is None for c in data)
|
|
|
|
def test_marks_live_channels(self) -> None:
|
|
with patch("src.main.find_live_video", _mock_discovery_live()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/channels")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(c["isLive"] for c in data)
|
|
assert all(c["videoId"] == "dQw4w9WgXcQ" 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/UCSJ4g0vg1503/live")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["channelId"] == "UCSJ4g0vg1503"
|
|
assert data["isLive"] is True
|
|
assert data["videoId"] == "dQw4w9WgXcQ"
|
|
|
|
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/UCSJ4g0vg1503/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_stream_url(self) -> None:
|
|
with patch("src.main.extract_audio_stream", _mock_extractor_hls()):
|
|
with patch("src.main.validate_stream", _mock_validator_true()):
|
|
client = TestClient(app)
|
|
response = client.get("/api/stream/dQw4w9WgXcQ")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["url"] == "https://manifest.hls.tv/pl.m3u8"
|
|
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
|
|
|
|
def test_returns_503_when_validation_fails(self) -> None:
|
|
with patch("src.main.extract_audio_stream", _mock_extractor_hls()):
|
|
with patch("src.main.validate_stream", _mock_validator_false()):
|
|
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
|
|
|
|
|
|
class TestCORS:
|
|
def test_allows_frontend_origin(self) -> 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:
|
|
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
|