117 lines
4.0 KiB
Python
117 lines
4.0 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
import httpx
|
|
|
|
from src.modules.validator import validate_stream
|
|
|
|
|
|
class TestValidateStream:
|
|
def test_validates_hls_stream(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.text = "#EXTM3U\n#EXT-X-VERSION:3\n/audio/stream.m3u8"
|
|
|
|
with patch("src.modules.validator.httpx.get", return_value=mock_response):
|
|
result = validate_stream("https://stream.m3u8", "hls")
|
|
|
|
assert result is True
|
|
|
|
def test_validates_hls_with_playlist_content(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.text = "#EXTM3U\n#EXTINF:10\nchunk0.ts\n#EXTINF:10\nchunk1.ts"
|
|
|
|
with patch("src.modules.validator.httpx.get", return_value=mock_response):
|
|
result = validate_stream("https://stream.m3u8", "hls")
|
|
|
|
assert result is True
|
|
|
|
def test_rejects_hls_wrong_status(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 403
|
|
|
|
with patch("src.modules.validator.httpx.get", return_value=mock_response):
|
|
result = validate_stream("https://stream.m3u8", "hls")
|
|
|
|
assert result is False
|
|
|
|
def test_rejects_hls_no_m3u8_content(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.text = "This is not an m3u8 playlist"
|
|
|
|
with patch("src.modules.validator.httpx.get", return_value=mock_response):
|
|
result = validate_stream("https://stream.m3u8", "hls")
|
|
|
|
assert result is False
|
|
|
|
def test_validates_direct_stream_200(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {"content-type": "audio/mp4"}
|
|
|
|
with patch("src.modules.validator.httpx.head", return_value=mock_response):
|
|
result = validate_stream("https://audio.mp4", "direct")
|
|
|
|
assert result is True
|
|
|
|
def test_validates_direct_stream_206(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 206
|
|
mock_response.headers = {"content-type": "audio/mpeg"}
|
|
|
|
with patch("src.modules.validator.httpx.head", return_value=mock_response):
|
|
result = validate_stream("https://audio.mp3", "direct")
|
|
|
|
assert result is True
|
|
|
|
def test_rejects_direct_wrong_status(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
|
|
with patch("src.modules.validator.httpx.head", return_value=mock_response):
|
|
result = validate_stream("https://audio.mp4", "direct")
|
|
|
|
assert result is False
|
|
|
|
def test_rejects_direct_no_content_type(self) -> None:
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.headers = {}
|
|
|
|
with patch("src.modules.validator.httpx.head", return_value=mock_response):
|
|
result = validate_stream("https://audio.mp4", "direct")
|
|
|
|
assert result is False
|
|
|
|
def test_rejects_unknown_stream_type(self) -> None:
|
|
result = validate_stream("https://stream", "unknown")
|
|
assert result is False
|
|
|
|
def test_handles_timeout(self) -> None:
|
|
with patch(
|
|
"src.modules.validator.httpx.get",
|
|
side_effect=httpx.TimeoutException("Timeout"),
|
|
):
|
|
result = validate_stream("https://stream.m3u8", "hls")
|
|
|
|
assert result is False
|
|
|
|
def test_handles_request_error_hls(self) -> None:
|
|
with patch(
|
|
"src.modules.validator.httpx.get",
|
|
side_effect=httpx.RequestError("Connection refused"),
|
|
):
|
|
result = validate_stream("https://stream.m3u8", "hls")
|
|
|
|
assert result is False
|
|
|
|
def test_handles_request_error_direct(self) -> None:
|
|
with patch(
|
|
"src.modules.validator.httpx.head",
|
|
side_effect=httpx.RequestError("Connection refused"),
|
|
):
|
|
result = validate_stream("https://audio.mp4", "direct")
|
|
|
|
assert result is False
|