From 95bf0f9701ce57a44f09fdbe601a99bccb246969 Mon Sep 17 00:00:00 2001 From: Stefano Date: Wed, 28 May 2025 23:31:04 -0500 Subject: [PATCH] Created unit tests for check_streams --- .vscode/settings.json | 5 + app/utils/check_streams.py | 2 + tests/utils/test_check_streams.py | 309 ++++++++++++++++++++++++++++++ 3 files changed, 316 insertions(+) create mode 100644 tests/utils/test_check_streams.py diff --git a/.vscode/settings.json b/.vscode/settings.json index 9373c89..e254150 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -15,6 +15,7 @@ "botocore", "BURSTABLE", "cabletv", + "capsys", "CDUF", "cduflogo", "cdulogo", @@ -33,6 +34,7 @@ "CUNF", "cunflogo", "cuulogo", + "deadstreams", "delenv", "delogo", "devel", @@ -40,6 +42,7 @@ "dmlogo", "dotenv", "EXTINF", + "EXTM", "fastapi", "filterwarnings", "fiorinis", @@ -51,6 +54,8 @@ "KHTML", "lclogo", "LETSENCRYPT", + "levelname", + "mpegurl", "nohup", "ondelete", "onupdate", diff --git a/app/utils/check_streams.py b/app/utils/check_streams.py index 55b3ca4..0d6648a 100644 --- a/app/utils/check_streams.py +++ b/app/utils/check_streams.py @@ -66,6 +66,8 @@ class StreamValidator: "application/octet-stream", "application/x-mpegURL", ] + if content_type is None: + return False return any(ct in content_type for ct in valid_types) def parse_playlist(self, file_path): diff --git a/tests/utils/test_check_streams.py b/tests/utils/test_check_streams.py new file mode 100644 index 0000000..654e193 --- /dev/null +++ b/tests/utils/test_check_streams.py @@ -0,0 +1,309 @@ +import os +from unittest.mock import MagicMock, Mock, mock_open, patch + +import pytest +import requests +from requests.exceptions import ConnectionError, HTTPError, RequestException, Timeout + +from app.utils.check_streams import StreamValidator, main + + +@pytest.fixture +def validator(): + """Create a StreamValidator instance for testing""" + return StreamValidator(timeout=1) + + +def test_validator_init(): + """Test StreamValidator initialization with default and custom values""" + # Test with default user agent + validator = StreamValidator() + assert validator.timeout == 10 + assert "Mozilla" in validator.session.headers["User-Agent"] + + # Test with custom values + custom_agent = "CustomAgent/1.0" + validator = StreamValidator(timeout=5, user_agent=custom_agent) + assert validator.timeout == 5 + assert validator.session.headers["User-Agent"] == custom_agent + + +def test_is_valid_content_type(validator): + """Test content type validation""" + valid_types = [ + "video/mp4", + "video/mp2t", + "application/vnd.apple.mpegurl", + "application/dash+xml", + "video/webm", + "application/octet-stream", + "application/x-mpegURL", + "video/mp4; charset=utf-8", # Test with additional parameters + ] + + invalid_types = [ + "text/html", + "application/json", + "image/jpeg", + "", + ] + + for content_type in valid_types: + assert validator._is_valid_content_type(content_type) + + for content_type in invalid_types: + assert not validator._is_valid_content_type(content_type) + + # Test None case explicitly + assert not validator._is_valid_content_type(None) + + +@pytest.mark.parametrize( + "status_code,content_type,should_succeed", + [ + (200, "video/mp4", True), + (206, "video/mp4", True), # Partial content + (404, "video/mp4", False), + (500, "video/mp4", False), + (200, "text/html", False), + (200, "application/json", False), + ], +) +def test_validate_stream_response_handling(status_code, content_type, should_succeed): + """Test stream validation with different response scenarios""" + mock_response = MagicMock() + mock_response.status_code = status_code + mock_response.headers = {"Content-Type": content_type} + mock_response.iter_content.return_value = iter([b"some content"]) + + mock_session = MagicMock() + mock_session.get.return_value.__enter__.return_value = mock_response + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert valid == should_succeed + mock_session.get.assert_called_once() + + +def test_validate_stream_connection_error(): + """Test stream validation with connection error""" + mock_session = MagicMock() + mock_session.get.side_effect = ConnectionError("Connection failed") + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert not valid + assert "Connection Error" in message + + +def test_validate_stream_timeout(): + """Test stream validation with timeout""" + mock_session = MagicMock() + mock_session.get.side_effect = Timeout("Request timed out") + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert not valid + assert "timeout" in message.lower() + + +def test_validate_stream_http_error(): + """Test stream validation with HTTP error""" + mock_session = MagicMock() + mock_session.get.side_effect = HTTPError("HTTP Error occurred") + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert not valid + assert "HTTP Error" in message + + +def test_validate_stream_request_exception(): + """Test stream validation with general request exception""" + mock_session = MagicMock() + mock_session.get.side_effect = RequestException("Request failed") + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert not valid + assert "Request Exception" in message + + +def test_validate_stream_content_read_error(): + """Test stream validation when content reading fails""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"Content-Type": "video/mp4"} + mock_response.iter_content.side_effect = ConnectionError("Read failed") + + mock_session = MagicMock() + mock_session.get.return_value.__enter__.return_value = mock_response + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert not valid + assert "Connection failed during content read" in message + + +def test_validate_stream_general_exception(): + """Test validate_stream with an unexpected exception""" + mock_session = MagicMock() + mock_session.get.side_effect = Exception("Unexpected error") + + with patch("requests.Session", return_value=mock_session): + validator = StreamValidator() + valid, message = validator.validate_stream("http://example.com/stream") + + assert not valid + assert "Validation error" in message + + +def test_parse_playlist(validator, tmp_path): + """Test playlist file parsing""" + playlist_content = """ +#EXTM3U +#EXTINF:-1,Channel 1 +http://example.com/stream1 +#EXTINF:-1,Channel 2 +http://example.com/stream2 + +http://example.com/stream3 +""" + playlist_file = tmp_path / "test_playlist.m3u" + playlist_file.write_text(playlist_content) + + urls = validator.parse_playlist(str(playlist_file)) + assert len(urls) == 3 + assert urls == [ + "http://example.com/stream1", + "http://example.com/stream2", + "http://example.com/stream3", + ] + + +def test_parse_playlist_error(validator): + """Test playlist parsing with non-existent file""" + with pytest.raises(Exception): + validator.parse_playlist("nonexistent_file.m3u") + + +@patch("app.utils.check_streams.logging") +@patch("app.utils.check_streams.StreamValidator") +def test_main_with_urls(mock_validator_class, mock_logging, tmp_path, capsys): + """Test main function with direct URLs""" + # Setup mock validator + mock_validator = Mock() + mock_validator_class.return_value = mock_validator + mock_validator.validate_stream.return_value = (True, "Stream is valid") + + # Setup test arguments + test_args = ["script", "http://example.com/stream1", "http://example.com/stream2"] + with patch("sys.argv", test_args): + main() + + # Verify validator was called correctly + assert mock_validator.validate_stream.call_count == 2 + mock_validator.validate_stream.assert_any_call("http://example.com/stream1") + mock_validator.validate_stream.assert_any_call("http://example.com/stream2") + + +@patch("app.utils.check_streams.logging") +@patch("app.utils.check_streams.StreamValidator") +def test_main_with_playlist(mock_validator_class, mock_logging, tmp_path): + """Test main function with a playlist file""" + # Create test playlist + playlist_content = "http://example.com/stream1\nhttp://example.com/stream2" + playlist_file = tmp_path / "test.m3u" + playlist_file.write_text(playlist_content) + + # Setup mock validator + mock_validator = Mock() + mock_validator_class.return_value = mock_validator + mock_validator.parse_playlist.return_value = [ + "http://example.com/stream1", + "http://example.com/stream2", + ] + mock_validator.validate_stream.return_value = (True, "Stream is valid") + + # Setup test arguments + test_args = ["script", str(playlist_file)] + with patch("sys.argv", test_args): + main() + + # Verify validator was called correctly + mock_validator.parse_playlist.assert_called_once_with(str(playlist_file)) + assert mock_validator.validate_stream.call_count == 2 + + +@patch("app.utils.check_streams.logging") +@patch("app.utils.check_streams.StreamValidator") +def test_main_with_dead_streams(mock_validator_class, mock_logging, tmp_path): + """Test main function handling dead streams""" + # Setup mock validator + mock_validator = Mock() + mock_validator_class.return_value = mock_validator + mock_validator.validate_stream.return_value = (False, "Stream is dead") + + # Setup test arguments + test_args = ["script", "http://example.com/dead1", "http://example.com/dead2"] + + # Mock file operations + mock_file = mock_open() + with patch("sys.argv", test_args), patch("builtins.open", mock_file): + main() + + # Verify dead streams were written to file + mock_file().write.assert_called_once_with( + "http://example.com/dead1\nhttp://example.com/dead2" + ) + + +@patch("app.utils.check_streams.logging") +@patch("app.utils.check_streams.StreamValidator") +@patch("os.path.isfile") +def test_main_with_playlist_error( + mock_isfile, mock_validator_class, mock_logging, tmp_path +): + """Test main function handling playlist parsing errors""" + # Setup mock validator + mock_validator = Mock() + mock_validator_class.return_value = mock_validator + + # Configure mock validator behavior + error_msg = "Failed to parse playlist" + mock_validator.parse_playlist.side_effect = [ + Exception(error_msg), # First call fails + ["http://example.com/stream1"], # Second call succeeds + ] + mock_validator.validate_stream.return_value = (True, "Stream is valid") + + # Configure isfile mock to return True for our test files + mock_isfile.side_effect = lambda x: x in ["/invalid.m3u", "/valid.m3u"] + + # Setup test arguments + test_args = ["script", "/invalid.m3u", "/valid.m3u"] + with patch("sys.argv", test_args): + main() + + # Verify error was logged correctly + mock_logging.error.assert_called_with( + "Failed to process file /invalid.m3u: Failed to parse playlist" + ) + + # Verify processing continued with valid playlist + mock_validator.parse_playlist.assert_called_with("/valid.m3u") + assert ( + mock_validator.validate_stream.call_count == 1 + ) # Called for the URL from valid playlist