Created unit tests for check_streams

.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",

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):

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